[SDESK-7463] - Create new async PlanningTypes resource and service

server/planning/content_profiles/__init__.py

@@ -17,6 +17,11 @@
 from .service import PlanningTypesService
 from planning.common import get_config_event_related_item_search_provider_name
 
+from .planning_types_async_service import PlanningTypesAsyncService
+from .module import planning_types_resource_config
+
+__all__ = ["planning_types_resource_config", "PlanningTypesAsyncService"]
+
 
 def init_app(app: Eve):
     superdesk.privilege(

server/planning/content_profiles/module.py

@@ -0,0 +1,12 @@
+from superdesk.core.resources import ResourceConfig, RestEndpointConfig
+
+from planning.types import PlanningTypesResourceModel
+from .planning_types_async_service import PlanningTypesAsyncService
+
+
+planning_types_resource_config = ResourceConfig(
+    name="planning_types",
+    data_class=PlanningTypesResourceModel,
+    service=PlanningTypesAsyncService,
+    rest_endpoints=RestEndpointConfig(resource_methods=["GET", "POST"], item_methods=["GET", "PATCH"]),
+)

server/planning/content_profiles/planning_types_async_service.py

@@ -0,0 +1,129 @@
+# -*- coding: utf-8; -*-
+#
+# This file is part of Superdesk.
+#
+# Copyright 2021 Sourcefabric z.u. and contributors.
+#
+# For the full copyright and license information, please see the
+# AUTHORS and LICENSE files distributed with this source code, or
+# at https://www.sourcefabric.org/superdesk/license
+
+from typing import Any
+from copy import deepcopy
+
+from superdesk.core.resources import AsyncResourceService
+from superdesk.core.types import SearchRequest, ProjectedFieldArg
+from superdesk.utils import ListCursor
+
+from planning.common import planning_link_updates_to_coverage, get_config_event_related_item_search_provider_name
+from planning.types import PlanningTypesResourceModel
+from .profiles import DEFAULT_PROFILES
+
+
+class PlanningTypesAsyncService(AsyncResourceService[PlanningTypesResourceModel]):
+    """Planning types async service
+
+    Provide a service that returns what fields should be shown in the edit forms in planning, in the edit dictionary.
+    Also provide a schema to allow the client to validate the values entered in the forms.
+    Entries can be overridden by providing alternates in the planning_types mongo collection.
+    """
+
+    async def find_one(
+        self,
+        req: SearchRequest | None = None,
+        projection: ProjectedFieldArg | None = None,
+        use_mongo: bool = False,
+        version: int | None = None,
+        **lookup,
+    ) -> PlanningTypesResourceModel | None:
+        try:
+            if req is None:
+                planning_type = await super().find_one(
+                    projection=projection, use_mongo=use_mongo, version=version, **lookup
+                )
+            else:
+                planning_type = await self.find_one(req)
+
+            # lookup name from either **lookup of planning_item(if lookup has only '_id')
+            lookup_name = lookup.get("name")
+            if not lookup_name and planning_type:
+                lookup_name = planning_type.to_dict().get("name")
+
+            default_planning_type = deepcopy(
+                next(
+                    (ptype for ptype in DEFAULT_PROFILES if ptype.get("name") == lookup_name),
+                    {},
+                )
+            )
+            if not planning_type:
+                await self._remove_unsupported_fields(default_planning_type)
+                return PlanningTypesResourceModel(**default_planning_type)
+
+            await self.merge_planning_type(planning_type.to_dict(), default_planning_type)
+            return planning_type
+        except IndexError:
+            return None
+
+    async def get(self, req, lookup) -> ListCursor:
+        cursor = await super().search(lookup)
+        planning_types = await cursor.to_list_raw()
+        merged_planning_types = []
+
+        for default_planning_type in deepcopy(DEFAULT_PROFILES):
+            planning_type = next(
+                (p for p in planning_types if p.get("name") == default_planning_type.get("name")),
+                None,
+            )
+
+            # If nothing is defined in database for this planning_type, use default
+            if planning_type is None:
+                await self._remove_unsupported_fields(default_planning_type)
+                merged_planning_types.append(default_planning_type)
+            else:
+                await self.merge_planning_type(planning_type, default_planning_type)
+                merged_planning_types.append(planning_type)
+
+        return ListCursor(merged_planning_types)
+
+    async def merge_planning_type(self, planning_type: dict[str, Any], default_planning_type: dict[str, Any]):
+        # Update schema fields with database schema fields
+        default_type: dict[str, Any] = {"schema": {}, "editor": {}}
+        updated_planning_type = deepcopy(default_planning_type or default_type)
+
+        updated_planning_type.setdefault("groups", {})
+        updated_planning_type["groups"].update(planning_type.get("groups", {}))
+
+        if planning_type["name"] == "advanced_search":
+            updated_planning_type["schema"].update(planning_type.get("schema", {}))
+            updated_planning_type["editor"]["event"].update((planning_type.get("editor") or {}).get("event"))
+            updated_planning_type["editor"]["planning"].update((planning_type.get("editor") or {}).get("planning"))
+            updated_planning_type["editor"]["combined"].update((planning_type.get("editor") or {}).get("combined"))
+        elif planning_type["name"] in ["event", "planning", "coverage"]:
+            for config_type in ["editor", "schema"]:
+                planning_type.setdefault(config_type, {})
+                for field, options in updated_planning_type[config_type].items():
+                    # If this field is none, then it is of type `schema.NoneField()`
+                    # no need to copy any schema
+                    if updated_planning_type[config_type][field]:
+                        updated_planning_type[config_type][field].update(planning_type[config_type].get(field) or {})
+        else:
+            updated_planning_type["editor"].update(planning_type.get("editor", {}))
+            updated_planning_type["schema"].update(planning_type.get("schema", {}))
+
+        planning_type["schema"] = updated_planning_type["schema"]
+        planning_type["editor"] = updated_planning_type["editor"]
+        planning_type["groups"] = updated_planning_type["groups"]
+        await self._remove_unsupported_fields(planning_type)
+
+    async def _remove_unsupported_fields(self, planning_type: dict[str, Any]):
+        # Disable Event ``related_items`` field
+        # if ``EVENT_RELATED_ITEM_SEARCH_PROVIDER_NAME`` config is not set
+        if planning_type.get("name") == "event" and not get_config_event_related_item_search_provider_name():
+            planning_type["editor"].pop("related_items", None)
+            planning_type["schema"].pop("related_items", None)
+
+        # Disable Coverage ``no_content_linking`` field
+        # if ``PLANNING_LINK_UPDATES_TO_COVERAGES`` config is not ``True``
+        if planning_type.get("name") == "coverage" and not planning_link_updates_to_coverage():
+            planning_type["editor"].pop("no_content_linking", None)
+            planning_type["schema"].pop("no_content_linking", None)

server/planning/content_profiles/resource.py

@@ -88,3 +88,4 @@ class PlanningTypesResource(superdesk.Resource):
         "POST": "planning_manage_content_profiles",
         "PATCH": "planning_manage_content_profiles",
     }
