Support for complex json-schemas with external $ref

Author: phenobarbital

datamodel/abstract.py

@@ -23,6 +23,7 @@ class Meta:
     datasource: Optional[str] = None
     connection: Optional[Callable] = None
     remove_nulls: bool = False
+    endpoint: str = ""
 
 
 def set_connection(cls, conn: Callable):

datamodel/base.py

@@ -61,11 +61,12 @@ def _get_ref_info(_type, field):
             }
         }
     elif isinstance(_type, ModelMeta):
+        _schema = _type.schema(as_dict=True)
         return {
             "type": "object",
-            "enum_type": None,
-            "schema": _type.schema(as_dict=True),
-            "ref": field.metadata.get('fk').split("|") if 'fk' in field.metadata else _type.schema(as_dict=True)['$id']
+            "schema": _schema,
+            "$ref": _schema.get('$id', f"/schemas/{_type.__name__}"),
+            "columns": field.metadata.get('fk').split("|") if 'fk' in field.metadata else []
         }
     return None
 
@@ -411,6 +412,11 @@ def schema(cls, as_dict=False):
         except Exception:
             pass
 
+        try:
+            endpoint = cls.Meta.endpoint
+        except AttributeError:
+            endpoint = None
+
         schema = cls.Meta.schema
         table = cls.Meta.name.lower() if cls.Meta.name else title.lower()
         columns = cls.get_columns().items()
@@ -423,12 +429,16 @@ def schema(cls, as_dict=False):
             _type = field.type
             type_info = _get_type_info(_type, name, title)
             ref_info = _get_ref_info(_type, field)
+            if ref_info:
+                defs[name] = ref_info.pop('schema', None)
+            else:
+                ref_info = {}
+
             minimum = field.metadata.get('min', None)
             maximum = field.metadata.get('max', None)
             secret = field.metadata.get('secret', None)
-            label = field.metadata.get('label', None)
 
-            if field.metadata.get('required', False):
+            if field.metadata.get('required', False) or field.metadata.get('primary', False):
                 required.append(name)
 
             # UI objects:
@@ -441,24 +451,23 @@ def schema(cls, as_dict=False):
             fields[name] = {
                 "type": type_info,
                 "nullable": field.metadata.get('nullable', False),
-                "label": label,
+                "label": field.metadata.get('label', None),
                 "attrs": {
                     "placeholder": field.metadata.get('description', None),
                     "format": field.metadata.get('format', None),
                 },
                 "readOnly": field.metadata.get('readonly', False),
-                "writeOnly": False,
                 **ui_objects,
-                **schema_extra
+                **schema_extra,
+                **ref_info
             }
 
+            if 'write_only' in field.metadata:
+                fields[name]["writeOnly"] = field.metadata.get('write_only', False)
+
             if 'pattern' in field.metadata:
                 fields[name]["attrs"]["pattern"] = field.metadata['pattern']
 
-            ref = ref_info.get('ref') if ref_info else None
-            if ref:
-                fields[name]["$ref"] = ref
-
             if field.repr is False:
                 fields[name]["attrs"]["visible"] = False
 
@@ -477,17 +486,22 @@ def schema(cls, as_dict=False):
                 if maximum:
                     fields[name]['maximum'] = maximum
 
+        endpoint_kwargs = {}
+        if endpoint is not None:
+            endpoint_kwargs["endpoint"] = endpoint
+
         base_schema = {
             "$schema": "https://json-schema.org/draft/2020-12/schema",
             "$id": f"/schemas/{table}",
+            **endpoint_kwargs,
             "additionalProperties": cls.Meta.strict,
             "title": title,
             "description": description,
             "type": "object",
             "table": table,
             "schema": schema,
             "properties": fields,
-            "required": required
+            "required": required,
         }
 
         if defs:

datamodel/version.py

@@ -3,7 +3,7 @@
 __title__ = 'python-datamodel'
 __description__ = ('simple library based on python +3.8 to use Dataclass-syntax'
                    'for interacting with Data')
-__version__ = '0.6.5'
+__version__ = '0.6.6'
 __author__ = 'Jesus Lara'
 __author_email__ = '[email protected]'
 __license__ = 'BSD'

examples/form_schema.py

@@ -0,0 +1,103 @@
+from typing import Optional
+from datetime import datetime
+import pprint
+from enum import Enum
+from datamodel import BaseModel, Field
+
+pp = pprint.PrettyPrinter(width=41, compact=True)
+
+
+class Reward(BaseModel):
+    reward_id: int = Field(
+        primary_key=True, required=False, db_default="auto", repr=False
+    )
+    reward: str = Field(required=True, nullable=False)
+    description: str = Field(required=False)
+    points: int = Field(required=False, default=1)
+    programs: str = Field(required=False)
+    program_slug: str = Field(required=False, default="")
+    region: str = Field(required=False, default="")
+    department: list = Field(required=False, default_factory=list)
+    icon: str = Field(
+        required=False,
+        default="",
+        ui_widget="ImageUploader",
+        ui_help="Badge Icon, Hint: please use a transparent PNG."
+    )
+    attributes: Optional[dict] = Field(
+        required=False, default_factory=dict, db_type="jsonb"
+    )
+    resource_link: str = Field(required=False, default="")
+    effective_date: datetime = Field(required=False, default=datetime.now())
+    inserted_at: datetime = Field(required=False, default=datetime.now())
+
+    class Meta:
+        driver = "pg"
+        name = "rewards"
+        schema = "navigator"
+        app_label = "navigator"
+        strict = True
+
+class Employee(BaseModel):
+    associate_id: str = Field(required=True)
+    position_id: str = Field(required=True)
+    file_number: str = Field(required=True)
+    operator_name: str = Field(required=True)
+    first_name: str = Field(required=False)
+    last_name: str = Field(required=False)
+    display_name: str = Field(required=False)
+    corporate_email: str = Field(required=True)
+    job_code: str = Field(required=False)
+    job_code_title: str = Field(required=False)
+    region_code: str = Field(required=False)
+    department: str = Field(required=False)
+    department_code: str = Field(required=False)
+    location_code: str = Field(required=False)
+    work_location: str = Field(required=False)
+    reports_to_associate_oid: str = Field(required=False)
+    reports_to_associate_id: str = Field(required=False)
+    reports_to_position_id: str = Field(required=False)
+
+    def email(self):
+        return self.corporate_email
+
+    class Meta:
+        name = 'vw_active_employees'
+        schema = 'troc'
+        strict = True
+
+
+class BadgeAssign(BaseModel):
+    reward_id: Reward = Field(
+        required=True, fk='reward_id|reward', endpoint='rewards', label="Badge"
+    )
+    reward: str = Field(repr=False)
+    receiver_email: Employee = Field(
+        required=True,
+        fk='corporate_email|display_name',
+        api='adp_employees'
+    )
+    # receiver_user: User = Field(
+    #     required=True,
+    #     fk='user_id|display_name',
+    #     api='programs',
+    #     repr=False
+    # )
+    giver_user: int = Field(required=True)
+    giver_email: str = Field(required=True)
+    giver_employee: str = Field(required=False)
+    giver_message: str = Field(
+        ui_widget='textarea',
+        ui_help='Message to the receiver',
+        label="Message"
+    )
+
+    class Meta:
+        name = 'users_rewards'
+        schema = 'navigator'
+        endpoint: str = 'api/v1/assign_rewards'
+        strict = True
+
+schema = BadgeAssign.schema(as_dict=False)
+print(schema)
+# pp.pprint(schema)