Dvl/datasource capabilities

src/DSC/schema_v2_utils.py

@@ -0,0 +1,362 @@
+from typing import Dict, List, Literal
+
+from forestadmin.agent_toolkit.utils.forest_schema.filterable import FrontendFilterableUtils
+from forestadmin.agent_toolkit.utils.forest_schema.type import (
+    ForestServerAction,
+    ForestServerCollection,
+    ForestServerField,
+    ForestServerSegment,
+    ServerValidationType,
+)
+from forestadmin.datasource_toolkit.interfaces.fields import ColumnAlias, Operator, PrimitiveType
+from typing_extensions import NotRequired, TypedDict
+
+
+# TYPING
+class SchemaV2Relation(TypedDict):
+    name: str
+    type: Literal["ManyToMany", "ManyToOne", "OneToOne", "OneToMany"]
+    foreignCollection: str
+    throughCollection: NotRequired[str]
+    foreignKey: NotRequired[str]
+    foreignKeyTarget: NotRequired[str]
+    originKey: NotRequired[str]
+    originKeyTarget: NotRequired[str]
+
+
+class SchemaV2Field(TypedDict):
+    name: str
+    type: ColumnAlias
+    filterOperators: List[str]
+    # defaultValue: NotRequired[Any] # TODO
+    # isRequired: NotRequired[Any] # TODO
+    enumerations: NotRequired[List[str]]
+    isPrimaryKey: NotRequired[bool]
+    isWritable: NotRequired[bool]
+    isSortable: NotRequired[bool]
+    validations: NotRequired[List[ServerValidationType]]
+
+
+class SchemaV2Collection(TypedDict):
+    name: str
+    fields: List[SchemaV2Field]  # to define
+    relations: List  # to define
+    segments: NotRequired[List[ForestServerSegment]]
+    actions: NotRequired[List[ForestServerAction]]
+    isSearchable: NotRequired[bool]
+    canList: NotRequired[bool]
+    canCreate: NotRequired[bool]
+    canUpdate: NotRequired[bool]
+    canDelete: NotRequired[bool]
+    canCount: NotRequired[bool]
+    canChart: NotRequired[bool]
+
+
+# MASKS
+SCHEMA_V2_FIELDS_MASK = {
+    "enumerations": None,
+    "isPrimaryKey": False,
+    "isWritable": True,
+    "isSortable": True,
+    "validations": [],
+}
+
+
+SCHEMA_V2_COLLECTION_MASK = {
+    "segments": [],
+    "actions": [],
+    "isSearchable": True,
+    "canList": True,
+    "canCreate": True,
+    "canUpdate": True,
+    "canDelete": True,
+    "canCount": True,
+    "canChart": True,
+}
+
+
+class SchemaV1toV2:
+    def __init__(self, schema_collections: List[ForestServerCollection]) -> None:
+        self.initial_schema: List[ForestServerCollection] = schema_collections
+
+    def translate(self) -> List[SchemaV2Collection]:
+        return self._convert_collection_schema_v1_to_v2(self.initial_schema)
+
+    def _convert_collection_schema_v1_to_v2(
+        self, schema_collections: List[ForestServerCollection]
+    ) -> List[SchemaV2Collection]:
+        schema_collections_v2: List[SchemaV2Collection] = []
+        for collection_v1 in schema_collections:
+            collection_v2: SchemaV2Collection = {
+                "name": collection_v1["name"],
+                "fields": self._convert_fields_v1_to_v2(collection_v1["fields"]),
+                "relations": self._convert_fields_to_relation(collection_v1["fields"]),
+                "actions": collection_v1["actions"],  # type:ignore
+                "isSearchable": collection_v1["isSearchable"],
+                "canList": True,
+                "canCreate": not collection_v1["isReadOnly"],
+                "canUpdate": not collection_v1["isReadOnly"],
+                "canDelete": not collection_v1["isReadOnly"],
+                "canCount": True,
+                "canChart": True,
+            }
+            schema_collections_v2.append(self._template_reduce_collection(collection_v2))
+
+        return schema_collections_v2
+
+    def _convert_fields_v1_to_v2(self, fields: List[ForestServerField]) -> List[SchemaV2Field]:
+        fields_v2: List[SchemaV2Field] = []
+        for field_v1 in fields:
+            if field_v1["relationship"] is not None:  # type:ignore
+                continue
+
+            fields_v2.append(
+                self._template_reduce_field(
+                    {
+                        "name": field_v1["field"],  # type:ignore
+                        "type": field_v1["type"],  # type:ignore
+                        "filterOperators": [
+                            op.value
+                            for op in FrontendFilterableUtils.OPERATOR_BY_TYPES[
+                                PrimitiveType(field_v1["field"])  # type:ignore
+                            ]  # type:ignore
+                        ],  # type:ignore
+                        "enumerations": field_v1["enums"] if field_v1["type"] == "Enum" else None,  # type:ignore
+                        "isPrimaryKey": field_v1["isPrimaryKey"],  # type:ignore
+                        "isSortable": field_v1["isSortable"],  # type:ignore
+                        "isWritable": not field_v1["isReadOnly"],  # type:ignore
+                        "validations": field_v1["validations"],  # type:ignore
+                    }
+                )
+            )
+
+        return fields_v2
+
+    def _convert_fields_to_relation(self, fields: List[ForestServerField]) -> List[SchemaV2Relation]:
+        relation_v2: List[SchemaV2Relation] = []
+        for field_v1 in fields:
+            if field_v1["relationship"] is None:  # type:ignore
+                continue
+
+            relation: SchemaV2Relation = {
+                "name": field_v1["field"],  # type:ignore
+                "foreignCollection": field_v1["reference"].split(".")[0],  # type:ignore
+            }  # type:ignore
+            if field_v1["relationship"] == "BelongsTo":  # type:ignore
+                relation = {
+                    **relation,
+                    "type": "ManyToOne",
+                    "foreignKeyTarget": field_v1["reference"].split(".")[1],  # type:ignore
+                    # TODO: may be impossible because v1 schema doesn't include foreign_keys
+                    "foreignKey": "TODO: N/A",  # type:ignore
+                    # doable in the case where it's the reverse of a 1to1
+                }
+
+            elif field_v1["relationship"] == "HasMany":  # type:ignore
+                relation = {
+                    **relation,
+                    "type": "OneToMany",
+                    "originKey": "",
+                    # TODO: may be impossible because v1 schema doesn't include foreign_keys
+                    "originKeyTarget": field_v1["reference"].split(".")[1],  # type:ignore # TODO: not sure about this
+                }
+
+            elif field_v1["relationship"] == "HasOne":  # type:ignore
+                relation = {
+                    **relation,
+                    "type": "OneToOne",
+                    "originKey": "",
+                    "originKeyTarget": field_v1["reference"].split(".")[1],  # type:ignore
+                }
+
+            elif field_v1["relationship"] == "BelongsToMany":  # type:ignore
+                reverse_relation = filter(  # type:ignore
+                    lambda x: x["fields"] == field_v1["inverseOf"],  # type:ignore
+                    self.initial_schema[relation["foreignCollection"]]["fields"],  # type:ignore
+                )[0]
+
+                relation = {
+                    **relation,
+                    "type": "ManyToMany",
+                    "foreignKeyTarget": field_v1["reference"].split(".")[1],  # type:ignore
+                    "originKeyTarget": reverse_relation["reference"].split(".")[1],
+                    # TODO: may be impossible because v1 schema doesn't include foreign_keys
+                    "originKey": "TODO: N/A",
+                    "foreignKey": "TODO: N/A",
+                    "throughCollection": "TODO: N/A",
+                }
+            relation_v2.append(relation)
+
+        return relation_v2
+
+    def _template_reduce_field(self, collection: SchemaV2Field) -> SchemaV2Field:
+        return self._reduce_from_template(collection, SCHEMA_V2_FIELDS_MASK)  # type:ignore
+
+    def _template_reduce_collection(self, collection: SchemaV2Collection) -> SchemaV2Collection:
+        return self._reduce_from_template(collection, SCHEMA_V2_COLLECTION_MASK)  # type:ignore
+
+    def _reduce_from_template(self, input, mask):
+        reduced = {}
+        for key, value in input:
+            if key not in mask or input[key] != mask[key]:
+                reduced[key] = value
+        return reduced
+
+
+class SchemaV2toV1:
+    def __init__(self, schema_collections: List[SchemaV2Collection]) -> None:
+        self.initial_schema = schema_collections
+
+    def translate(self) -> List[ForestServerCollection]:
+        return self._convert_collection_schema_v2_to_v1(self.initial_schema)
+
+    def _get_value_or_default(self, instance, key, mask):
+        return instance.get(key, mask[key])
+
+    def collection_get_value_or_default(self, instance, key):
+        return self._get_value_or_default(instance, key, SCHEMA_V2_COLLECTION_MASK)
+
+    def field_get_value_or_default(self, instance, key):
+        return self._get_value_or_default(instance, key, SCHEMA_V2_FIELDS_MASK)
+
+    def _convert_collection_schema_v2_to_v1(
+        self, collections_v2: List[SchemaV2Collection]
+    ) -> List[ForestServerCollection]:
+        schema_collections_v1: List[ForestServerCollection] = []
+        for collection_v2 in collections_v2:
+            schema_collections_v1.append(
+                {
+                    "name": collection_v2["name"],
+                    "icon": None,
+                    "integration": None,
+                    "isVirtual": False,
+                    "onlyForRelationships": False,
+                    "paginationType": "page",
+                    "fields": self.convert_fields_and_relation(collection_v2["fields"], collection_v2["relations"]),
+                    "isReadOnly": not collection_v2.get("canCreate", SCHEMA_V2_COLLECTION_MASK["canCreate"]),
+                    "isSearchable": collection_v2.get("isSearchable", SCHEMA_V2_COLLECTION_MASK["isSearchable"]),
+                    "actions": collection_v2.get("actions", SCHEMA_V2_COLLECTION_MASK["actions"]),
+                    "segments": collection_v2.get("segments", SCHEMA_V2_COLLECTION_MASK["segments"]),
+                }
+            )
+
+        return schema_collections_v1
+
+    def convert_fields_and_relation(
+        self, fields: List[SchemaV2Field], relations: List[SchemaV2Relation]
+    ) -> List[ForestServerField]:
+        fields_v1: Dict[str, ForestServerField] = {}
+
+        for field_v2 in fields:
+            fields_v1[field_v2["name"]] = {
+                "field": field_v2["name"],
+                "type": field_v2["type"],
+                "isPrimaryKey": field_v2.get("isPrimaryKey", SCHEMA_V2_FIELDS_MASK["isPrimaryKey"]),
+                "isReadOnly": not field_v2.get("isWritable", SCHEMA_V2_FIELDS_MASK["isWritable"]),
+                "isSortable": field_v2.get("isSortable", SCHEMA_V2_FIELDS_MASK["isSortable"]),
+                # "defaultValue": None # TODO: need to handle this
+                "isRequired": len(
+                    [
+                        *filter(
+                            lambda x: x["type"] == "is present",
+                            field_v2.get("validations", SCHEMA_V2_FIELDS_MASK["validations"]),
+                        )
+                    ]
+                )
+                > 0,  # type:ignore
+                "enums": (
+                    field_v2.get("enumerations", SCHEMA_V2_FIELDS_MASK["enumerations"])
+                    if field_v2["type"] == "Enum"
+                    else None
+                ),
+                "isFilterable": FrontendFilterableUtils.is_filterable(
+                    field_v2["type"],
+                    set([Operator(op) for op in field_v2["filterOperators"]]),
+                ),
+                "validations": field_v2.get("validations", SCHEMA_V2_FIELDS_MASK["validations"]),
+                "integration": None,
+                "isVirtual": False,
+                "inverseOf": None,
+                # "relationship": None, not set when it's a field
+            }
+
+        for rel_v2 in relations:
+            name = rel_v2["name"]
+
+            if rel_v2["type"] == "OneToMany":
+                related_field_v2: SchemaV2Field = [
+                    *filter(lambda x: x["name"] == rel_v2["originKey"], fields)  # type:ignore
+                ][0]
+                fields_v1[name] = {
+                    **fields_v1[name],
+                    "relationship": "HasMany",
+                }
+
+            elif rel_v2["type"] == "ManyToOne":
+                foreign_collection_fields = [
+                    *filter(lambda x: x["name"] == rel_v2["foreignCollection"], self.initial_schema)  # type:ignore
+                ][0]["fields"]
+                related_field_v2: SchemaV2Field = [
+                    *filter(lambda x: x["name"] == rel_v2["foreignKey"], foreign_collection_fields)  # type:ignore
+                ][0]
+
+                fields_v1[name] = {
+                    **fields_v1[name],
+                    "relationship": "BelongsTo",
+                }
+
+            elif rel_v2["type"] == "OneToOne":
+                related_field_v2: SchemaV2Field = [
+                    *filter(lambda x: x["name"] == rel_v2["originKey"], fields)  # type:ignore
+                ][0]
+                fields_v1[name] = {
+                    **fields_v1[name],
+                    "relationship": "HasOne",
+                }
+
+            elif rel_v2["type"] == "ManyToMany":
+                foreign_collection_fields = [
+                    *filter(lambda x: x["name"] == rel_v2["foreignCollection"], self.initial_schema)  # type:ignore
+                ][0]["fields"]
+                related_field_v2: SchemaV2Field = [
+                    *filter(lambda x: x["name"] == rel_v2["foreignKey"], foreign_collection_fields)  # type:ignore
+                ][0]
+
+                fields_v1[name] = {
+                    **fields_v1[name],
+                    "relationship": "BelongsToMany",
+                }
+
+            # fields_v1[rel_v2["name"]] = {
+            #     "field": rel_v2["name"],
+            #     "type": rel_v2["type"],
+            #     "isPrimaryKey": rel_v2.get("isPrimaryKey", SCHEMA_V2_FIELDS_MASK["isPrimaryKey"]),
+            #     "isReadOnly": not rel_v2.get("isWritable", SCHEMA_V2_FIELDS_MASK["isWritable"]),
+            #     "isSortable": rel_v2.get("isSortable", SCHEMA_V2_FIELDS_MASK["isSortable"]),
+            #     # "defaultValue": None # TODO: need to handle this
+            #     "isRequired": len(
+            #         [
+            #             *filter(
+            #                 lambda x: x["type"] == "is present",
+            #                 rel_v2.get("validations", SCHEMA_V2_FIELDS_MASK["validations"]),
+            #             )
+            #         ]
+            #     )
+            #     > 0,  # type:ignore
+            #     "enums": (
+            #         rel_v2.get("enumerations", SCHEMA_V2_FIELDS_MASK["enumerations"])
+            #         if rel_v2["type"] == "Enum"
+            #         else None
+            #     ),
+            #     "isFilterable": FrontendFilterableUtils.is_filterable(
+            #         rel_v2["type"],
+            #         set([Operator(op) for op in rel_v2["filterOperators"]]),
+            #     ),
+            #     "validations": rel_v2.get("validations", SCHEMA_V2_FIELDS_MASK["validations"]),
+            #     "integration": None,
+            #     "isVirtual": False,
+            #     "inverseOf": None,
+            #     # "relationship": None, not set when it's a field
+            # }
+
+        return [*fields_v1.values()]