diff --git a/frontends/api/src/generated/v1/api.ts b/frontends/api/src/generated/v1/api.ts index a71480a838..c24749b7d4 100644 --- a/frontends/api/src/generated/v1/api.ts +++ b/frontends/api/src/generated/v1/api.ts @@ -5004,6 +5004,19 @@ export interface LearningResourceTopic { */ channel_url: string | null } +/** + * Serializer for the generic ``/api/v1/webhooks/learning_resources/`` endpoint. Accepts a batch of pre-computed canonical LearningResource payloads pushed from the OL Data Platform (Dagster). Each resource must carry at minimum ``readable_id``, ``etl_source`` and ``resource_type`` so the handler can route it to the correct loader; all other keys are preserved and passed through to the loaders unchanged. + * @export + * @interface LearningResourceWebhookRequestRequest + */ +export interface LearningResourceWebhookRequestRequest { + /** + * + * @type {Array<{ [key: string]: any; }>} + * @memberof LearningResourceWebhookRequestRequest + */ + resources: Array<{ [key: string]: any }> +} /** * SearchResponseSerializer with OpenAPI annotations for Learning Resources search * @export @@ -33308,6 +33321,59 @@ export const WebhooksApiAxiosParamCreator = function ( options: localVarRequestOptions, } }, + /** + * Generic webhook handler for pre-computed LearningResource batches delivered by the OL Data Platform (Dagster). The request body is ``{\"resources\": [ ... ]}`` where each resource is a canonical LearningResource dict carrying at minimum ``readable_id``, ``etl_source`` and ``resource_type``. Resources are grouped by ``(etl_source, resource_type)`` and routed to the matching loader (``load_courses`` / ``load_programs`` / ``load_documents`` / ``load_videos`` / ``load_podcasts``). Each loader performs a full sync for that source and upserts the OpenSearch index, so a batch must contain the authoritative set of resources for the (etl_source, resource_type) it represents. Resource types without a loader are logged and skipped rather than failing the whole batch. + * @param {LearningResourceWebhookRequestRequest} LearningResourceWebhookRequestRequest + * @param {*} [options] Override http request option. + * @throws {RequiredError} + */ + webhooksLearningResourcesCreate: async ( + LearningResourceWebhookRequestRequest: LearningResourceWebhookRequestRequest, + options: RawAxiosRequestConfig = {}, + ): Promise => { + // verify required parameter 'LearningResourceWebhookRequestRequest' is not null or undefined + assertParamExists( + "webhooksLearningResourcesCreate", + "LearningResourceWebhookRequestRequest", + LearningResourceWebhookRequestRequest, + ) + const localVarPath = `/api/v1/webhooks/learning_resources/` + // use dummy base URL string because the URL constructor only accepts absolute URLs. + const localVarUrlObj = new URL(localVarPath, DUMMY_BASE_URL) + let baseOptions + if (configuration) { + baseOptions = configuration.baseOptions + } + + const localVarRequestOptions = { + method: "POST", + ...baseOptions, + ...options, + } + const localVarHeaderParameter = {} as any + const localVarQueryParameter = {} as any + + localVarHeaderParameter["Content-Type"] = "application/json" + + setSearchParams(localVarUrlObj, localVarQueryParameter) + let headersFromBaseOptions = + baseOptions && baseOptions.headers ? baseOptions.headers : {} + localVarRequestOptions.headers = { + ...localVarHeaderParameter, + ...headersFromBaseOptions, + ...options.headers, + } + localVarRequestOptions.data = serializeDataIfNeeded( + LearningResourceWebhookRequestRequest, + localVarRequestOptions, + configuration, + ) + + return { + url: toPathString(localVarUrlObj), + options: localVarRequestOptions, + } + }, /** * Webhook handler for OVS video upserts and deletes from the dagster pipeline * @param {OVSVideoWebhookRequestRequest} [OVSVideoWebhookRequestRequest] @@ -33442,6 +33508,39 @@ export const WebhooksApiFp = function (configuration?: Configuration) { configuration, )(axios, operationBasePath || basePath) }, + /** + * Generic webhook handler for pre-computed LearningResource batches delivered by the OL Data Platform (Dagster). The request body is ``{\"resources\": [ ... ]}`` where each resource is a canonical LearningResource dict carrying at minimum ``readable_id``, ``etl_source`` and ``resource_type``. Resources are grouped by ``(etl_source, resource_type)`` and routed to the matching loader (``load_courses`` / ``load_programs`` / ``load_documents`` / ``load_videos`` / ``load_podcasts``). Each loader performs a full sync for that source and upserts the OpenSearch index, so a batch must contain the authoritative set of resources for the (etl_source, resource_type) it represents. Resource types without a loader are logged and skipped rather than failing the whole batch. + * @param {LearningResourceWebhookRequestRequest} LearningResourceWebhookRequestRequest + * @param {*} [options] Override http request option. + * @throws {RequiredError} + */ + async webhooksLearningResourcesCreate( + LearningResourceWebhookRequestRequest: LearningResourceWebhookRequestRequest, + options?: RawAxiosRequestConfig, + ): Promise< + ( + axios?: AxiosInstance, + basePath?: string, + ) => AxiosPromise + > { + const localVarAxiosArgs = + await localVarAxiosParamCreator.webhooksLearningResourcesCreate( + LearningResourceWebhookRequestRequest, + options, + ) + const index = configuration?.serverIndex ?? 0 + const operationBasePath = + operationServerMap["WebhooksApi.webhooksLearningResourcesCreate"]?.[ + index + ]?.url + return (axios, basePath) => + createRequestFunction( + localVarAxiosArgs, + globalAxios, + BASE_PATH, + configuration, + )(axios, operationBasePath || basePath) + }, /** * Webhook handler for OVS video upserts and deletes from the dagster pipeline * @param {OVSVideoWebhookRequestRequest} [OVSVideoWebhookRequestRequest] @@ -33525,6 +33624,23 @@ export const WebhooksApiFactory = function ( ) .then((request) => request(axios, basePath)) }, + /** + * Generic webhook handler for pre-computed LearningResource batches delivered by the OL Data Platform (Dagster). The request body is ``{\"resources\": [ ... ]}`` where each resource is a canonical LearningResource dict carrying at minimum ``readable_id``, ``etl_source`` and ``resource_type``. Resources are grouped by ``(etl_source, resource_type)`` and routed to the matching loader (``load_courses`` / ``load_programs`` / ``load_documents`` / ``load_videos`` / ``load_podcasts``). Each loader performs a full sync for that source and upserts the OpenSearch index, so a batch must contain the authoritative set of resources for the (etl_source, resource_type) it represents. Resource types without a loader are logged and skipped rather than failing the whole batch. + * @param {WebhooksApiWebhooksLearningResourcesCreateRequest} requestParameters Request parameters. + * @param {*} [options] Override http request option. + * @throws {RequiredError} + */ + webhooksLearningResourcesCreate( + requestParameters: WebhooksApiWebhooksLearningResourcesCreateRequest, + options?: RawAxiosRequestConfig, + ): AxiosPromise { + return localVarFp + .webhooksLearningResourcesCreate( + requestParameters.LearningResourceWebhookRequestRequest, + options, + ) + .then((request) => request(axios, basePath)) + }, /** * Webhook handler for OVS video upserts and deletes from the dagster pipeline * @param {WebhooksApiWebhooksOvsVideosCreateRequest} requestParameters Request parameters. @@ -33601,6 +33717,20 @@ export interface WebhooksApiWebhooksContentFilesDeleteCreateRequest { readonly ContentFileWebHookRequestRequest: ContentFileWebHookRequestRequest } +/** + * Request parameters for webhooksLearningResourcesCreate operation in WebhooksApi. + * @export + * @interface WebhooksApiWebhooksLearningResourcesCreateRequest + */ +export interface WebhooksApiWebhooksLearningResourcesCreateRequest { + /** + * + * @type {LearningResourceWebhookRequestRequest} + * @memberof WebhooksApiWebhooksLearningResourcesCreate + */ + readonly LearningResourceWebhookRequestRequest: LearningResourceWebhookRequestRequest +} + /** * Request parameters for webhooksOvsVideosCreate operation in WebhooksApi. * @export @@ -33664,6 +33794,25 @@ export class WebhooksApi extends BaseAPI { .then((request) => request(this.axios, this.basePath)) } + /** + * Generic webhook handler for pre-computed LearningResource batches delivered by the OL Data Platform (Dagster). The request body is ``{\"resources\": [ ... ]}`` where each resource is a canonical LearningResource dict carrying at minimum ``readable_id``, ``etl_source`` and ``resource_type``. Resources are grouped by ``(etl_source, resource_type)`` and routed to the matching loader (``load_courses`` / ``load_programs`` / ``load_documents`` / ``load_videos`` / ``load_podcasts``). Each loader performs a full sync for that source and upserts the OpenSearch index, so a batch must contain the authoritative set of resources for the (etl_source, resource_type) it represents. Resource types without a loader are logged and skipped rather than failing the whole batch. + * @param {WebhooksApiWebhooksLearningResourcesCreateRequest} requestParameters Request parameters. + * @param {*} [options] Override http request option. + * @throws {RequiredError} + * @memberof WebhooksApi + */ + public webhooksLearningResourcesCreate( + requestParameters: WebhooksApiWebhooksLearningResourcesCreateRequest, + options?: RawAxiosRequestConfig, + ) { + return WebhooksApiFp(this.configuration) + .webhooksLearningResourcesCreate( + requestParameters.LearningResourceWebhookRequestRequest, + options, + ) + .then((request) => request(this.axios, this.basePath)) + } + /** * Webhook handler for OVS video upserts and deletes from the dagster pipeline * @param {WebhooksApiWebhooksOvsVideosCreateRequest} requestParameters Request parameters. diff --git a/openapi/specs/v1.yaml b/openapi/specs/v1.yaml index 778e50b76a..d53e912feb 100644 --- a/openapi/specs/v1.yaml +++ b/openapi/specs/v1.yaml @@ -9917,6 +9917,38 @@ paths: schema: $ref: '#/components/schemas/ContentFileWebHookRequest' description: '' + /api/v1/webhooks/learning_resources/: + post: + operationId: webhooks_learning_resources_create + description: |- + Generic webhook handler for pre-computed LearningResource batches delivered + by the OL Data Platform (Dagster). + + The request body is ``{"resources": [ ... ]}`` where each resource is a + canonical LearningResource dict carrying at minimum ``readable_id``, + ``etl_source`` and ``resource_type``. Resources are grouped by + ``(etl_source, resource_type)`` and routed to the matching loader + (``load_courses`` / ``load_programs`` / ``load_documents`` / ``load_videos`` + / ``load_podcasts``). + Each loader performs a full sync for that source and upserts the OpenSearch + index, so a batch must contain the authoritative set of resources for the + (etl_source, resource_type) it represents. Resource types without a loader + are logged and skipped rather than failing the whole batch. + tags: + - webhooks + requestBody: + content: + application/json: + schema: + $ref: '#/components/schemas/LearningResourceWebhookRequestRequest' + required: true + responses: + '200': + content: + application/json: + schema: + $ref: '#/components/schemas/WebhookResponse' + description: '' /api/v1/webhooks/ovs_videos/: post: operationId: webhooks_ovs_videos_create @@ -13507,6 +13539,24 @@ components: - channel_url - id - name + LearningResourceWebhookRequestRequest: + type: object + description: |- + Serializer for the generic ``/api/v1/webhooks/learning_resources/`` endpoint. + + Accepts a batch of pre-computed canonical LearningResource payloads pushed + from the OL Data Platform (Dagster). Each resource must carry at minimum + ``readable_id``, ``etl_source`` and ``resource_type`` so the handler can + route it to the correct loader; all other keys are preserved and passed + through to the loaders unchanged. + properties: + resources: + type: array + items: + type: object + additionalProperties: {} + required: + - resources LearningResourcesSearchResponse: type: object description: |- diff --git a/webhooks/learning_resources_webhook_test.py b/webhooks/learning_resources_webhook_test.py new file mode 100644 index 0000000000..b6f102e362 --- /dev/null +++ b/webhooks/learning_resources_webhook_test.py @@ -0,0 +1,224 @@ +import hashlib +import hmac +import json + +import pytest +from django.urls import reverse + +from learning_resources.constants import LearningResourceType +from learning_resources.etl.constants import ETLSource + +WEBHOOK_URL_NAME = "webhooks:v1:learning_resources_webhook" + + +def _post(client, settings, payload, *, signature=None): + """POST a JSON payload to the learning_resources webhook, signing the body.""" + body = json.dumps(payload) + if signature is None: + signature = hmac.new( + settings.WEBHOOK_SECRET.encode(), + body.encode("utf-8"), + hashlib.sha256, + ).hexdigest() + return client.post( + reverse(WEBHOOK_URL_NAME), + data=body, + content_type="application/json", + headers={"X-MITLearn-Signature": signature}, + ) + + +def _resource(readable_id, etl_source, resource_type, **extra): + return { + "readable_id": readable_id, + "etl_source": etl_source, + "resource_type": resource_type, + "title": f"Title {readable_id}", + **extra, + } + + +@pytest.mark.django_db +def test_courses_routed_to_load_courses(settings, client, mocker): + """Course resources are routed to load_courses with their etl_source.""" + mock_clear = mocker.patch("webhooks.views.clear_views_cache") + mock_load_courses = mocker.patch("webhooks.views.load_courses", return_value=[]) + mock_load_programs = mocker.patch("webhooks.views.load_programs", return_value=[]) + + payload = { + "resources": [ + _resource( + "course-1", ETLSource.mitpe.name, LearningResourceType.course.name + ), + _resource( + "course-2", ETLSource.mitpe.name, LearningResourceType.course.name + ), + ] + } + response = _post(client, settings, payload) + + assert response.status_code == 200 + assert response.json()["status"] == "success" + mock_load_courses.assert_called_once() + etl_source_arg, courses_arg = mock_load_courses.call_args.args + assert etl_source_arg == ETLSource.mitpe.name + assert [r["readable_id"] for r in courses_arg] == ["course-1", "course-2"] + mock_load_programs.assert_not_called() + mock_clear.assert_called_once() + + +@pytest.mark.django_db +def test_groups_by_source_and_type(settings, client, mocker): + """Resources are grouped by (etl_source, resource_type) before dispatch.""" + mocker.patch("webhooks.views.clear_views_cache") + mock_load_courses = mocker.patch("webhooks.views.load_courses", return_value=[]) + mock_load_programs = mocker.patch("webhooks.views.load_programs", return_value=[]) + + payload = { + "resources": [ + _resource( + "c-mitpe", ETLSource.mitpe.name, LearningResourceType.course.name + ), + _resource("c-oll", ETLSource.oll.name, LearningResourceType.course.name), + _resource( + "p-edx", ETLSource.mit_edx.name, LearningResourceType.program.name + ), + ] + } + response = _post(client, settings, payload) + + assert response.status_code == 200 + # one load_courses call per distinct etl_source that has course resources + assert mock_load_courses.call_count == 2 + called_sources = {call.args[0] for call in mock_load_courses.call_args_list} + assert called_sources == {ETLSource.mitpe.name, ETLSource.oll.name} + mock_load_programs.assert_called_once() + assert mock_load_programs.call_args.args[0] == ETLSource.mit_edx.name + + +@pytest.mark.django_db +def test_videos_and_podcasts_routed(settings, client, mocker): + """Video and podcast resources reach load_videos / load_podcasts.""" + mocker.patch("webhooks.views.clear_views_cache") + mock_load_videos = mocker.patch("webhooks.views.load_videos", return_value=[]) + mock_load_podcasts = mocker.patch("webhooks.views.load_podcasts", return_value=[]) + + payload = { + "resources": [ + _resource("v1", ETLSource.youtube.name, LearningResourceType.video.name), + _resource( + "pod1", ETLSource.podcast.name, LearningResourceType.podcast.name + ), + ] + } + response = _post(client, settings, payload) + + assert response.status_code == 200 + mock_load_videos.assert_called_once() + assert [r["readable_id"] for r in mock_load_videos.call_args.args[0]] == ["v1"] + mock_load_podcasts.assert_called_once() + assert [r["readable_id"] for r in mock_load_podcasts.call_args.args[0]] == ["pod1"] + + +@pytest.mark.django_db +def test_documents_routed_to_load_documents(settings, client, mocker): + """Document resources (e.g. MIT Climate articles) route to load_documents.""" + mocker.patch("webhooks.views.clear_views_cache") + mock_load_documents = mocker.patch("webhooks.views.load_documents", return_value=[]) + mock_load_courses = mocker.patch("webhooks.views.load_courses", return_value=[]) + + payload = { + "resources": [ + _resource( + "climate-1", + ETLSource.mit_climate.name, + LearningResourceType.document.name, + resource_category="Article", + ), + ] + } + response = _post(client, settings, payload) + + assert response.status_code == 200 + mock_load_documents.assert_called_once() + etl_source_arg, docs_arg = mock_load_documents.call_args.args + assert etl_source_arg == ETLSource.mit_climate.name + assert [r["readable_id"] for r in docs_arg] == ["climate-1"] + mock_load_courses.assert_not_called() + + +@pytest.mark.django_db +def test_unsupported_resource_type_skipped(settings, client, mocker): + """A resource_type with no loader (e.g. mit_climate 'article') is skipped, not 500.""" + mocker.patch("webhooks.views.clear_views_cache") + mock_load_courses = mocker.patch("webhooks.views.load_courses", return_value=[]) + mock_load_programs = mocker.patch("webhooks.views.load_programs", return_value=[]) + mock_load_documents = mocker.patch("webhooks.views.load_documents", return_value=[]) + mock_load_videos = mocker.patch("webhooks.views.load_videos", return_value=[]) + mock_load_podcasts = mocker.patch("webhooks.views.load_podcasts", return_value=[]) + + payload = { + "resources": [ + _resource("mystery-1", ETLSource.mit_climate.name, "some_unknown_type"), + ] + } + response = _post(client, settings, payload) + + assert response.status_code == 200 + assert response.json()["status"] == "success" + mock_load_courses.assert_not_called() + mock_load_programs.assert_not_called() + mock_load_documents.assert_not_called() + mock_load_videos.assert_not_called() + mock_load_podcasts.assert_not_called() + + +@pytest.mark.django_db +def test_missing_required_field_returns_400(settings, client, mocker): + """A resource missing readable_id/etl_source/resource_type is rejected.""" + mock_load_courses = mocker.patch("webhooks.views.load_courses", return_value=[]) + payload = { + "resources": [ + { + "etl_source": ETLSource.mitpe.name, + "resource_type": LearningResourceType.course.name, + }, + ] + } + response = _post(client, settings, payload) + + assert response.status_code == 400 + mock_load_courses.assert_not_called() + + +@pytest.mark.django_db +def test_invalid_signature_rejected(settings, client, mocker): + """A bad signature short-circuits before any loader runs.""" + mock_load_courses = mocker.patch("webhooks.views.load_courses", return_value=[]) + payload = { + "resources": [ + _resource("c1", ETLSource.mitpe.name, LearningResourceType.course.name), + ] + } + response = _post(client, settings, payload, signature="deadbeef") + + assert response.status_code != 200 + mock_load_courses.assert_not_called() + + +@pytest.mark.django_db +def test_invalid_json_returns_400(settings, client): + """A validly-signed but malformed JSON body returns 400.""" + body = "{not json" + signature = hmac.new( + settings.WEBHOOK_SECRET.encode(), + body.encode("utf-8"), + hashlib.sha256, + ).hexdigest() + response = client.post( + reverse(WEBHOOK_URL_NAME), + data=body, + content_type="application/json", + headers={"X-MITLearn-Signature": signature}, + ) + assert response.status_code == 400 diff --git a/webhooks/serializers.py b/webhooks/serializers.py index 8a5e735531..eb47ee6b0e 100644 --- a/webhooks/serializers.py +++ b/webhooks/serializers.py @@ -59,3 +59,29 @@ class ContentFileWebhookRequestSerializer(serializers.Serializer): source = serializers.CharField(required=False, allow_blank=True) run = serializers.CharField(required=False, allow_blank=True) course = serializers.CharField(required=False, allow_blank=True) + + +class LearningResourceWebhookRequestSerializer(serializers.Serializer): + """ + Serializer for the generic ``/api/v1/webhooks/learning_resources/`` endpoint. + + Accepts a batch of pre-computed canonical LearningResource payloads pushed + from the OL Data Platform (Dagster). Each resource must carry at minimum + ``readable_id``, ``etl_source`` and ``resource_type`` so the handler can + route it to the correct loader; all other keys are preserved and passed + through to the loaders unchanged. + """ + + resources = serializers.ListField(child=serializers.DictField(), allow_empty=True) + + def validate_resources(self, resources): + required_fields = ("readable_id", "etl_source", "resource_type") + for index, resource in enumerate(resources): + missing = [field for field in required_fields if not resource.get(field)] + if missing: + msg = ( + f"resources[{index}] is missing required field(s): " + f"{', '.join(missing)}" + ) + raise serializers.ValidationError(msg) + return resources diff --git a/webhooks/urls.py b/webhooks/urls.py index 3c0fefef40..cacd8814f2 100644 --- a/webhooks/urls.py +++ b/webhooks/urls.py @@ -22,6 +22,11 @@ views.OVSVideoWebhookView.as_view(), name="ovs_video_webhook", ), + re_path( + r"^learning_resources/$", + views.LearningResourceWebhookView.as_view(), + name="learning_resources_webhook", + ), ] urlpatterns = [ diff --git a/webhooks/views.py b/webhooks/views.py index 1b0efcd791..8c9fde8851 100644 --- a/webhooks/views.py +++ b/webhooks/views.py @@ -1,5 +1,6 @@ import json import logging +from collections import defaultdict from django.core.exceptions import BadRequest from django.db.transaction import non_atomic_requests @@ -14,7 +15,14 @@ from learning_resources.constants import LearningResourceType from learning_resources.etl.constants import ETLSource -from learning_resources.etl.loaders import load_ovs_video_from_webhook +from learning_resources.etl.loaders import ( + load_courses, + load_documents, + load_ovs_video_from_webhook, + load_podcasts, + load_programs, + load_videos, +) from learning_resources.models import LearningResource from learning_resources.tasks import ingest_canvas_course, ingest_edx_run_archive from learning_resources.utils import ( @@ -24,6 +32,7 @@ from webhooks.decorators import require_signature from webhooks.serializers import ( ContentFileWebHookRequestSerializer, + LearningResourceWebhookRequestSerializer, OVSVideoWebhookRequestSerializer, WebhookResponseSerializer, ) @@ -157,6 +166,120 @@ def post(self, request): return HttpResponseBadRequest("Invalid JSON format") +@extend_schema_view( + post=extend_schema( + request=LearningResourceWebhookRequestSerializer, + responses=WebhookResponseSerializer(), + ), +) +class LearningResourceWebhookView(BaseWebhookView): + """ + Generic webhook handler for pre-computed LearningResource batches delivered + by the OL Data Platform (Dagster). + + The request body is ``{"resources": [ ... ]}`` where each resource is a + canonical LearningResource dict carrying at minimum ``readable_id``, + ``etl_source`` and ``resource_type``. Resources are grouped by + ``(etl_source, resource_type)`` and routed to the matching loader + (``load_courses`` / ``load_programs`` / ``load_documents`` / ``load_videos`` + / ``load_podcasts``). + Each loader performs a full sync for that source and upserts the OpenSearch + index, so a batch must contain the authoritative set of resources for the + (etl_source, resource_type) it represents. Resource types without a loader + are logged and skipped rather than failing the whole batch. + """ + + permission_classes = [] + authentication_classes = [] + parser_classes = [JSONParser] + serializer_class = LearningResourceWebhookRequestSerializer + + def success(self, extra_data=None): + if not extra_data: + extra_data = {} + response = WebhookResponseSerializer( + data={"status": "success", "message": "Webhook received", **extra_data} + ) + if response.is_valid(): + return Response(response.data) + log.error("Invalid response data: %s", response.errors) + return HttpResponseBadRequest("Invalid response data") + + def post(self, request): + try: + payload = json.loads(request.body) + except json.JSONDecodeError: + return HttpResponseBadRequest("Invalid JSON format") + serializer = LearningResourceWebhookRequestSerializer(data=payload) + serializer.is_valid(raise_exception=True) + summary = process_learning_resources_webhook( + serializer.validated_data["resources"] + ) + log.info("learning_resources webhook processed: %s", summary) + clear_views_cache() + return self.success() + + +def _load_resource_group(etl_source, resource_type, resources): + """ + Dispatch a group of same-typed resources to the matching loader. + + Returns the list of loaded LearningResource objects, or ``None`` if the + resource_type has no supported loader (the group is then skipped). + """ + if resource_type == LearningResourceType.course.name: + return load_courses(etl_source, resources) + if resource_type == LearningResourceType.program.name: + return load_programs(etl_source, resources) + if resource_type == LearningResourceType.document.name: + return load_documents(etl_source, resources) + if resource_type == LearningResourceType.video.name: + return load_videos(resources) + if resource_type == LearningResourceType.podcast.name: + return load_podcasts(resources) + return None + + +def process_learning_resources_webhook(resources): + """ + Group canonical LearningResource dicts by (etl_source, resource_type) and + route each group to the appropriate loader. Unsupported resource types are + logged and skipped rather than failing the whole batch. + """ + grouped = defaultdict(list) + for resource in resources: + grouped[(resource["etl_source"], resource["resource_type"])].append(resource) + + summary = {"loaded": 0, "skipped": 0, "groups": []} + for (etl_source, resource_type), items in grouped.items(): + loaded = _load_resource_group(etl_source, resource_type, items) + if loaded is None: + log.warning( + "No loader for resource_type=%s (etl_source=%s); skipping %d " + "resource(s)", + resource_type, + etl_source, + len(items), + ) + summary["skipped"] += len(items) + status = "skipped" + loaded_count = 0 + else: + loaded_count = len(loaded) + summary["loaded"] += loaded_count + status = "loaded" + summary["groups"].append( + { + "etl_source": etl_source, + "resource_type": resource_type, + "received": len(items), + "loaded": loaded_count, + "status": status, + } + ) + return summary + + def process_create_content_file_request(data): """ Process a content file CREATE webhook request based on the ETL source