Coverage for fastapi_restly / schemas / _generator.py: 95%
168 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-10-09 18:41 +0000
1"""
2Schema generation utilities for auto-generating Pydantic schemas from SQLAlchemy models.
3"""
5import enum
6import sys
7import types
8from collections.abc import Collection
9from datetime import date, datetime, time
10from decimal import Decimal
11from typing import Any, ForwardRef, Union, get_args, get_origin
12from uuid import UUID
14import pydantic
15from pydantic import Field
16from sqlalchemy import Column
17from sqlalchemy import inspect as sa_inspect
18from sqlalchemy.orm import ColumnProperty, Mapped, RelationshipProperty
20from .._mapping import is_mapped_class
21from ._base import (
22 BaseSchema,
23 IDSchema,
24 ReadOnly,
25 TimestampsSchemaMixin,
26 _is_unresolved,
27 _own_annotations,
28)
31def get_sqlalchemy_field_type(field: Any) -> Any:
32 """
33 Extract the Python type from a SQLAlchemy Mapped field.
35 :param field: A SQLAlchemy Mapped field.
36 :returns: The Python type annotation.
37 """
38 # Get the type annotation from the Mapped field
39 if hasattr(field, "type"):
40 return field.type
41 elif hasattr(field, "__origin__"):
42 return field.__origin__
43 else:
44 # Fallback to Any if we can't determine the type
45 return Any
48def is_relationship_field(field: Any) -> bool:
49 """
50 Check if a field is a SQLAlchemy relationship.
52 :param field: A SQLAlchemy Mapped field.
53 :returns: True if the field is a relationship, False otherwise.
54 """
55 if isinstance(field, RelationshipProperty):
56 return True
57 return isinstance(getattr(field, "property", None), RelationshipProperty)
60def get_relationship_target_model(field: Any) -> type[Any] | None:
61 """
62 Get the target model class for a relationship field.
64 :param field: A SQLAlchemy relationship field.
65 :returns: The target model class, or None if not found.
66 """
67 if not is_relationship_field(field):
68 return None
70 # Try to get the target from the relationship property
71 relationship = field
72 if not isinstance(relationship, RelationshipProperty):
73 relationship = getattr(field, "property", None)
75 if (
76 relationship is not None
77 and hasattr(relationship, "mapper")
78 and hasattr(relationship.mapper, "class_")
79 ):
80 return relationship.mapper.class_
82 # Try to get from the type annotation
83 if hasattr(field, "type"): 83 ↛ 93line 83 didn't jump to line 93 because the condition on line 83 was always true
84 target_type = field.type
85 if hasattr(target_type, "__origin__") and target_type.__origin__ is list:
86 # Handle list[Model] case
87 args = get_args(target_type)
88 if args: 88 ↛ 93line 88 didn't jump to line 93 because the condition on line 88 was always true
89 return args[0]
90 elif is_mapped_class(target_type): 90 ↛ 93line 90 didn't jump to line 93 because the condition on line 90 was always true
91 return target_type
93 return None
96def _resolve_annotation(owner: type, annotation: Any) -> Any:
97 """Evaluate a string annotation against ``owner``'s module, or return None."""
98 if isinstance(annotation, ForwardRef): 98 ↛ 99line 98 didn't jump to line 99 because the condition on line 98 was never true
99 annotation = annotation.__forward_arg__
100 if isinstance(annotation, str):
101 module = sys.modules.get(owner.__module__)
102 # Module names win over class attributes, as in typing.get_type_hints:
103 # a column named ``date`` must not shadow the ``date`` type.
104 try:
105 annotation = eval(
106 annotation, dict(vars(owner)), dict(vars(module)) if module else {}
107 )
108 except Exception:
109 return None
110 return annotation
113def _model_annotations(model_cls: type, keys: Collection[str]) -> dict[str, Any]:
114 """Map each annotated name in ``keys`` to its inner type, or None if unresolvable.
116 ``Mapped[X]`` gives ``X``; a plain annotation (SQLModel) is kept as is.
117 The class nearest ``model_cls`` on the MRO wins.
118 """
119 annotations: dict[str, Any] = {}
120 for cls in model_cls.mro():
121 for name, annotation in _own_annotations(cls).items():
122 if name in annotations or name not in keys:
123 continue
124 annotation = _resolve_annotation(cls, annotation)
125 if get_origin(annotation) is Mapped:
126 args = get_args(annotation)
127 annotation = _resolve_annotation(cls, args[0]) if args else None
128 if annotation is not None and _is_unresolved(annotation):
129 annotation = None
130 annotations[name] = annotation
131 return annotations
134def _split_optional(annotation: Any) -> tuple[Any, bool]:
135 """Return ``(type, is_optional)`` with ``None`` removed from a union."""
136 if isinstance(annotation, types.UnionType) or get_origin(annotation) is Union:
137 union_args = get_args(annotation)
138 if type(None) in union_args: 138 ↛ 141line 138 didn't jump to line 141 because the condition on line 138 was always true
139 non_none_types = [arg for arg in union_args if arg is not type(None)]
140 return (non_none_types[0] if non_none_types else annotation), True
141 return annotation, False
144def get_model_fields(model_cls: type[Any]) -> dict[str, Any]:
145 """
146 Extract field information from a SQLAlchemy model.
148 The mapper decides which attributes exist; annotations only refine a
149 column's type. Fields follow annotation order, then the mapper's order.
151 :param model_cls: A SQLAlchemy model class.
152 :returns: Dictionary mapping field names to their types and metadata.
153 :raises TypeError: If a column's Python type can be read from neither its
154 annotation nor its column type.
155 """
156 fields: dict[str, Any] = {}
158 mapper = sa_inspect(model_cls)
159 props = {prop.key: prop for prop in mapper.attrs}
160 annotations = _model_annotations(model_cls, props.keys())
161 names = list(annotations)
162 names += [name for name in props if name not in annotations]
164 for name in names:
165 if name.startswith("_"): 165 ↛ 166line 165 didn't jump to line 166 because the condition on line 165 was never true
166 continue
167 prop = props[name]
169 if isinstance(prop, RelationshipProperty):
170 target_model = prop.mapper.class_
171 # Listed so a caller can tell a relationship from a column.
172 # derive_schema leaves relationships out.
173 fields[name] = {
174 "type": list[target_model] if prop.uselist else target_model,
175 "is_relationship": True,
176 "target_model": target_model,
177 "is_optional": True,
178 "is_read_only": False,
179 "default": None,
180 }
181 continue
183 if not isinstance(prop, ColumnProperty): 183 ↛ 185line 183 didn't jump to line 185 because the condition on line 183 was never true
184 # Synonyms and composites are views of attributes listed already.
185 continue
187 column = prop.columns[0]
188 annotation = annotations.get(name)
189 if annotation is not None:
190 field_type, is_optional = _split_optional(annotation)
191 else:
192 try:
193 field_type = column.type.python_type
194 except NotImplementedError:
195 field_type = object
196 if field_type is object:
197 raise TypeError(
198 f"Cannot determine the Python type of "
199 f"{model_cls.__name__}.{name} for an auto-generated schema: "
200 f"it has no annotation that resolves at runtime, and its "
201 f"column type {column.type!r} has no concrete python_type. Annotate "
202 f"it as Mapped[...] or set schema= on the view."
203 ) from None
204 is_optional = bool(getattr(column, "nullable", False))
206 default = None
207 column_default = getattr(column, "default", None)
208 server_default = getattr(column, "server_default", None)
209 if column_default is not None or server_default is not None:
210 default = column_default or server_default
211 is_optional = True
213 fields[name] = {
214 "type": field_type,
215 "is_relationship": False,
216 "target_model": None,
217 "is_optional": is_optional,
218 # A column_property over an expression cannot be written.
219 "is_read_only": not isinstance(column, Column),
220 "default": default,
221 }
223 return fields
226def derive_schema(model: type[Any], *, name: str | None = None) -> type[BaseSchema]:
227 """
228 Generate a Pydantic schema from a SQLAlchemy model.
230 This is the same schema that a view generates when it has no ``schema``:
231 one field per column, and no relationship fields. A foreign key column,
232 such as ``author_id``, is an ordinary field. The fields ``id``,
233 ``created_at`` and ``updated_at``, and read-only columns such as a
234 ``column_property``, are marked ``ReadOnly``.
236 :param model: The SQLAlchemy model class.
237 :param name: Name for the generated schema class. Defaults to the
238 model name followed by ``Schema``, as in ``UserSchema``.
239 :returns: A Pydantic schema class.
240 """
241 if name is None:
242 name = f"{model.__name__}Schema"
244 # Get field information from the model
245 model_fields = get_model_fields(model)
247 # Determine base classes - start with the most specific ones
248 bases: list[type] = []
250 # Check if model has timestamp fields (inherits from TimestampsMixin)
251 has_timestamps = "created_at" in model_fields and "updated_at" in model_fields
252 if has_timestamps:
253 bases.append(TimestampsSchemaMixin)
255 # Check if model has an id field (inherits from IDBase)
256 has_id = "id" in model_fields
257 if has_id:
258 bases.append(IDSchema)
260 # Always include BaseSchema as the base
261 bases.append(BaseSchema)
263 # Create field definitions for the schema
264 field_definitions: dict[str, Any] = {}
265 read_only_fields: list[str] = []
267 for field_name, field_info in model_fields.items():
268 if field_info["is_relationship"]:
269 continue
271 if (
272 field_name in ["id", "created_at", "updated_at"]
273 or field_info["is_read_only"]
274 ):
275 read_only_fields.append(field_name)
277 pydantic_type = convert_sqlalchemy_type_to_pydantic(
278 field_info["type"], field_info["is_optional"]
279 )
281 # Add field to definitions - use proper Pydantic field format
282 # Don't include SQLAlchemy defaults as they're not JSON-serializable
283 if field_info["is_optional"]:
284 field_definitions[field_name] = (pydantic_type, Field(default=None))
285 else:
286 field_definitions[field_name] = (pydantic_type, ...)
288 # Apply ReadOnly annotation to read-only fields
289 for field_name in read_only_fields:
290 if field_name in field_definitions: 290 ↛ 289line 290 didn't jump to line 289 because the condition on line 290 was always true
291 original_type, field_info = field_definitions[field_name]
292 # Apply ReadOnly annotation to the type
293 field_definitions[field_name] = (ReadOnly[original_type], field_info)
295 # Create the schema class using pydantic.create_model
296 # The model's module, so OpenAPI's long names for this class point to
297 # the project and not to Restly.
298 schema_cls = pydantic.create_model( # type: ignore[call-overload]
299 name,
300 __doc__=f"Auto-generated schema for {model.__name__}",
301 __base__=tuple(bases),
302 __module__=model.__module__,
303 **field_definitions,
304 )
306 return schema_cls
309def convert_sqlalchemy_type_to_pydantic(
310 sqlalchemy_type: Any, is_optional: bool = False
311) -> Any:
312 """
313 Convert a SQLAlchemy type to a Pydantic-compatible type.
315 :param sqlalchemy_type: The SQLAlchemy type.
316 :param is_optional: Whether the field is optional.
317 :returns: A Pydantic-compatible type.
318 """
319 type_name = getattr(sqlalchemy_type, "__name__", str(sqlalchemy_type))
321 if sqlalchemy_type is Any:
322 pydantic_type = Any
323 elif sqlalchemy_type in (
324 str,
325 int,
326 float,
327 bool,
328 dict,
329 list,
330 datetime,
331 date,
332 time,
333 UUID,
334 Decimal,
335 ):
336 pydantic_type = sqlalchemy_type
337 elif isinstance(sqlalchemy_type, type) and issubclass(sqlalchemy_type, enum.Enum):
338 pydantic_type = sqlalchemy_type
339 elif is_mapped_class(sqlalchemy_type):
340 # Relationship targets are replaced with nested schemas later.
341 pydantic_type = sqlalchemy_type
342 elif getattr(sqlalchemy_type, "__origin__", None) is not None:
343 # Preserve parameterized container types like dict[str, Any] or list[int].
344 pydantic_type = sqlalchemy_type
345 elif type_name in {"Text", "String"}:
346 pydantic_type = str
347 elif type_name in {"Integer"}:
348 pydantic_type = int
349 elif type_name in {"Float"}:
350 pydantic_type = float
351 elif type_name in {"Boolean"}:
352 pydantic_type = bool
353 elif type_name in {"DateTime"}:
354 pydantic_type = datetime
355 elif type_name in {"Date"}:
356 pydantic_type = date
357 elif type_name in {"Time"}:
358 pydantic_type = time
359 else:
360 raise TypeError(
361 f"Unsupported field type for auto-generated schema: {sqlalchemy_type!r}"
362 )
364 # Handle optional types
365 if is_optional:
366 pydantic_type = pydantic_type | None
368 return pydantic_type