+    internal_resource = True

server/planning/module.py

@@ -3,6 +3,7 @@
 from planning.planning import planning_resource_config
 from planning.assignments import assignments_resource_config, delivery_resource_config
 from planning.published import published_resource_config
+from planning.content_profiles import planning_types_resource_config
 
 
 module = Module(
@@ -13,5 +14,6 @@
         assignments_resource_config,
         published_resource_config,
         delivery_resource_config,
+        planning_types_resource_config,
     ],
 )

server/planning/types/__init__.py

@@ -21,6 +21,7 @@
 from .assignment import AssignmentResourceModel
 from .published import PublishedPlanningModel
 from .enums import PostStates, UpdateMethods, WorkflowState
+from .planning_types import PlanningTypesResourceModel
 
 
 __all__ = [
@@ -30,6 +31,7 @@
     "PlanningResourceModel",
     "AssignmentResourceModel",
     "PublishedPlanningModel",
+    "PlanningTypesResourceModel",
     "PlanningSchedule",
     "PostStates",
     "UpdateMethods",

server/planning/types/planning_types.py

@@ -0,0 +1,48 @@
+from datetime import datetime
+from pydantic import Field
+from typing import Annotated, Any
+
+from superdesk.core.resources import ResourceModel, fields
+from superdesk.utc import utcnow
+from superdesk.core.resources.fields import ObjectId
+from superdesk.core.resources.validators import validate_iunique_value_async, validate_data_relation_async
+
+
+class PlanningTypesResourceModel(ResourceModel):
+    # The name identifies the form in the UI to which the type relates
+    name: Annotated[fields.Keyword, validate_iunique_value_async("planning_types", "name")]
+
+    editor: dict[str, Any] = Field(
+        default_factory=dict,
+        description="Editor controls which fields are visible in the UI",
+    )
+    schema_config: dict[str, Any] = Field(
+        alias="schema",
+        default_factory=dict,
+        description="Schema controls the validation of fields at the front end",
+    )
+    groups: dict[str, Any] = Field(
+        default_factory=dict,
+        description="List of groups (and their translations) for grouping of fields in the editor",
+    )
+    post_schema: dict[str, Any] = Field(
+        alias="postSchema",
+        default_factory=dict,
+        description="Controls the validation of fields when posting",
+    )
+    list_fields_config: dict[str, Any] = Field(
+        alias="list",
+        default_factory=dict,
+        description="List fields when seeing events/planning during export/download",
+    )
+    export_list: list[str] = Field(
+        default_factory=list,
+        description="Fields visible in exports or downloads for events/planning",
+    )
+
+    # Audit information
+    created_by: Annotated[ObjectId, validate_data_relation_async("users")] | None = None
+    updated_by: Annotated[ObjectId, validate_data_relation_async("users")] | None = None
+    firstcreated: datetime = Field(default_factory=utcnow)
+    versioncreated: datetime = Field(default_factory=utcnow)
+    init_version: int