bedrock.meta.model

  1import datetime  # pragma: unit
  2import decimal  # pragma: unit
  3import uuid  # pragma: unit
  4import yaml  # pragma: unit
  5from enum import Enum  # pragma: unit
  6from sqlalchemy.orm import RelationshipProperty  # pragma: unit
  7from bedrock._helpers.string import snake_case_to_camelCase  # pragma: unit
  8from bedrock._helpers.classes import find_all_attributes  # pragma: unit
  9from bedrock.endpoints.endpoint import Endpoint  # pragma: unit
 10
 11
 12def model_has_relationships(model_cls):  # pragma: unit
 13    """
 14    Check if the model has relationships to other ones (which means the endpoint supports `load_options`).
 15    :param model_cls: Class of the model
 16    """
 17    if issubclass(model_cls, Enum):
 18        return False
 19    properties = [
 20        getattr(model_cls, prop) for prop in find_all_attributes(model_cls) if prop not in ['metadata', 'registry']
 21    ]
 22    schema_properties = [isinstance(p.prop, RelationshipProperty) for p in properties if p.key != "uuid"]
 23    return True in schema_properties
 24
 25
 26def convert_model_to_schema(model_cls, name=None) -> dict:  # pragma: unit
 27    """
 28    Converts a model class to an OpenAPI schema definition.
 29    :param model_cls: The model class
 30    :param name: Name of the schema
 31    """
 32    if name is None:
 33        name = model_cls.__name__
 34
 35    if issubclass(model_cls, Enum):
 36        return {
 37            name: {
 38                "type": "string",
 39                "enum": [e.name for e in model_cls]
 40            }
 41        }
 42    if model_cls.has_custom_schema():
 43        custom_schema = model_cls.as_schema()
 44        if isinstance(custom_schema, str):
 45            return {
 46                name: yaml.load(model_cls.as_schema(), yaml.CLoader)[name]
 47            }
 48        return {
 49            name: custom_schema[name]
 50        }
 51    return auto_convert_model_to_schema(model_cls, name)
 52
 53
 54def auto_convert_model_to_schema(model_cls, name=None):  # pragma: unit
 55    properties = [getattr(model_cls, prop) for prop in find_all_attributes(model_cls) if
 56                  prop not in ['metadata', 'registry', *model_cls.__excluded_attributes_from_json__]]
 57    looped_properties = get_looping_properties(model_cls).keys()
 58    schema_properties = [model_property_to_schema_property(p) for p in properties if
 59                         p.key != "uuid" and p.key not in looped_properties]
 60    joined_schema_properties = {}
 61    for schema_property in sorted(schema_properties, key=lambda d: list(d.keys())):
 62        joined_schema_properties.update(schema_property)
 63    return {
 64        name: {
 65            "type": "object",
 66            "properties": joined_schema_properties
 67        }
 68    }
 69
 70
 71def get_looping_properties(model_cls, as_camel_case=False) -> dict:  # pragma: unit
 72    """
 73    Get the properties in nested columns of model_cls that loop back to model_cls itself.
 74    :param model_cls:
 75    :return:
 76    """
 77    if issubclass(model_cls, Enum):
 78        return {}
 79    properties = [getattr(model_cls, prop) for prop in find_all_attributes(model_cls) if
 80                  prop not in ['metadata', 'registry', *model_cls.__excluded_attributes_from_json__]]
 81    looped_properties = {}
 82    for prop in properties:
 83        if isinstance(prop.prop, RelationshipProperty):
 84            related_model = prop.prop.entity.entity
 85            related_model_properties = [getattr(related_model, prop) for prop in find_all_attributes(related_model) if
 86                                        prop not in ['metadata', 'registry',
 87                                                     *related_model.__excluded_attributes_from_json__]]
 88            if model_cls in [p.prop.entity.entity for p in related_model_properties if
 89                             isinstance(p.prop, RelationshipProperty)]:
 90                key_name = prop.key if not as_camel_case else snake_case_to_camelCase(prop.key)
 91                if key_name.endswith("s"):
 92                    looped_properties[key_name] = {
 93                        "type": "array",
 94                        "items": {
 95                            "allOf": [
 96                                {
 97                                    "$ref": f"../schemas/{model_as_schema_filename(related_model)}#/{model_as_schema_name(related_model)}"
 98                                },
 99                                {
100                                    "properties": {
101                                        "uuid": {
102                                            "type": "string",
103                                            "example": "43f5e9fd-1ffd-4469-8e01-f6d1b0b78d34"
104                                        }
105                                    }
106                                }
107                            ]
108                        }
109                    }
110                else:
111                    looped_properties[key_name] = {
112                        "$ref": f"../schemas/{model_as_schema_filename(related_model)}#/{model_as_schema_name(related_model)}"
113                    }
114    return looped_properties
115
116
117def model_property_to_schema_property(prop) -> dict:  # pragma: unit
118    """
119    Resolves the property into a schema property.
120    See `resolve_schema_type` for more information.
121    :param prop: The property to resolve.
122    """
123    return {
124        snake_case_to_camelCase(prop.key): resolve_schema_type(prop.prop)
125    }
126
127
128def resolve_schema_type(prop) -> dict:  # pragma: unit
129    """
130    Resolve the schema type for a given SQLAlchemy property.
131
132    This makes use of `python_type_to_openapi_type` for simple types.
133
134    If the property is of type `RelationshipProperty`, it'll return a reference to the schema of the related model.
135
136    It uses the plurality of the property name to determine if it's a single entity or a list of entities. I.e.:
137    * `account` means `{"$ref": "account.yml"}`
138    * `accounts` means `{"type": "array", "items": {"$ref": "account.yml"}}`.
139    :param prop: The property to resolve the schema type for.
140    """
141    if isinstance(prop, RelationshipProperty):
142        model_cls = prop.entity.entity
143        schema_location = f"./{model_as_schema_filename(model_cls)}"
144        if prop.key.endswith("s"):
145            return {
146                "type": "array",
147                "items": {"$ref": f"{schema_location}#/{model_as_schema_name(model_cls)}"}
148            }
149        return {"$ref": f"{schema_location}#/{model_as_schema_name(model_cls)}"}
150
151    # For calculated columns annotated with calculated_type(), use the explicit type.
152    if hasattr(prop, '_calculated_type'):
153        try:
154            python_type = prop._calculated_type.python_type
155        except NotImplementedError:
156            python_type = str
157        is_nullable = getattr(prop.columns[0], 'nullable', True)
158        return resolve_python_type_as_openapi_schema_type(prop.key, python_type, is_nullable)
159
160    column = prop.columns[0]
161
162    # Fallback for unannotated column_property (e.g., scalar subqueries without calculated_type).
163    try:
164        python_type = column.type.python_type
165    except NotImplementedError:
166        python_type = str
167
168    is_nullable = getattr(column, 'nullable', True)
169    return resolve_python_type_as_openapi_schema_type(prop.key, python_type, is_nullable)
170
171
172def resolve_python_type_as_openapi_schema_type(name: str, python_type: type or None, is_nullable: bool,
173                                               values: list = None) -> dict:  # pragma: unit
174    data_type = python_type_to_openapi_type(python_type, name, custom_enum_values=values)
175    if is_nullable:
176        return {**data_type, "nullable": True}
177    return data_type
178
179
180def resolve_python_type_as_json_schema_type(name: str, python_type: type or None, is_nullable: bool,
181                                            values: list = None) -> dict:  # pragma: unit
182    openapi_schema_type = python_type_to_openapi_type(python_type, name, custom_enum_values=values)
183    try:
184        del openapi_schema_type["example"]
185    except KeyError:
186        pass
187    if is_nullable:
188        if isinstance(openapi_schema_type["type"], list):  # pragma: no cover
189            # this case doesn't currently happen because python_type_to_openapi_type doesn't predict dual types (e.g. ["string", "number"])
190            openapi_schema_type["type"] = [*openapi_schema_type["type"], "null"]
191        elif openapi_schema_type["type"] != "null":
192            openapi_schema_type["type"] = [openapi_schema_type["type"], "null"]
193    return openapi_schema_type
194
195
196def model_as_schema_filename(model_cls) -> str:  # pragma: unit
197    """
198    Get the intended YAML filename for a given model/endpoint.
199    :param model_cls: The model class.
200    """
201    return f"{model_as_schema_name(model_cls)}.yml"
202
203
204def model_as_schema_name(model_cls) -> str:  # pragma: unit
205    """
206    Get the intended name for a given model/endpoint.
207    :param model_cls: The model class.
208    """
209    if not model_cls:
210        raise ValueError("A non-null model_cls is required")
211    if issubclass(model_cls, Endpoint) and model_cls.__body_schema_name__:
212        return model_cls.__body_schema_name__
213    return model_cls.__name__
214
215
216def python_type_to_openapi_type(data_type: type or None, column_name: str, is_query_param=False,
217                                custom_enum_values: list = None) -> dict:  # pragma: unit
218    """
219    Convert a Python type to an OpenAPI type.
220    :param data_type: Python type to convert.
221    :param column_name: Name of the column (used for examples).
222    :param is_query_param: Whether the type is for a query parameter (as opposed to a json object).
223    :param custom_enum_values: Custom enum values to use (if the type is an Enum).
224    :return: A dictionary with the OpenAPI type.
225
226    Examples:
227    * `bool` -> `{"type": "boolean"}`
228    * `list` -> `{"type": "array", "items": {"type": "string"}}`
229    """
230    if data_type is None:
231        return {"type": "null"}
232    if data_type == bool:
233        return {"type": "boolean", "example": True}
234    if data_type == int:
235        return {"type": "integer", "example": 123456}
236    if data_type == list:
237        if is_query_param:
238            return {}
239        return {
240            "type": "array",
241            "items": {"type": "string"},
242            "example": [f"{column_name}-A", f"{column_name}-B"]
243        }
244    if data_type == dict:
245        if is_query_param:
246            return {}
247        return {
248            "type": "object",
249            "example": {"key1": "value1", "key2": "value2"}
250        }
251    if data_type in [float, complex, decimal.Decimal]:
252        return {"type": "number", "example": 1.23}
253    if data_type == datetime.datetime:
254        if is_query_param:
255            return {
256                "type": "string",
257                "examples": {
258                    "iso datetime (Z)": {
259                        "value": "2023-12-14T17:32:28Z",
260                        "summary": "ISO datetime (Z is alias for +00:00)"
261                    },
262                    "iso datetime": {
263                        "value": "2023-12-14T17:32:28+00:00",
264                        "summary": "ISO datetime"
265                    },
266                    "date": {
267                        "value": "2023-12-14",
268                        "summary": "Date (where time is 00:00:00)"
269                    },
270                    "date and time": {
271                        "value": "2023-12-14 17:32",
272                        "summary": "Date and time (where seconds are 00)"
273                    },
274                    "unix": {
275                        "value": "1702575148000",
276                        "summary": "Unix timestamp (in milliseconds)"
277                    }
278                }
279            }
280        return {"type": "number", "example": 1707998714}
281    if data_type == uuid.UUID:
282        return {"type": "string", "format": "uuid", "example": "43f5e9fd-1ffd-4469-8e01-f6d1b0b78d34"}
283    if issubclass(data_type, Enum):
284        enum_values = custom_enum_values if custom_enum_values else [e.name for e in data_type]
285        return {"type": "string", "example": enum_values[0], "enum": enum_values}
286    return {"type": "string", "example": f"{column_name}"}