import logging import requests from requests.auth import HTTPBasicAuth from rest_framework.views import APIView from rest_framework.response import Response from rest_framework.permissions import AllowAny from rest_framework.request import Request from rest_framework.response import Response from django.conf import settings from labs.labs_core import stateless_functions as stateless from labs.api.serializers.cert import ( TLMSCallbackSerializer, ) from labs.certificate.tlms import sync_tlms_certificates_for_user logger = logging.getLogger(__name__) TLMS_API_BASE_URL = "https://edu.pingcap.com/" def get_tlms_user_info(user_id: str): url = f"{TLMS_API_BASE_URL}/api/v1/users/id:{user_id}" response = requests.get(url, auth=HTTPBasicAuth(settings.TLMS_API_KEY, "")) return response.json() class TLMSCallbackEndpoint(APIView): permission_classes = (AllowAny,) def get(self, request: Request) -> Response: serializer = TLMSCallbackSerializer(data=request.query_params) serializer.is_valid(raise_exception=True) user = serializer.validated_data.get("user") course = serializer.validated_data.get("course") user_info = get_tlms_user_info(user) stateless.learner_exam_complete_handler(user_info, course) sync_tlms_certificates_for_user(user_info) return Response({"message": "Callback received"})