"""platform_core generic API views. All endpoints follow the pattern:: /api/v1/content/{app}/{model}/{id}// where ``app`` is the Django ``app_label``, ``model`` is ``model_name`` (lower), and ``id`` is the primary key of the entity. Security -------- - Only models that inherit from ``TenantScopedModel`` can be targeted so that cross-tenant access is structurally impossible. - Objects are always fetched with ``tenant=request.tenant`` so a user cannot read another tenant's data by guessing a PK. - Write operations require ``IsAuthenticated``; read operations also require authentication (no anonymous access to activity feeds or comments). URL structure ------------- Attachments: GET /api/v1/content/{app}/{model}/{id}/attachments/ POST /api/v1/content/{app}/{model}/{id}/attachments/ DELETE /api/v1/content/{app}/{model}/{id}/attachments/{attachment_id}/ Comments: GET /api/v1/content/{app}/{model}/{id}/comments/ POST /api/v1/content/{app}/{model}/{id}/comments/ PATCH /api/v1/content/{app}/{model}/{id}/comments/{comment_id}/ DELETE /api/v1/content/{app}/{model}/{id}/comments/{comment_id}/ Activity: GET /api/v1/content/{app}/{model}/{id}/activity/ Custom Fields: GET /api/v1/custom-fields/{entity_type}/definitions/ POST /api/v1/custom-fields/{entity_type}/definitions/ GET /api/v1/content/{app}/{model}/{id}/custom-fields/ PUT /api/v1/content/{app}/{model}/{id}/custom-fields/ """ from __future__ import annotations from django.apps import apps from django.contrib.contenttypes.models import ContentType from rest_framework import status from rest_framework.decorators import api_view, permission_classes from rest_framework.permissions import IsAuthenticated from rest_framework.request import Request from rest_framework.response import Response from simorgh.apps.platform_core.api.serializers import ( ActivitySerializer, AttachmentSerializer, AttachmentUploadSerializer, CommentCreateSerializer, CommentEditSerializer, CommentSerializer, CustomFieldDefinitionCreateSerializer, CustomFieldDefinitionSerializer, CustomFieldValuesSerializer, ExportJobCreateSerializer, ExportJobSerializer, FormDefinitionCreateSerializer, FormDefinitionSerializer, FormDefinitionUpdateSerializer, FormSubmissionSerializer, FormSubmitSerializer, ImportJobCreateSerializer, ImportJobErrorSerializer, ImportJobSerializer, MessageTemplateCreateSerializer, MessageTemplatePreviewSerializer, MessageTemplateSerializer, MessageTemplateUpdateSerializer, NoteCreateSerializer, NoteEditSerializer, NoteSerializer, PrintPreviewRequestSerializer, PrintTemplateCreateSerializer, PrintTemplateSerializer, PrintTemplateUpdateSerializer, TagAssignSerializer, TagCreateSerializer, TagSerializer, ) from simorgh.apps.platform_core.models import ( Attachment, Comment, CustomFieldDefinition, ExportJob, FormDefinition, ImportJob, MessageTemplate, Note, PrintTemplate, Tag, TagAssignment, ) from simorgh.apps.platform_core.selectors import ( get_custom_field_values, get_export_job, get_import_job, get_message_template, get_print_template, get_tags_for_entity, list_activity, list_attachments, list_comments, list_custom_field_definitions, list_export_jobs, list_form_definitions, list_form_submissions, list_import_jobs, list_message_templates, list_notes, list_print_templates, list_tags, ) from simorgh.apps.platform_core.services import ( PlatformCoreError, add_tags, attach_file, cancel_export_job, cancel_import_job, create_export_job, create_form_definition, create_import_job, create_message_template, create_note, create_print_template, create_tag, define_custom_field, delete_comment, delete_form_definition, delete_message_template, delete_note, delete_print_template, delete_tag, edit_comment, edit_note, post_comment, remove_attachment, remove_tags, render_html_preview, render_pdf, render_template, set_custom_field_value, submit_form, update_form_definition, update_message_template, update_print_template, ) from simorgh.core.models import TenantScopedModel # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _get_tenant_or_404(request: Request) -> object: tenant = getattr(request, "tenant", None) if tenant is None: return None return tenant def _resolve_entity( request: Request, app: str, model: str, pk: str, ) -> tuple[object | None, Response | None]: """Resolve an entity from (app, model, pk) within the current tenant. Returns ``(entity, None)`` on success or ``(None, error_response)`` on failure. Only models that subclass :class:`TenantScopedModel` are permitted; anything else returns 404 to avoid leaking internal schema info. """ tenant = _get_tenant_or_404(request) if tenant is None: return None, Response({"detail": "Tenant not found."}, status=status.HTTP_404_NOT_FOUND) # Guard against app/model names that don't exist; return 404 to avoid # leaking internal information about the schema. try: model_class = apps.get_model(app, model) except (LookupError, ValueError): return None, Response({"detail": "Not found."}, status=status.HTTP_404_NOT_FOUND) # Only tenant-scoped models may be accessed through this generic API. if not issubclass(model_class, TenantScopedModel): return None, Response({"detail": "Not found."}, status=status.HTTP_404_NOT_FOUND) try: entity = model_class.objects.get(pk=pk, tenant=tenant) except (model_class.DoesNotExist, ValueError): return None, Response({"detail": "Not found."}, status=status.HTTP_404_NOT_FOUND) return entity, None def _entity_type_label(app: str, model: str) -> str: return f"{app}.{model}" # --------------------------------------------------------------------------- # Attachments # --------------------------------------------------------------------------- @api_view(["GET", "POST"]) @permission_classes([IsAuthenticated]) def content_attachments(request: Request, app: str, model: str, pk: str) -> Response: """List or upload attachments for any tenant-scoped entity. GET — returns serialized list of attachments. POST — uploads a file (multipart/form-data) and attaches it to the entity. Requires a ``file`` field. Optional: ``description``, ``kind``. """ entity, err = _resolve_entity(request, app, model, pk) if err is not None: return err if request.method == "GET": qs = list_attachments(entity) serializer = AttachmentSerializer(qs, many=True) return Response(serializer.data) # POST — upload from simorgh.apps.storage.services import StorageError, store_file upload_ser = AttachmentUploadSerializer(data=request.data) if not upload_ser.is_valid(): return Response(upload_ser.errors, status=status.HTTP_400_BAD_REQUEST) uploaded_file = request.FILES.get("file") if not uploaded_file: return Response({"detail": "No file provided."}, status=status.HTTP_400_BAD_REQUEST) try: file_meta = store_file( filename=uploaded_file.name, content=uploaded_file, content_type=uploaded_file.content_type or "", tenant_id=entity.tenant_id, organization_node_id=getattr(entity, "organization_node_id", None), uploaded_by_id=request.user.pk, app_context="platform_core", ) except StorageError as exc: return Response({"detail": str(exc)}, status=status.HTTP_400_BAD_REQUEST) try: att = attach_file( entity, file_meta, kind=upload_ser.validated_data["kind"], description=upload_ser.validated_data["description"], uploaded_by=request.user, ) except PlatformCoreError as exc: return Response({"detail": str(exc)}, status=status.HTTP_400_BAD_REQUEST) return Response(AttachmentSerializer(att).data, status=status.HTTP_201_CREATED) @api_view(["DELETE"]) @permission_classes([IsAuthenticated]) def content_attachment_detail( request: Request, app: str, model: str, pk: str, attachment_id: str ) -> Response: """DELETE — remove an attachment from an entity.""" entity, err = _resolve_entity(request, app, model, pk) if err is not None: return err ct = ContentType.objects.get_for_model(entity.__class__) try: att = Attachment.objects.get(pk=attachment_id, content_type=ct, object_id=str(entity.pk)) except (Attachment.DoesNotExist, ValueError): return Response({"detail": "Not found."}, status=status.HTTP_404_NOT_FOUND) remove_attachment(att, actor=request.user) return Response(status=status.HTTP_204_NO_CONTENT) # --------------------------------------------------------------------------- # Comments # --------------------------------------------------------------------------- @api_view(["GET", "POST"]) @permission_classes([IsAuthenticated]) def content_comments(request: Request, app: str, model: str, pk: str) -> Response: """List or create comments on an entity. GET — returns top-level comments (with nested replies) ordered by ``created_at`` ascending. POST — creates a new comment. Body: ``{body, parent_id?}``. """ entity, err = _resolve_entity(request, app, model, pk) if err is not None: return err if request.method == "GET": qs = list_comments(entity, include_replies=True) return Response(CommentSerializer(list(qs), many=True).data) # POST — create ser = CommentCreateSerializer(data=request.data) if not ser.is_valid(): return Response(ser.errors, status=status.HTTP_400_BAD_REQUEST) parent = None parent_id = ser.validated_data.get("parent_id") if parent_id: try: ct = ContentType.objects.get_for_model(entity.__class__) parent = Comment.objects.get( pk=parent_id, content_type=ct, object_id=str(entity.pk), tenant_id=entity.tenant_id, ) except Comment.DoesNotExist: return Response( {"detail": "Parent comment not found."}, status=status.HTTP_404_NOT_FOUND ) try: comment = post_comment( entity, body=ser.validated_data["body"], author=request.user, parent=parent, ) except PlatformCoreError as exc: return Response({"detail": str(exc)}, status=status.HTTP_400_BAD_REQUEST) return Response(CommentSerializer(comment).data, status=status.HTTP_201_CREATED) @api_view(["PATCH", "DELETE"]) @permission_classes([IsAuthenticated]) def content_comment_detail( request: Request, app: str, model: str, pk: str, comment_id: str ) -> Response: """Edit or delete a single comment. PATCH — update the body. Body: ``{body}``. DELETE — soft-delete the comment (replies cascade). """ entity, err = _resolve_entity(request, app, model, pk) if err is not None: return err ct = ContentType.objects.get_for_model(entity.__class__) try: comment = Comment.objects.get( pk=comment_id, content_type=ct, object_id=str(entity.pk), tenant_id=entity.tenant_id, ) except (Comment.DoesNotExist, ValueError): return Response({"detail": "Not found."}, status=status.HTTP_404_NOT_FOUND) # Only the author or staff may edit/delete. if not request.user.is_staff and comment.author_id != request.user.pk: return Response({"detail": "Forbidden."}, status=status.HTTP_403_FORBIDDEN) if request.method == "PATCH": ser = CommentEditSerializer(data=request.data) if not ser.is_valid(): return Response(ser.errors, status=status.HTTP_400_BAD_REQUEST) try: comment = edit_comment(comment, body=ser.validated_data["body"], actor=request.user) except PlatformCoreError as exc: return Response({"detail": str(exc)}, status=status.HTTP_400_BAD_REQUEST) return Response(CommentSerializer(comment).data) # DELETE delete_comment(comment, actor=request.user) return Response(status=status.HTTP_204_NO_CONTENT) # --------------------------------------------------------------------------- # Activity feed # --------------------------------------------------------------------------- @api_view(["GET"]) @permission_classes([IsAuthenticated]) def content_activity(request: Request, app: str, model: str, pk: str) -> Response: """Return the activity feed for an entity. Optional query param: ``limit`` (default 50, max 200). """ entity, err = _resolve_entity(request, app, model, pk) if err is not None: return err limit = min(200, max(1, int(request.query_params.get("limit", 50)))) qs = list_activity(entity, limit=limit) return Response(ActivitySerializer(list(qs), many=True).data) # --------------------------------------------------------------------------- # Custom Fields — definitions # --------------------------------------------------------------------------- @api_view(["GET", "POST"]) @permission_classes([IsAuthenticated]) def custom_field_definitions(request: Request, entity_type: str) -> Response: """List or create custom field definitions for an entity type. ``entity_type`` is the URL path segment, e.g. ``crm.lead``. GET — list all definitions for the entity type in the current tenant. POST — create a new definition. Requires staff or tenant admin. """ tenant = _get_tenant_or_404(request) if tenant is None: return Response({"detail": "Tenant not found."}, status=status.HTTP_404_NOT_FOUND) # Normalise entity_type: URL uses dots, we store as "app.model". # Also accept dash-separated variants by normalising to lowercase dot form. entity_type = entity_type.lower().replace("-", ".") if request.method == "GET": qs = list_custom_field_definitions(tenant.pk, entity_type) return Response(CustomFieldDefinitionSerializer(list(qs), many=True).data) # POST — only staff/admin may define custom fields if not request.user.is_staff: return Response({"detail": "Forbidden."}, status=status.HTTP_403_FORBIDDEN) ser = CustomFieldDefinitionCreateSerializer(data=request.data) if not ser.is_valid(): return Response(ser.errors, status=status.HTTP_400_BAD_REQUEST) # Resolve an org_node for the definition (use the user's primary org node). from simorgh.apps.memberships.models import Membership org_node_id: int | None = None membership_qs = Membership.objects.filter( tenant=tenant, user=request.user ).order_by("id") membership = membership_qs.first() if membership: org_node_id = membership.organization_node_id if org_node_id is None: return Response( {"detail": "Could not determine organization node for user."}, status=status.HTTP_400_BAD_REQUEST, ) try: cfd = define_custom_field( tenant_id=tenant.pk, organization_node_id=org_node_id, entity_type=entity_type, **ser.validated_data, ) except PlatformCoreError as exc: return Response({"detail": str(exc)}, status=status.HTTP_400_BAD_REQUEST) return Response( CustomFieldDefinitionSerializer(cfd).data, status=status.HTTP_201_CREATED, ) # --------------------------------------------------------------------------- # Custom Fields — values on a specific entity # --------------------------------------------------------------------------- @api_view(["GET", "PUT"]) @permission_classes([IsAuthenticated]) def content_custom_fields(request: Request, app: str, model: str, pk: str) -> Response: """Read or update all custom field values for an entity. GET — returns ``{key: value}`` dict of all set custom field values. PUT — accepts a ``{key: value}`` dict and upserts each value. Keys that are not defined for this entity type are silently ignored. """ entity, err = _resolve_entity(request, app, model, pk) if err is not None: return err entity_type = _entity_type_label(app, model) if request.method == "GET": values = get_custom_field_values(entity) return Response(values) # PUT ser = CustomFieldValuesSerializer(data=request.data) if not ser.is_valid(): return Response(ser.errors, status=status.HTTP_400_BAD_REQUEST) incoming: dict = ser.validated_data # Fetch known definitions for this entity_type to validate keys. known_defs = { d.key: d for d in CustomFieldDefinition.objects.filter( tenant_id=entity.tenant_id, entity_type=entity_type, ) } errors: dict[str, str] = {} for key, value in incoming.items(): if key not in known_defs: errors[key] = "No custom field definition found for this key." if errors: return Response(errors, status=status.HTTP_400_BAD_REQUEST) for key, value in incoming.items(): try: set_custom_field_value(entity, definition=known_defs[key], value=value) except PlatformCoreError as exc: return Response({key: str(exc)}, status=status.HTTP_400_BAD_REQUEST) values = get_custom_field_values(entity) return Response(values) # --------------------------------------------------------------------------- # Tags — tenant-level CRUD # --------------------------------------------------------------------------- def _get_user_org_node_id(request: Request, tenant) -> int | None: from simorgh.apps.memberships.models import Membership m = Membership.objects.filter(tenant=tenant, users=request.user).order_by("id").first() return m.organization_node_id if m else None @api_view(["GET", "POST"]) @permission_classes([IsAuthenticated]) def tags_list(request: Request) -> Response: """List or create platform-level tags for the current tenant. GET — returns all tags (optional ``?entity_type=app.model`` filter). POST — creates a new tag. Requires staff. Body: ``{name, slug?, color?, entity_type?}``. """ tenant = _get_tenant_or_404(request) if tenant is None: return Response({"detail": "Tenant not found."}, status=status.HTTP_404_NOT_FOUND) if request.method == "GET": entity_type = request.query_params.get("entity_type", None) qs = list_tags(tenant.pk, entity_type=entity_type) return Response(TagSerializer(list(qs), many=True).data) # POST if not request.user.is_staff: return Response({"detail": "Forbidden."}, status=status.HTTP_403_FORBIDDEN) ser = TagCreateSerializer(data=request.data) if not ser.is_valid(): return Response(ser.errors, status=status.HTTP_400_BAD_REQUEST) org_node_id = _get_user_org_node_id(request, tenant) if org_node_id is None: return Response( {"detail": "Could not determine organization node for user."}, status=status.HTTP_400_BAD_REQUEST, ) try: tag = create_tag( tenant_id=tenant.pk, organization_node_id=org_node_id, name=ser.validated_data["name"], slug=ser.validated_data.get("slug", ""), color=ser.validated_data.get("color", "#6B7280"), entity_type=ser.validated_data.get("entity_type", ""), created_by=request.user, ) except PlatformCoreError as exc: return Response({"detail": str(exc)}, status=status.HTTP_400_BAD_REQUEST) return Response(TagSerializer(tag).data, status=status.HTTP_201_CREATED) @api_view(["DELETE"]) @permission_classes([IsAuthenticated]) def tag_detail(request: Request, tag_id: int) -> Response: """DELETE — remove a tag and all its assignments. Requires staff.""" tenant = _get_tenant_or_404(request) if tenant is None: return Response({"detail": "Tenant not found."}, status=status.HTTP_404_NOT_FOUND) if not request.user.is_staff: return Response({"detail": "Forbidden."}, status=status.HTTP_403_FORBIDDEN) try: tag = Tag.objects.get(pk=tag_id, tenant=tenant) except Tag.DoesNotExist: return Response({"detail": "Not found."}, status=status.HTTP_404_NOT_FOUND) delete_tag(tag, actor=request.user) return Response(status=status.HTTP_204_NO_CONTENT) # --------------------------------------------------------------------------- # Tags — assign / remove on entities # --------------------------------------------------------------------------- @api_view(["GET", "POST"]) @permission_classes([IsAuthenticated]) def content_tags(request: Request, app: str, model: str, pk: str) -> Response: """List or assign tags on an entity. GET — returns the list of tags currently assigned. POST — assigns tags. Body: ``{tag_ids: [...]}``. """ entity, err = _resolve_entity(request, app, model, pk) if err is not None: return err if request.method == "GET": qs = get_tags_for_entity(entity) return Response(TagSerializer(list(qs), many=True).data) # POST — assign ser = TagAssignSerializer(data=request.data) if not ser.is_valid(): return Response(ser.errors, status=status.HTTP_400_BAD_REQUEST) try: add_tags(entity, ser.validated_data["tag_ids"], actor=request.user) except PlatformCoreError as exc: return Response({"detail": str(exc)}, status=status.HTTP_400_BAD_REQUEST) qs = get_tags_for_entity(entity) return Response(TagSerializer(list(qs), many=True).data, status=status.HTTP_201_CREATED) @api_view(["DELETE"]) @permission_classes([IsAuthenticated]) def content_tag_detail( request: Request, app: str, model: str, pk: str, tag_id: int ) -> Response: """DELETE — remove a tag assignment from an entity.""" entity, err = _resolve_entity(request, app, model, pk) if err is not None: return err ct = ContentType.objects.get_for_model(entity.__class__) exists = TagAssignment.objects.filter( tag_id=tag_id, content_type=ct, object_id=str(entity.pk), tenant_id=entity.tenant_id, ).exists() if not exists: return Response({"detail": "Not found."}, status=status.HTTP_404_NOT_FOUND) from simorgh.apps.platform_core.services import remove_tags as _remove_tags _remove_tags(entity, [tag_id], actor=request.user) return Response(status=status.HTTP_204_NO_CONTENT) # --------------------------------------------------------------------------- # Notes # --------------------------------------------------------------------------- @api_view(["GET", "POST"]) @permission_classes([IsAuthenticated]) def content_notes(request: Request, app: str, model: str, pk: str) -> Response: """List or create notes on an entity. GET — returns notes visible to the requesting user, ordered by ``-is_pinned`` then ``-created_at``. POST — creates a new note. Body: ``{body, visibility?}``. """ entity, err = _resolve_entity(request, app, model, pk) if err is not None: return err if request.method == "GET": qs = list_notes(entity, viewer_id=request.user.pk) return Response(NoteSerializer(list(qs), many=True).data) # POST — create ser = NoteCreateSerializer(data=request.data) if not ser.is_valid(): return Response(ser.errors, status=status.HTTP_400_BAD_REQUEST) try: note = create_note( entity, body=ser.validated_data["body"], visibility=ser.validated_data["visibility"], author=request.user, ) except PlatformCoreError as exc: return Response({"detail": str(exc)}, status=status.HTTP_400_BAD_REQUEST) return Response(NoteSerializer(note).data, status=status.HTTP_201_CREATED) @api_view(["PATCH", "DELETE"]) @permission_classes([IsAuthenticated]) def content_note_detail( request: Request, app: str, model: str, pk: str, note_id: str ) -> Response: """Edit or delete a single note. PATCH — update body and/or is_pinned. Body: ``{body?, is_pinned?}``. DELETE — hard-delete the note. Only the author or staff may edit/delete. """ entity, err = _resolve_entity(request, app, model, pk) if err is not None: return err ct = ContentType.objects.get_for_model(entity.__class__) try: note = Note.objects.get( pk=note_id, content_type=ct, object_id=str(entity.pk), tenant_id=entity.tenant_id, ) except (Note.DoesNotExist, ValueError): return Response({"detail": "Not found."}, status=status.HTTP_404_NOT_FOUND) # Only the author or staff may mutate. if not request.user.is_staff and note.author_id != request.user.pk: return Response({"detail": "Forbidden."}, status=status.HTTP_403_FORBIDDEN) if request.method == "PATCH": ser = NoteEditSerializer(data=request.data) if not ser.is_valid(): return Response(ser.errors, status=status.HTTP_400_BAD_REQUEST) try: note = edit_note( note, body=ser.validated_data.get("body"), is_pinned=ser.validated_data.get("is_pinned"), actor=request.user, ) except PlatformCoreError as exc: return Response({"detail": str(exc)}, status=status.HTTP_400_BAD_REQUEST) return Response(NoteSerializer(note).data) # DELETE delete_note(note, actor=request.user) return Response(status=status.HTTP_204_NO_CONTENT) # --------------------------------------------------------------------------- # Dynamic Forms — definitions # --------------------------------------------------------------------------- @api_view(["GET", "POST"]) @permission_classes([IsAuthenticated]) def forms_list(request: Request) -> Response: """List or create form definitions for the current tenant. GET — returns all form definitions (filter with ``?is_active=true/false``). POST — creates a new form definition. Requires staff. """ tenant = _get_tenant_or_404(request) if tenant is None: return Response({"detail": "Tenant not found."}, status=status.HTTP_404_NOT_FOUND) if request.method == "GET": is_active_param = request.query_params.get("is_active") is_active: bool | None = None if is_active_param is not None: is_active = is_active_param.lower() in ("1", "true", "yes") qs = list_form_definitions(tenant.pk, is_active=is_active) return Response(FormDefinitionSerializer(list(qs), many=True).data) # POST if not request.user.is_staff: return Response({"detail": "Forbidden."}, status=status.HTTP_403_FORBIDDEN) ser = FormDefinitionCreateSerializer(data=request.data) if not ser.is_valid(): return Response(ser.errors, status=status.HTTP_400_BAD_REQUEST) org_node_id = _get_user_org_node_id(request, tenant) if org_node_id is None: return Response( {"detail": "Could not determine organization node for user."}, status=status.HTTP_400_BAD_REQUEST, ) try: form = create_form_definition( tenant_id=tenant.pk, organization_node_id=org_node_id, created_by=request.user, **ser.validated_data, ) except PlatformCoreError as exc: return Response({"detail": str(exc)}, status=status.HTTP_400_BAD_REQUEST) return Response(FormDefinitionSerializer(form).data, status=status.HTTP_201_CREATED) @api_view(["GET", "PATCH", "DELETE"]) @permission_classes([IsAuthenticated]) def form_detail(request: Request, form_id: str) -> Response: """Retrieve, update, or delete a single form definition. GET — returns full form definition (any authenticated user). PATCH — partial update (staff only). Body: any subset of FormDefinition fields. DELETE — delete the form and all its submissions (staff only). """ tenant = _get_tenant_or_404(request) if tenant is None: return Response({"detail": "Tenant not found."}, status=status.HTTP_404_NOT_FOUND) try: form = FormDefinition.objects.get(pk=form_id, tenant=tenant) except (FormDefinition.DoesNotExist, ValueError): return Response({"detail": "Not found."}, status=status.HTTP_404_NOT_FOUND) if request.method == "GET": return Response(FormDefinitionSerializer(form).data) if not request.user.is_staff: return Response({"detail": "Forbidden."}, status=status.HTTP_403_FORBIDDEN) if request.method == "PATCH": ser = FormDefinitionUpdateSerializer(data=request.data) if not ser.is_valid(): return Response(ser.errors, status=status.HTTP_400_BAD_REQUEST) try: form = update_form_definition(form, actor=request.user, **ser.validated_data) except PlatformCoreError as exc: return Response({"detail": str(exc)}, status=status.HTTP_400_BAD_REQUEST) return Response(FormDefinitionSerializer(form).data) # DELETE delete_form_definition(form, actor=request.user) return Response(status=status.HTTP_204_NO_CONTENT) # --------------------------------------------------------------------------- # Dynamic Forms — submit # --------------------------------------------------------------------------- @api_view(["POST"]) @permission_classes([IsAuthenticated]) def form_submit(request: Request, form_id: str) -> Response: """Submit data to a form definition. POST — validates data against the form's field schema and creates a :class:`FormSubmission`. Returns the created submission. """ tenant = _get_tenant_or_404(request) if tenant is None: return Response({"detail": "Tenant not found."}, status=status.HTTP_404_NOT_FOUND) try: form = FormDefinition.objects.get(pk=form_id, tenant=tenant) except (FormDefinition.DoesNotExist, ValueError): return Response({"detail": "Not found."}, status=status.HTTP_404_NOT_FOUND) ser = FormSubmitSerializer(data=request.data) if not ser.is_valid(): return Response(ser.errors, status=status.HTTP_400_BAD_REQUEST) ip = _get_client_ip(request) try: submission = submit_form( form, ser.validated_data["data"], user=request.user, ip=ip, ) except PlatformCoreError as exc: return Response({"detail": str(exc)}, status=status.HTTP_400_BAD_REQUEST) return Response(FormSubmissionSerializer(submission).data, status=status.HTTP_201_CREATED) # --------------------------------------------------------------------------- # Dynamic Forms — public submit (unauthenticated) # --------------------------------------------------------------------------- @api_view(["GET", "POST"]) def public_form(request: Request, slug: str) -> Response: """Public form endpoint — no authentication required. GET — returns the form definition (title, description, fields) for rendering in an external or embedded form widget. POST — submits the form anonymously. ``submitted_by`` is left NULL. """ # Resolve tenant from request (e.g. via Host header middleware). tenant = getattr(request, "tenant", None) if tenant is None: return Response({"detail": "Not found."}, status=status.HTTP_404_NOT_FOUND) try: from simorgh.apps.platform_core.services import get_form_definition_by_slug form = get_form_definition_by_slug(slug, tenant_id=tenant.pk) except PlatformCoreError: return Response({"detail": "Not found."}, status=status.HTTP_404_NOT_FOUND) if request.method == "GET": return Response({ "id": str(form.pk), "title": form.title, "description": form.description, "fields": form.fields, }) # POST — anonymous submission ser = FormSubmitSerializer(data=request.data) if not ser.is_valid(): return Response(ser.errors, status=status.HTTP_400_BAD_REQUEST) ip = _get_client_ip(request) try: submission = submit_form( form, ser.validated_data["data"], user=None, ip=ip, ) except PlatformCoreError as exc: return Response({"detail": str(exc)}, status=status.HTTP_400_BAD_REQUEST) return Response( {"id": str(submission.pk), "processed": submission.processed}, status=status.HTTP_201_CREATED, ) # --------------------------------------------------------------------------- # Dynamic Forms — submissions list # --------------------------------------------------------------------------- @api_view(["GET"]) @permission_classes([IsAuthenticated]) def form_submissions(request: Request, form_id: str) -> Response: """List submissions for a form definition. Requires staff.""" tenant = _get_tenant_or_404(request) if tenant is None: return Response({"detail": "Tenant not found."}, status=status.HTTP_404_NOT_FOUND) if not request.user.is_staff: return Response({"detail": "Forbidden."}, status=status.HTTP_403_FORBIDDEN) try: form = FormDefinition.objects.get(pk=form_id, tenant=tenant) except (FormDefinition.DoesNotExist, ValueError): return Response({"detail": "Not found."}, status=status.HTTP_404_NOT_FOUND) limit = min(500, max(1, int(request.query_params.get("limit", 100)))) qs = list_form_submissions(form, limit=limit) return Response(FormSubmissionSerializer(list(qs), many=True).data) # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _get_client_ip(request: Request) -> str | None: """Extract the real client IP from request headers.""" x_forwarded = request.META.get("HTTP_X_FORWARDED_FOR") if x_forwarded: return x_forwarded.split(",")[0].strip() return request.META.get("REMOTE_ADDR") # --------------------------------------------------------------------------- # Import Engine # --------------------------------------------------------------------------- @api_view(["POST"]) @permission_classes([IsAuthenticated]) def import_upload(request: Request, entity_type: str) -> Response: """Create and enqueue a bulk-import job. ``POST /api/v1/import/{entity_type}/`` Body: ``{ "file_id": }`` """ tenant = _get_tenant_or_404(request) if tenant is None: return Response({"detail": "Tenant not found."}, status=status.HTTP_404_NOT_FOUND) ser = ImportJobCreateSerializer(data=request.data) if not ser.is_valid(): return Response(ser.errors, status=status.HTTP_400_BAD_REQUEST) from simorgh.apps.memberships.models import Membership membership = Membership.objects.filter( tenant=tenant, users=request.user, status="active" ).first() org_node_id = membership.organization_node_id if membership else None try: job = create_import_job( tenant_id=tenant.pk, organization_node_id=org_node_id, entity_type=entity_type, file_id=ser.validated_data["file_id"], actor=request.user, enqueue=True, ) except PlatformCoreError as exc: return Response({"detail": str(exc)}, status=status.HTTP_400_BAD_REQUEST) return Response(ImportJobSerializer(job).data, status=status.HTTP_201_CREATED) @api_view(["POST"]) @permission_classes([IsAuthenticated]) def import_cancel(request: Request, job_id: str) -> Response: """Cancel a pending import job. ``POST /api/v1/import/jobs/{job_id}/cancel/`` """ tenant = _get_tenant_or_404(request) if tenant is None: return Response({"detail": "Tenant not found."}, status=status.HTTP_404_NOT_FOUND) try: job_pk = int(job_id) except (ValueError, TypeError): return Response({"detail": "Invalid job ID."}, status=status.HTTP_400_BAD_REQUEST) job = get_import_job(job_pk, tenant_id=tenant.pk) if job is None: return Response({"detail": "Not found."}, status=status.HTTP_404_NOT_FOUND) try: job = cancel_import_job(job, actor=request.user) except PlatformCoreError as exc: return Response({"detail": str(exc)}, status=status.HTTP_400_BAD_REQUEST) return Response(ImportJobSerializer(job).data) @api_view(["GET"]) @permission_classes([IsAuthenticated]) def import_status(request: Request, job_id: str) -> Response: """Return status of a single import job. ``GET /api/v1/import/jobs/{job_id}/status/`` """ tenant = _get_tenant_or_404(request) if tenant is None: return Response({"detail": "Tenant not found."}, status=status.HTTP_404_NOT_FOUND) try: job_pk = int(job_id) except (ValueError, TypeError): return Response({"detail": "Invalid job ID."}, status=status.HTTP_400_BAD_REQUEST) job = get_import_job(job_pk, tenant_id=tenant.pk) if job is None: return Response({"detail": "Not found."}, status=status.HTTP_404_NOT_FOUND) return Response(ImportJobSerializer(job).data) @api_view(["GET"]) @permission_classes([IsAuthenticated]) def import_errors(request: Request, job_id: str) -> Response: """Return the row-level error report for a finished import job. ``GET /api/v1/import/jobs/{job_id}/errors/`` """ tenant = _get_tenant_or_404(request) if tenant is None: return Response({"detail": "Tenant not found."}, status=status.HTTP_404_NOT_FOUND) try: job_pk = int(job_id) except (ValueError, TypeError): return Response({"detail": "Invalid job ID."}, status=status.HTTP_400_BAD_REQUEST) job = get_import_job(job_pk, tenant_id=tenant.pk) if job is None: return Response({"detail": "Not found."}, status=status.HTTP_404_NOT_FOUND) return Response(ImportJobErrorSerializer(job).data) @api_view(["GET"]) @permission_classes([IsAuthenticated]) def import_list(request: Request) -> Response: """List import jobs for the current tenant. ``GET /api/v1/import/jobs/`` Query params: ``entity_type``, ``status``, ``limit`` (default 100). """ tenant = _get_tenant_or_404(request) if tenant is None: return Response({"detail": "Tenant not found."}, status=status.HTTP_404_NOT_FOUND) entity_type = request.query_params.get("entity_type") or None job_status = request.query_params.get("status") or None limit = min(500, max(1, int(request.query_params.get("limit", 100)))) qs = list_import_jobs( tenant.pk, entity_type=entity_type, status=job_status, limit=limit ) return Response(ImportJobSerializer(list(qs), many=True).data) @api_view(["GET"]) @permission_classes([IsAuthenticated]) def import_template(request: Request, entity_type: str) -> Response: """Return a CSV template with headers for *entity_type*. ``GET /api/v1/import/{entity_type}/template/`` """ from io import StringIO import csv as _csv from django.http import HttpResponse from simorgh.apps.platform_core.import_registry import import_registry try: spec = import_registry.get_or_raise(entity_type) except KeyError: return Response( {"detail": f"No importer registered for {entity_type!r}."}, status=status.HTTP_404_NOT_FOUND, ) buf = StringIO() writer = _csv.writer(buf) writer.writerow([col.key for col in spec.columns]) buf.seek(0) response = HttpResponse(buf.read(), content_type="text/csv") safe_name = entity_type.replace("/", "_").replace(" ", "_") response["Content-Disposition"] = f'attachment; filename="{safe_name}_template.csv"' return response # --------------------------------------------------------------------------- # Export Engine # --------------------------------------------------------------------------- @api_view(["POST"]) @permission_classes([IsAuthenticated]) def export_create(request, entity_type: str): """``POST /api/v1/export/{entity_type}/`` -- create and enqueue an export job.""" serializer = ExportJobCreateSerializer(data=request.data) if not serializer.is_valid(): return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) data = serializer.validated_data membership = request.user.memberships.filter( tenant=request.tenant, status="active" ).first() if membership is None: return Response({"detail": "No active membership."}, status=status.HTTP_403_FORBIDDEN) try: job = create_export_job( tenant_id=request.tenant.pk, organization_node_id=membership.organization_node_id, entity_type=entity_type, fmt=data["format"], filters=data.get("filters", {}), actor=request.user, enqueue=True, ) except PlatformCoreError as exc: return Response({"detail": str(exc)}, status=status.HTTP_400_BAD_REQUEST) return Response(ExportJobSerializer(job).data, status=status.HTTP_201_CREATED) @api_view(["GET"]) @permission_classes([IsAuthenticated]) def export_list(request): """``GET /api/v1/export/jobs/`` -- list export jobs for the current tenant.""" entity_type = request.query_params.get("entity_type") job_status = request.query_params.get("status") jobs = list_export_jobs( request.tenant.pk, entity_type=entity_type or None, status=job_status or None, ) return Response(ExportJobSerializer(jobs, many=True).data) @api_view(["GET"]) @permission_classes([IsAuthenticated]) def export_status(request, job_id: str): """``GET /api/v1/export/jobs/{job_id}/status/`` -- get job status.""" job = get_export_job(job_id, tenant_id=request.tenant.pk) if job is None: return Response({"detail": "Export job not found."}, status=status.HTTP_404_NOT_FOUND) return Response(ExportJobSerializer(job).data) @api_view(["GET"]) @permission_classes([IsAuthenticated]) def export_download(request, job_id: str): """``GET /api/v1/export/jobs/{job_id}/download/`` -- stream the export file.""" job = get_export_job(job_id, tenant_id=request.tenant.pk) if job is None: return Response({"detail": "Export job not found."}, status=status.HTTP_404_NOT_FOUND) if job.file_id is None: return Response( {"detail": "Export file not ready yet."}, status=status.HTTP_404_NOT_FOUND, ) try: from simorgh.apps.storage.services import get_file_url url = get_file_url(job.file) from django.shortcuts import redirect return redirect(url) except Exception: pass return Response( {"detail": "File download unavailable."}, status=status.HTTP_404_NOT_FOUND, ) @api_view(["POST"]) @permission_classes([IsAuthenticated]) def export_cancel(request, job_id: str): """``POST /api/v1/export/jobs/{job_id}/cancel/`` -- cancel a pending export job.""" job = get_export_job(job_id, tenant_id=request.tenant.pk) if job is None: return Response({"detail": "Export job not found."}, status=status.HTTP_404_NOT_FOUND) try: job = cancel_export_job(job, actor=request.user) except PlatformCoreError as exc: return Response({"detail": str(exc)}, status=status.HTTP_400_BAD_REQUEST) return Response(ExportJobSerializer(job).data) # --------------------------------------------------------------------------- # Print Engine # --------------------------------------------------------------------------- @api_view(["GET", "POST"]) @permission_classes([IsAuthenticated]) def print_template_list(request): """``GET /api/v1/print/templates/`` -- list templates for the current tenant. ``POST /api/v1/print/templates/`` -- create a new template. """ if request.method == "GET": entity_type = request.query_params.get("entity_type") is_active_param = request.query_params.get("is_active") is_active = None if is_active_param is not None: is_active = is_active_param.lower() in ("1", "true", "yes") qs = list_print_templates( request.tenant.pk, entity_type=entity_type, is_active=is_active, ) return Response(PrintTemplateSerializer(qs, many=True).data) # POST ser = PrintTemplateCreateSerializer(data=request.data) ser.is_valid(raise_exception=True) d = ser.validated_data # Resolve organization_node_id from the user's active membership from simorgh.apps.memberships.models import Membership membership = Membership.objects.filter( tenant=request.tenant, users=request.user, status="active" ).first() org_node_id = membership.organization_node_id if membership else None tmpl = create_print_template( tenant_id=request.tenant.pk, organization_node_id=org_node_id, entity_type=d["entity_type"], name=d["name"], html_template=d["html_template"], css=d.get("css", ""), paper_size=d["paper_size"], orientation=d["orientation"], actor=request.user, ) return Response(PrintTemplateSerializer(tmpl).data, status=status.HTTP_201_CREATED) @api_view(["GET", "PATCH", "DELETE"]) @permission_classes([IsAuthenticated]) def print_template_detail(request, template_id: int): """``GET/PATCH/DELETE /api/v1/print/templates/{template_id}/``.""" tmpl = get_print_template(template_id, tenant_id=request.tenant.pk) if tmpl is None: return Response( {"detail": "Print template not found."}, status=status.HTTP_404_NOT_FOUND, ) if request.method == "GET": return Response(PrintTemplateSerializer(tmpl).data) if request.method == "PATCH": ser = PrintTemplateUpdateSerializer(data=request.data, partial=True) ser.is_valid(raise_exception=True) try: tmpl = update_print_template(tmpl, actor=request.user, **ser.validated_data) except PlatformCoreError as exc: return Response({"detail": str(exc)}, status=status.HTTP_400_BAD_REQUEST) return Response(PrintTemplateSerializer(tmpl).data) # DELETE delete_print_template(tmpl, actor=request.user) return Response(status=status.HTTP_204_NO_CONTENT) @api_view(["POST"]) @permission_classes([IsAuthenticated]) def print_template_preview(request, template_id: int): """``POST /api/v1/print/templates/{template_id}/preview/`` Renders the template with the supplied entity data and returns HTML. This is safe to display in an iframe or sandboxed preview pane. """ tmpl = get_print_template(template_id, tenant_id=request.tenant.pk) if tmpl is None: return Response( {"detail": "Print template not found."}, status=status.HTTP_404_NOT_FOUND, ) ser = PrintPreviewRequestSerializer(data=request.data) ser.is_valid(raise_exception=True) d = ser.validated_data try: html = render_html_preview( tmpl, entity=d.get("entity", {}), extra_context=d.get("extra_context") or None, ) except PlatformCoreError as exc: return Response({"detail": str(exc)}, status=status.HTTP_400_BAD_REQUEST) from django.http import HttpResponse return HttpResponse(html, content_type="text/html; charset=utf-8") @api_view(["POST"]) @permission_classes([IsAuthenticated]) def print_generate(request, app_label: str, model_name: str, object_id: str, template_id: int): """``POST /api/v1/content/{app}/{model}/{id}/print/{template_id}/`` Fetch the target object, render the template as PDF, and return the PDF file as ``application/pdf``. """ from django.apps import apps as django_apps tmpl = get_print_template(template_id, tenant_id=request.tenant.pk) if tmpl is None: return Response( {"detail": "Print template not found."}, status=status.HTTP_404_NOT_FOUND, ) # Resolve model try: model = django_apps.get_model(app_label, model_name) except LookupError: return Response( {"detail": f"Model '{app_label}.{model_name}' not found."}, status=status.HTTP_404_NOT_FOUND, ) # Fetch entity with tenant scope obj = model.objects.filter(pk=object_id, tenant=request.tenant).first() if obj is None: return Response( {"detail": "Object not found."}, status=status.HTTP_404_NOT_FOUND, ) # Convert to a plain dict for template context entity_dict: dict = {} for field in obj._meta.get_fields(): if hasattr(field, "attname"): entity_dict[field.name] = getattr(obj, field.attname, None) elif not field.is_relation: entity_dict[field.name] = getattr(obj, field.name, None) try: pdf_bytes = render_pdf(tmpl, entity=entity_dict) except PlatformCoreError as exc: return Response({"detail": str(exc)}, status=status.HTTP_400_BAD_REQUEST) from django.http import HttpResponse filename = f"{model_name}_{object_id}.pdf" response = HttpResponse(pdf_bytes, content_type="application/pdf") response["Content-Disposition"] = f'attachment; filename="{filename}"' return response # --------------------------------------------------------------------------- # Template Engine — Message Templates # --------------------------------------------------------------------------- @api_view(["GET", "POST"]) @permission_classes([IsAuthenticated]) def message_template_list(request: Request) -> Response: """List or create message templates. GET /api/v1/templates/messages/ POST /api/v1/templates/messages/ """ tenant = _get_tenant_or_404(request) if tenant is None: return Response({"detail": "Tenant not found."}, status=status.HTTP_404_NOT_FOUND) if request.method == "GET": channel = request.query_params.get("channel") or None language = request.query_params.get("language") or None is_active_param = request.query_params.get("is_active") is_active = ( is_active_param.lower() in ("1", "true") if is_active_param is not None else None ) qs = list_message_templates( tenant.pk, channel=channel, language=language, is_active=is_active ) return Response(MessageTemplateSerializer(list(qs), many=True).data) # POST — create from simorgh.apps.memberships.models import Membership serializer = MessageTemplateCreateSerializer(data=request.data) if not serializer.is_valid(): return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) membership = Membership.objects.filter( tenant=request.tenant, users=request.user, status="active" ).first() organization_node_id = membership.organization_node_id if membership else None try: tmpl = create_message_template( tenant_id=tenant.pk, organization_node_id=organization_node_id, actor=request.user, **serializer.validated_data, ) except PlatformCoreError as exc: return Response({"detail": str(exc)}, status=status.HTTP_400_BAD_REQUEST) return Response(MessageTemplateSerializer(tmpl).data, status=status.HTTP_201_CREATED) @api_view(["GET", "PATCH", "DELETE"]) @permission_classes([IsAuthenticated]) def message_template_detail(request: Request, template_id: int) -> Response: """Retrieve, update, or delete a message template. GET /api/v1/templates/messages/{id}/ PATCH /api/v1/templates/messages/{id}/ DELETE /api/v1/templates/messages/{id}/ """ tenant = _get_tenant_or_404(request) if tenant is None: return Response({"detail": "Tenant not found."}, status=status.HTTP_404_NOT_FOUND) tmpl = get_message_template(template_id, tenant_id=tenant.pk) if tmpl is None: return Response({"detail": "Not found."}, status=status.HTTP_404_NOT_FOUND) if request.method == "GET": return Response(MessageTemplateSerializer(tmpl).data) if request.method == "PATCH": serializer = MessageTemplateUpdateSerializer(data=request.data) if not serializer.is_valid(): return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) try: tmpl = update_message_template(tmpl, actor=request.user, **serializer.validated_data) except PlatformCoreError as exc: return Response({"detail": str(exc)}, status=status.HTTP_400_BAD_REQUEST) return Response(MessageTemplateSerializer(tmpl).data) # DELETE try: delete_message_template(tmpl, actor=request.user) except PlatformCoreError as exc: return Response({"detail": str(exc)}, status=status.HTTP_400_BAD_REQUEST) return Response(status=status.HTTP_204_NO_CONTENT) @api_view(["POST"]) @permission_classes([IsAuthenticated]) def message_template_preview(request: Request, template_id: int) -> Response: """Preview a rendered message template body. POST /api/v1/templates/messages/{id}/preview/ Request body: ``{"context": {...}, "language": "en"}`` Returns: ``{"subject": "...", "body": "..."}`` """ tenant = _get_tenant_or_404(request) if tenant is None: return Response({"detail": "Tenant not found."}, status=status.HTTP_404_NOT_FOUND) tmpl = get_message_template(template_id, tenant_id=tenant.pk) if tmpl is None: return Response({"detail": "Not found."}, status=status.HTTP_404_NOT_FOUND) serializer = MessageTemplatePreviewSerializer(data=request.data) if not serializer.is_valid(): return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) ctx = serializer.validated_data.get("context", {}) from simorgh.apps.platform_core.services import _get_jinja2_env env = _get_jinja2_env() try: rendered_body = env.from_string(tmpl.body).render(**ctx) rendered_subject = env.from_string(tmpl.subject).render(**ctx) if tmpl.subject else "" except Exception as exc: return Response({"detail": f"Render error: {exc}"}, status=status.HTTP_400_BAD_REQUEST) return Response({"subject": rendered_subject, "body": rendered_body})