"""Read-only query selectors for platform_core generic services. These are the *only* entry points for reading Attachment, Comment, Activity, and CustomFieldValue data. Views and services call selectors; selectors never mutate state. Conventions ----------- - All functions accept an *entity* (any saved TenantScopedModel instance) as the first argument and return a lazy ``QuerySet`` unless noted. - The QuerySet is already filtered to the entity's tenant and content type; callers may add further ``.filter()`` clauses if needed. """ from __future__ import annotations from typing import TYPE_CHECKING, Any from django.contrib.contenttypes.models import ContentType from django.db.models import QuerySet from simorgh.apps.platform_core.models import ( Activity, Attachment, Comment, CustomFieldDefinition, CustomFieldValue, DocumentTemplate, ExportJob, ExportLayout, FormDefinition, FormSubmission, ImportJob, MessageTemplate, Note, PrintTemplate, Tag, TagAssignment, ) if TYPE_CHECKING: from django.db.models import Model __all__ = [ "list_attachments", "list_comments", "list_activity", "get_custom_field_values", "list_custom_field_definitions", "get_tags_for_entity", "find_entities_by_tag", "list_tags", "list_notes", "list_form_definitions", "list_form_submissions", # Import engine "list_import_jobs", "get_import_job", # Export engine "list_export_jobs", "get_export_job", # Print engine "list_print_templates", "get_print_template", # Template engine "list_message_templates", "get_message_template", # Document template engine "list_document_templates", "get_document_template", # Export layout engine "list_export_layouts", "get_export_layout", ] def _ct_and_id(entity: Model) -> tuple[ContentType, str]: """Return (ContentType, str(pk)) for *entity*. Raises ValueError if unsaved.""" if not hasattr(entity, "pk") or entity.pk is None: raise ValueError("entity must be a saved model instance") ct = ContentType.objects.get_for_model(entity.__class__) return ct, str(entity.pk) def list_attachments(entity: Model) -> QuerySet: """Return a QuerySet of :class:`Attachment` for *entity*, ordered by ``sort_order`` then ``-created_at``. The QuerySet selects ``file`` via ``select_related`` so callers get the full ``FileMetadata`` object without extra queries. """ ct, object_id = _ct_and_id(entity) return ( Attachment.objects.filter(content_type=ct, object_id=object_id) .select_related("file", "uploaded_by") .order_by("sort_order", "-created_at") ) def list_comments(entity: Model, *, include_replies: bool = True) -> QuerySet: """Return a QuerySet of top-level :class:`Comment` for *entity*. Args: entity: The entity whose comments to list. include_replies: When ``True`` (default) top-level comments are returned together with their nested ``replies`` via ``prefetch_related``. When ``False`` only root comments (``parent=None``) are returned. """ ct, object_id = _ct_and_id(entity) qs = ( Comment.objects.filter(content_type=ct, object_id=object_id, parent=None) .select_related("author") .order_by("created_at") ) if include_replies: qs = qs.prefetch_related("replies__author") return qs def list_activity(entity: Model, *, limit: int = 50) -> QuerySet: """Return the most recent *limit* :class:`Activity` records for *entity*, newest first. Args: entity: The entity whose activity feed to return. limit: Maximum number of records (default 50, hard-capped at 200). """ limit = min(limit, 200) ct, object_id = _ct_and_id(entity) return ( Activity.objects.filter(content_type=ct, object_id=object_id) .select_related("actor") .order_by("-occurred_at")[:limit] ) def get_custom_field_values(entity: Model) -> dict[str, Any]: """Return a ``{key: value}`` mapping of all custom field values for *entity*. This is a convenience function that returns a plain dict rather than a QuerySet because callers typically need random access by key. """ ct, object_id = _ct_and_id(entity) tenant_id = getattr(entity, "tenant_id", None) qs = ( CustomFieldValue.objects.filter( content_type=ct, object_id=object_id, tenant_id=tenant_id, ) .select_related("definition") ) return {row.definition.key: row.value for row in qs} def list_custom_field_definitions( tenant_id: int, entity_type: str, ) -> QuerySet: """Return a QuerySet of :class:`CustomFieldDefinition` for *entity_type* within *tenant_id*, ordered by ``sort_order`` then ``key``. Args: tenant_id: Tenant PK to scope the definitions to. entity_type: Dotted ``app_label.model_name`` string, e.g. ``"crm.lead"``. """ return ( CustomFieldDefinition.objects.filter( tenant_id=tenant_id, entity_type=entity_type, ) .order_by("sort_order", "key") ) # --------------------------------------------------------------------------- # Tags # --------------------------------------------------------------------------- def list_tags(tenant_id: int, *, entity_type: str | None = None) -> QuerySet: """Return a QuerySet of :class:`Tag` for *tenant_id*. Args: tenant_id: Tenant PK. entity_type: When given, filter to tags scoped to this entity type **plus** global tags (``entity_type=""``). When ``None`` (default), return all tags. """ qs = Tag.objects.filter(tenant_id=tenant_id).order_by("name") if entity_type is not None: from django.db.models import Q qs = qs.filter(Q(entity_type="") | Q(entity_type=entity_type)) return qs def get_tags_for_entity(entity: Model) -> QuerySet: """Return a QuerySet of :class:`Tag` assigned to *entity*. The QuerySet is ordered by ``name`` and does not include tags that belong to other entities in the tenant but are not assigned to *entity*. """ ct, object_id = _ct_and_id(entity) assignment_ids = TagAssignment.objects.filter( content_type=ct, object_id=object_id, ).values_list("tag_id", flat=True) return Tag.objects.filter(pk__in=assignment_ids).order_by("name") def find_entities_by_tag( tenant_id: int, tag_slug: str, entity_type: str | None = None, ) -> QuerySet: """Return a QuerySet of :class:`TagAssignment` for all entities tagged with *tag_slug* in *tenant_id*. Args: tenant_id: Tenant PK. tag_slug: Slug of the tag to look up. entity_type: When given (e.g. ``"crm.lead"``), further filters to assignments whose ``content_type`` matches. """ from django.contrib.contenttypes.models import ContentType as CT qs = TagAssignment.objects.filter( tenant_id=tenant_id, tag__slug=tag_slug, ).select_related("tag", "content_type") if entity_type: try: app_label, model_name = entity_type.split(".", 1) ct = CT.objects.get_by_natural_key(app_label, model_name) qs = qs.filter(content_type=ct) except (ValueError, CT.DoesNotExist): return TagAssignment.objects.none() return qs # --------------------------------------------------------------------------- # Notes # --------------------------------------------------------------------------- def list_notes(entity: Model, *, viewer_id: int | None = None) -> QuerySet: """Return notes for *entity*, filtered by visibility for *viewer_id*. Visibility rules: * ``private`` — only returned when ``viewer_id`` matches the author. * ``team`` — returned for all authenticated users (team-scoped). * ``workspace`` — returned for all authenticated users (tenant-scoped). When *viewer_id* is ``None`` (unauthenticated), only ``"team"`` and ``"workspace"`` notes are returned. Args: entity: The entity whose notes to list. viewer_id: PK of the requesting user, or ``None`` for anonymous callers. """ from django.db.models import Q ct, object_id = _ct_and_id(entity) qs = Note.objects.filter(content_type=ct, object_id=object_id).select_related("author") if viewer_id is not None: qs = qs.filter( Q(visibility__in=["team", "workspace"]) | Q(author_id=viewer_id) ) else: qs = qs.exclude(visibility="private") return qs.order_by("-is_pinned", "-created_at") # --------------------------------------------------------------------------- # Dynamic Forms # --------------------------------------------------------------------------- def list_form_definitions(tenant_id: int, *, is_active: bool | None = None) -> QuerySet: """Return a QuerySet of :class:`FormDefinition` for *tenant_id*. Args: tenant_id: Tenant PK. is_active: When given, filter by the active flag. """ qs = FormDefinition.objects.filter(tenant_id=tenant_id).order_by("-created_at") if is_active is not None: qs = qs.filter(is_active=is_active) return qs def list_form_submissions( form: FormDefinition, *, processed: bool | None = None, limit: int = 100, ) -> QuerySet: """Return a QuerySet of :class:`FormSubmission` for *form*, newest first. Args: form: The :class:`FormDefinition` to query submissions for. processed: When given, filter by the processed flag. limit: Maximum records returned (default 100, hard-capped at 500). """ limit = min(limit, 500) qs = ( FormSubmission.objects.filter(form=form) .select_related("submitted_by") .order_by("-created_at") ) if processed is not None: qs = qs.filter(processed=processed) return qs[:limit] # --------------------------------------------------------------------------- # Import Engine # --------------------------------------------------------------------------- def list_import_jobs( tenant_id: int, *, entity_type: str | None = None, status: str | None = None, limit: int = 100, ) -> QuerySet: """Return a QuerySet of :class:`ImportJob` for *tenant_id*, newest first. Args: tenant_id: Tenant to scope the query to. entity_type: When given, filter to this entity type. status: When given, filter to this status value. limit: Maximum records returned (default 100, hard-capped at 500). """ limit = min(limit, 500) qs = ( ImportJob.objects.filter(tenant_id=tenant_id) .select_related("created_by", "file") .order_by("-created_at") ) if entity_type is not None: qs = qs.filter(entity_type=entity_type) if status is not None: qs = qs.filter(status=status) return qs[:limit] def get_import_job(job_id: int, *, tenant_id: int) -> ImportJob | None: """Return the :class:`ImportJob` with *job_id* scoped to *tenant_id*, or None.""" return ( ImportJob.objects.filter(pk=job_id, tenant_id=tenant_id) .select_related("created_by", "file") .first() ) # --------------------------------------------------------------------------- # Export Engine # --------------------------------------------------------------------------- def list_export_jobs( tenant_id: int, *, entity_type: str | None = None, status: str | None = None, limit: int = 100, ) -> QuerySet: """Return a QuerySet of :class:`ExportJob` for *tenant_id*, newest first.""" limit = min(limit, 500) qs = ( ExportJob.objects.filter(tenant_id=tenant_id) .select_related("created_by", "file") .order_by("-created_at") ) if entity_type is not None: qs = qs.filter(entity_type=entity_type) if status is not None: qs = qs.filter(status=status) return qs[:limit] def get_export_job(job_id: int, *, tenant_id: int) -> ExportJob | None: """Return the :class:`ExportJob` with *job_id* scoped to *tenant_id*, or None.""" return ( ExportJob.objects.filter(pk=job_id, tenant_id=tenant_id) .select_related("created_by", "file") .first() ) # --------------------------------------------------------------------------- # Print engine # --------------------------------------------------------------------------- def list_print_templates( tenant_id: int, *, entity_type: str | None = None, is_active: bool | None = None, ) -> QuerySet[PrintTemplate]: """Return print templates for *tenant_id*, optionally filtered.""" qs = PrintTemplate.objects.filter(tenant_id=tenant_id).select_related("created_by") if entity_type is not None: qs = qs.filter(entity_type=entity_type) if is_active is not None: qs = qs.filter(is_active=is_active) return qs.order_by("entity_type", "name") def get_print_template(template_id: int, *, tenant_id: int) -> PrintTemplate | None: """Return the :class:`PrintTemplate` with *template_id* scoped to *tenant_id*, or None.""" return PrintTemplate.objects.filter(pk=template_id, tenant_id=tenant_id).first() # --------------------------------------------------------------------------- # Template engine (Message Templates) # --------------------------------------------------------------------------- def list_message_templates( tenant_id: int, *, channel: str | None = None, language: str | None = None, is_active: bool | None = None, ) -> QuerySet[MessageTemplate]: """Return :class:`MessageTemplate` records for *tenant_id*. Optionally filtered by *channel*, *language*, and/or *is_active*. """ qs = MessageTemplate.objects.filter(tenant_id=tenant_id).select_related("created_by") if channel is not None: qs = qs.filter(channel=channel) if language is not None: qs = qs.filter(language=language) if is_active is not None: qs = qs.filter(is_active=is_active) return qs.order_by("code", "language") def get_message_template(template_id: int, *, tenant_id: int) -> MessageTemplate | None: """Return the :class:`MessageTemplate` with *template_id* scoped to *tenant_id*, or None.""" return MessageTemplate.objects.filter(pk=template_id, tenant_id=tenant_id).first() # --------------------------------------------------------------------------- # Document template engine # --------------------------------------------------------------------------- def list_document_templates( tenant_id: int, *, entity_type: str | None = None, is_active: bool | None = None, ) -> QuerySet[DocumentTemplate]: """Return :class:`DocumentTemplate` records for *tenant_id*. Optionally filtered by *entity_type* and/or *is_active*. """ qs = DocumentTemplate.objects.filter(tenant_id=tenant_id).select_related("created_by") if entity_type is not None: qs = qs.filter(entity_type=entity_type) if is_active is not None: qs = qs.filter(is_active=is_active) return qs.order_by("entity_type", "code") def get_document_template(template_id: int, *, tenant_id: int) -> DocumentTemplate | None: """Return the :class:`DocumentTemplate` with *template_id* scoped to *tenant_id*, or None.""" return DocumentTemplate.objects.filter(pk=template_id, tenant_id=tenant_id).first() # --------------------------------------------------------------------------- # Export layout engine # --------------------------------------------------------------------------- def list_export_layouts( tenant_id: int, *, entity_type: str | None = None, export_format: str | None = None, is_active: bool | None = None, ) -> QuerySet[ExportLayout]: """Return :class:`ExportLayout` records for *tenant_id*. Optionally filtered by *entity_type*, *export_format*, and/or *is_active*. """ qs = ExportLayout.objects.filter(tenant_id=tenant_id).select_related("created_by") if entity_type is not None: qs = qs.filter(entity_type=entity_type) if export_format is not None: qs = qs.filter(export_format=export_format) if is_active is not None: qs = qs.filter(is_active=is_active) return qs.order_by("entity_type", "code") def get_export_layout(layout_id: int, *, tenant_id: int) -> ExportLayout | None: """Return the :class:`ExportLayout` with *layout_id* scoped to *tenant_id*, or None.""" return ExportLayout.objects.filter(pk=layout_id, tenant_id=tenant_id).first()