From b842acc80723e683761e86514098a1b74f73a257 Mon Sep 17 00:00:00 2001 From: iklo Date: Fri, 18 Sep 2026 11:10:35 -0300 Subject: [PATCH 1/5] refactor: split the 150-line metaclass body into small functions RestEndpointMeta._process did seven numbered steps in one function: collect annotations, build columns, inject the auto columns, build schemas, parse Meta, scan the MRO for HTTP method mixins, pick the table source. Reading how one field becomes one column meant scrolling past the other six. Each step is a function now: _field_annotations, _declared_columns, _column_for, _auto_columns, _parsed_meta, _allowed_methods and _reject_redeclared_auto_fields. _process is the 20 lines that put them in order, and a reflected endpoint returns early instead of threading `if reflect` through every step. Three things that did nothing are gone: - cls._fields_info was stored on every endpoint and never read. - The block that removed "fields inherited from RestEndpoint itself" could not match anything: RestEndpoint only has underscore annotations, which the line above already skips. - ReflectedTable's partial_columns was always an empty list, because columns are only built when Meta.reflect is falsy. Partial reflection maps columns that already exist in the database; it never added any. Tests: 345 passed, 1 xfailed, 2 xpassed. --- lightapi/rest.py | 273 ++++++++++++++++++-------------------- lightapi/table_mapping.py | 8 +- 2 files changed, 134 insertions(+), 147 deletions(-) diff --git a/lightapi/rest.py b/lightapi/rest.py index 080cc80..069f193 100644 --- a/lightapi/rest.py +++ b/lightapi/rest.py @@ -98,7 +98,6 @@ def __new__( if is_base_only: cls._allowed_methods = set(_ALL_METHODS) cls._meta = {} - cls._fields_info = {} return cls mcs._process(cls, name, namespace) @@ -106,151 +105,144 @@ def __new__( @staticmethod def _process(cls: type, name: str, namespace: dict[str, Any]) -> None: - import typing as _typing + _reject_redeclared_auto_fields(name, namespace) - from pydantic.fields import FieldInfo + meta_obj = namespace.get("Meta") or getattr(cls, "Meta", None) + reflect = getattr(meta_obj, "reflect", False) if meta_obj else False + is_reflected = reflect is True or reflect == "full" or reflect == "partial" - # ── Step 1: Collect annotations ────────────────────────────────────── - # Use get_type_hints() so that PEP-563 string annotations (from - # `from __future__ import annotations`) are resolved to real types. - try: - resolved = _typing.get_type_hints(cls) - except Exception: - resolved = {} - - annotations: dict[str, Any] = {} - for base in reversed(cls.__mro__): - for k, v in getattr(base, "__annotations__", {}).items(): - if not k.startswith("_") and k not in _AUTO_FIELDS: - # Prefer the resolved type; fall back to the raw annotation. - annotations[k] = resolved.get(k, v) - - # Remove fields inherited from RestEndpoint itself - if "RestEndpoint" in [b.__name__ for b in cls.__mro__[1:]]: - for b in cls.__mro__[1:]: - if b.__name__ == "RestEndpoint": - for k in list(annotations): - if k in getattr(b, "__annotations__", {}): - annotations.pop(k, None) - break - - # Guard: user must not redeclare auto-injected fields - for auto in _AUTO_FIELDS: - if auto in namespace.get("__annotations__", {}): - raise ConfigurationError( - f"RestEndpoint '{name}': '{auto}' is auto-injected " - "and must not be redeclared." - ) + cls._meta = _parsed_meta(name, meta_obj) # type: ignore[attr-defined] + cls._allowed_methods = _allowed_methods(cls) # type: ignore[attr-defined] - # ── Step 2: Build SQLAlchemy columns ───────────────────────────────── - columns: list[Column] = [] - fields_info: dict[str, FieldInfo] = {} - - meta_class = namespace.get("Meta") or getattr(cls, "Meta", None) - reflect = getattr(meta_class, "reflect", False) if meta_class else False - - if not reflect: - for field_name, annotation in annotations.items(): - field_val = namespace.get(field_name) or getattr(cls, field_name, None) - fi = field_val if isinstance(field_val, FieldInfo) else None - if fi: - fields_info[field_name] = fi - - extra: dict[str, Any] = (fi.json_schema_extra or {}) if fi else {} - if extra.get("exclude"): - continue - - is_opt, inner = _is_optional(annotation) - col_type = _TYPE_MAP.get(inner) - if col_type is None: - raise ConfigurationError( - f"RestEndpoint '{name}': annotation '{inner}' on field " - f"'{field_name}' is not in the type map. " - "Add exclude=True to skip column generation." - ) - - col_kwargs: dict[str, Any] = {"nullable": is_opt} - col_args: list[Any] = [] - - if inner is Decimal: - scale = extra.get("decimal_places", 10) - col_type = Numeric(scale=scale) - - if extra.get("foreign_key"): - col_args.append(ForeignKey(extra["foreign_key"])) - if extra.get("unique"): - col_kwargs["unique"] = True - if extra.get("index"): - col_kwargs["index"] = True - - columns.append(Column(field_name, col_type, *col_args, **col_kwargs)) - - # ── Step 3: Auto-inject id / created_at / updated_at / version ─────── - auto_cols = [ - Column("id", Integer, primary_key=True, autoincrement=True), - Column("created_at", DateTime, default=datetime.datetime.utcnow), - Column( - "updated_at", - DateTime, - default=datetime.datetime.utcnow, - onupdate=datetime.datetime.utcnow, - ), - Column("version", Integer, default=1, nullable=False), - ] - - # ── Step 4: SchemaFactory ───────────────────────────────────────────── - if reflect is True or reflect == "full" or reflect == "partial": - # Built by ReflectedTable.map() once the columns are known. + if is_reflected: + # ReflectedTable.map() builds the schemas once the columns are known. cls.__schema_create__ = None # type: ignore[attr-defined] cls.__schema_read__ = None # type: ignore[attr-defined] - else: - cls.__schema_create__, cls.__schema_read__ = SchemaFactory.build(cls) # type: ignore[attr-defined] + cls._table_source = ReflectedTable(partial=reflect == "partial") # type: ignore[attr-defined] + return - # ── Step 5: Parse Meta → _meta ──────────────────────────────────────── - meta_obj = namespace.get("Meta") or getattr(cls, "Meta", None) - raw_serializer = getattr(meta_obj, "serializer", None) if meta_obj else None - - # Guard: Meta.serializer must be Serializer instance/subclass - if raw_serializer is not None: - from pydantic import BaseModel as PydanticBaseModel - - if isinstance(raw_serializer, type): - if issubclass(raw_serializer, PydanticBaseModel): - raise ConfigurationError( - f"Meta.serializer on '{name}' must be a Serializer " - "instance or subclass, not a BaseModel subclass." - ) - serialiser_normalised = normalise_serializer(raw_serializer) - - cls._meta = { # type: ignore[attr-defined] - "authentication": ( - getattr(meta_obj, "authentication", None) if meta_obj else None - ), - "filtering": getattr(meta_obj, "filtering", None) if meta_obj else None, - "pagination": getattr(meta_obj, "pagination", None) if meta_obj else None, - "serializer_normalised": serialiser_normalised, - "cache": getattr(meta_obj, "cache", None) if meta_obj else None, - "reflect": getattr(meta_obj, "reflect", False) if meta_obj else False, - "table": getattr(meta_obj, "table", None) if meta_obj else None, - } - cls._fields_info = fields_info # type: ignore[attr-defined] - - # ── Step 6: MRO scan for HttpMethod markers ─────────────────────────── - - allowed: set[str] = set() - for base in cls.__mro__: - if base is cls: - continue - http_method = getattr(base, "_http_method", None) - if http_method: - allowed.add(http_method) - cls._allowed_methods = allowed if allowed else set(_ALL_METHODS) # type: ignore[attr-defined] - - # ── Step 7: Table source, mapped by LightApi.register() ─────────────── - if reflect is True or reflect == "full" or reflect == "partial": - cls._table_source = ReflectedTable(columns, partial=reflect == "partial") # type: ignore[attr-defined] - else: - cls._table_source = DeclaredTable(auto_cols + columns) # type: ignore[attr-defined] + columns = [] if reflect else _declared_columns(cls, name, namespace) + cls.__schema_create__, cls.__schema_read__ = SchemaFactory.build(cls) # type: ignore[attr-defined] + cls._table_source = DeclaredTable(_auto_columns() + columns) # type: ignore[attr-defined] + + +def _reject_redeclared_auto_fields(name: str, namespace: dict[str, Any]) -> None: + for auto in _AUTO_FIELDS: + if auto in namespace.get("__annotations__", {}): + raise ConfigurationError( + f"RestEndpoint '{name}': '{auto}' is auto-injected " + "and must not be redeclared." + ) + + +def _field_annotations(cls: type) -> dict[str, Any]: + """Public annotated fields of the class and its bases, with real types.""" + import typing + + # get_type_hints() resolves the string annotations that + # `from __future__ import annotations` leaves behind. + try: + resolved = typing.get_type_hints(cls) + except Exception: + resolved = {} + + annotations: dict[str, Any] = {} + for base in reversed(cls.__mro__): + for field_name, raw in getattr(base, "__annotations__", {}).items(): + if not field_name.startswith("_") and field_name not in _AUTO_FIELDS: + annotations[field_name] = resolved.get(field_name, raw) + return annotations + + +def _declared_columns(cls: type, name: str, namespace: dict[str, Any]) -> list[Column]: + from pydantic.fields import FieldInfo + + columns: list[Column] = [] + for field_name, annotation in _field_annotations(cls).items(): + field_val = namespace.get(field_name) or getattr(cls, field_name, None) + field_info = field_val if isinstance(field_val, FieldInfo) else None + column = _column_for(name, field_name, annotation, field_info) + if column is not None: + columns.append(column) + return columns + + +def _column_for( + endpoint_name: str, field_name: str, annotation: Any, field_info: Any +) -> Column | None: + """The column for one annotated field; None when the field has exclude=True.""" + extra: dict[str, Any] = (field_info.json_schema_extra or {}) if field_info else {} + if extra.get("exclude"): + return None + + is_opt, inner = _is_optional(annotation) + col_type = _TYPE_MAP.get(inner) + if col_type is None: + raise ConfigurationError( + f"RestEndpoint '{endpoint_name}': annotation '{inner}' on field " + f"'{field_name}' is not in the type map. " + "Add exclude=True to skip column generation." + ) + if inner is Decimal: + col_type = Numeric(scale=extra.get("decimal_places", 10)) + + col_args = [ForeignKey(extra["foreign_key"])] if extra.get("foreign_key") else [] + col_kwargs: dict[str, Any] = {"nullable": is_opt} + if extra.get("unique"): + col_kwargs["unique"] = True + if extra.get("index"): + col_kwargs["index"] = True + return Column(field_name, col_type, *col_args, **col_kwargs) + + +def _auto_columns() -> list[Column]: + """id, created_at, updated_at and version, injected into every declared table.""" + return [ + Column("id", Integer, primary_key=True, autoincrement=True), + Column("created_at", DateTime, default=datetime.datetime.utcnow), + Column( + "updated_at", + DateTime, + default=datetime.datetime.utcnow, + onupdate=datetime.datetime.utcnow, + ), + Column("version", Integer, default=1, nullable=False), + ] + + +def _parsed_meta(name: str, meta_obj: Any) -> dict[str, Any]: + raw_serializer = getattr(meta_obj, "serializer", None) if meta_obj else None + + # normalise_serializer() rejects every non-Serializer; a pydantic model gets + # its own message because it is the mistake people actually make. + if isinstance(raw_serializer, type): + from pydantic import BaseModel as PydanticBaseModel + + if issubclass(raw_serializer, PydanticBaseModel): + raise ConfigurationError( + f"Meta.serializer on '{name}' must be a Serializer " + "instance or subclass, not a BaseModel subclass." + ) + + return { + "authentication": getattr(meta_obj, "authentication", None), + "filtering": getattr(meta_obj, "filtering", None), + "pagination": getattr(meta_obj, "pagination", None), + "serializer_normalised": normalise_serializer(raw_serializer), + "cache": getattr(meta_obj, "cache", None), + "reflect": getattr(meta_obj, "reflect", False), + "table": getattr(meta_obj, "table", None), + } + + +def _allowed_methods(cls: type) -> set[str]: + """HTTP methods named by HttpMethod mixins in the MRO; all of them when none is.""" + allowed = { + base._http_method + for base in cls.__mro__[1:] + if getattr(base, "_http_method", None) + } + return allowed or set(_ALL_METHODS) class RestEndpoint(metaclass=RestEndpointMeta): @@ -264,7 +256,6 @@ class RestEndpoint(metaclass=RestEndpointMeta): _table_source: TableSource _meta: dict[str, Any] _allowed_methods: set[str] - _fields_info: dict[str, Any] def __init__(self, **kwargs: Any) -> None: self._background: BackgroundTasks | None = None diff --git a/lightapi/table_mapping.py b/lightapi/table_mapping.py index 4f74c8a..2fc0d80 100644 --- a/lightapi/table_mapping.py +++ b/lightapi/table_mapping.py @@ -49,11 +49,10 @@ def map(self, endpoint_cls: type, session_manager: SessionManager) -> None: class ReflectedTable: """A table that already exists in the database (``Meta.reflect``). - With ``partial`` the endpoint's own fields are added to the reflected columns. + With ``partial`` the endpoint also declares some of those columns as fields. """ - def __init__(self, partial_columns: list[Column], partial: bool) -> None: - self._partial_columns = partial_columns + def __init__(self, partial: bool) -> None: self._partial = partial def map(self, endpoint_cls: type, session_manager: SessionManager) -> None: @@ -64,9 +63,6 @@ def map(self, endpoint_cls: type, session_manager: SessionManager) -> None: _table_name(endpoint_cls, session_manager), session_manager ) if self._partial: - for column in self._partial_columns: - if column.name not in table.c: - table.append_column(column) _strip_field_infos(endpoint_cls, list(table.c)) session_manager.registry.map_imperatively(endpoint_cls, table) From 80fa532f13d397e865707206384b809ee0b318eb Mon Sep 17 00:00:00 2001 From: iklo Date: Fri, 18 Sep 2026 11:16:07 -0300 Subject: [PATCH 2/5] refactor: break up LightApi.register, run and the constructor register() was 270 lines when this work started. After the table sources and the auth change it still validated the mapping, detected async mode, checked the engine, mapped tables, built routes and rebuilt the login routes, all inline. Now it reads as the four steps it performs. The checks are functions (_require_rest_endpoint, _has_async_handlers, _require_async_engine) and one endpoint's routes are built in _register_endpoint. The constructor delegates to _engine_from_url, _checked_mode and _as_rate_limiter. run() built its own Starlette app, a copy of build_app() with debug added. It calls build_app(debug=debug) now; debug is a new keyword-only argument with the old default. The login routes were torn out of self._routes and re-inserted on every register() call. They are composed in _asgi_routes() when the app is built: login routes, endpoint routes, /healthz, in the same order as before. The "queryset is async but engine is sync" warning is removed. It could not fire: an async queryset switches the app to async mode in the first pass, before the second pass that checked for it. lightapi.py: 910 lines on dev, 395 now. Tests: 345 passed, 1 xfailed, 2 xpassed. --- lightapi/lightapi.py | 269 ++++++++++++++++++------------------------- 1 file changed, 112 insertions(+), 157 deletions(-) diff --git a/lightapi/lightapi.py b/lightapi/lightapi.py index 2d63bfd..fdf6a83 100644 --- a/lightapi/lightapi.py +++ b/lightapi/lightapi.py @@ -6,7 +6,6 @@ import importlib import logging import os -import warnings from typing import Any, Callable import uvicorn @@ -66,50 +65,13 @@ def __init__( login_validator: Callable[[str, str], dict[str, Any] | None] | None = None, use_test_isolation: bool = False, ) -> None: - if engine is None and database_url: - engine = create_engine(database_url) - elif engine is None: - url = os.environ.get("LIGHTAPI_DATABASE_URL") - if url is None: - raise ConfigurationError( - "No database configured. Provide engine=..., database_url=..., or set " - "LIGHTAPI_DATABASE_URL environment variable." - ) - engine = create_engine(url) - - # Store engine first (we'll detect mode later) - self._engine = engine - - # Use explicit mode if provided, otherwise will be auto-detected in register() - if mode is not None: - if mode not in ("sync", "async"): - raise ConfigurationError( - f"mode must be 'sync' or 'async', got '{mode}'" - ) - self._mode = mode - else: - self._mode = "sync" # Will be auto-detected in register() - - # Create session manager + self._engine = engine if engine is not None else _engine_from_url(database_url) + self._mode = _checked_mode(mode) self._session_manager = session_manager or SessionManager( - engine, use_test_isolation=use_test_isolation + self._engine, use_test_isolation=use_test_isolation ) - - # Store login_validator for backward compatibility self._login_validator = login_validator - - # Rate limiter setup - self._rate_limiter_global: RateLimiter | None = None - if rate_limiter is not None: - if isinstance(rate_limiter, RateLimiter): - self._rate_limiter_global = rate_limiter - elif isinstance(rate_limiter, dict): - self._rate_limiter_global = RateLimiter( - **{ - limit: rate_limiter.get(limit, default) - for limit, default in _LOGIN_RATE_LIMITS.items() - } - ) + self._rate_limiter_global = _as_rate_limiter(rate_limiter) self._routes: list[Route] = [] self._endpoint_map: dict[str, type] = {} @@ -128,116 +90,60 @@ def register(self, mapping: dict[str, type]) -> None: mapping: ``{"/path": EndpointClass}`` dictionary. Each class must be a ``RestEndpoint`` subclass. """ - - # First pass: detect if any endpoint has async methods to auto-set mode for path, cls in mapping.items(): - if not (isinstance(cls, type) and issubclass(cls, RestEndpoint)): - raise ConfigurationError( - f"register() value for '{path}' must be a RestEndpoint subclass, " - f"got {cls!r}." - ) - - # Auto-detect mode from async methods - if self._mode == "sync": - for method_name in ( - "queryset", - "get", - "post", - "put", - "patch", - "delete", - ): - method = getattr(cls, method_name, None) - # Check if it's a coroutine function (skip non-callable like SQLAlchemy Select) - if ( - method is not None - and callable(method) - and asyncio.iscoroutinefunction(method) - ): - self._mode = "async" - break - - # Validate mode if explicitly set - if self._mode == "async": - try: - importlib.import_module("sqlalchemy.ext.asyncio") - from sqlalchemy.ext.asyncio import AsyncEngine + _require_rest_endpoint(path, cls) - if not isinstance(self._engine, AsyncEngine): - raise ConfigurationError( - f"mode='async' requires AsyncEngine, got {type(self._engine).__name__}" - ) - except ImportError: - raise ConfigurationError( - "mode='async' requires async dependencies. " - "Install: uv add 'lightapi[async]'" - ) + # An `async def` verb or queryset on any endpoint switches the app to async. + if self._mode == "sync" and any(map(_has_async_handlers, mapping.values())): + self._mode = "async" + if self._mode == "async": + _require_async_engine(self._engine) + app_context = AppContext( + self._middlewares, self._mode == "async", self._login_validator + ) for path, cls in mapping.items(): - # Inject session manager into endpoint class - cls._session_manager = self._session_manager - - cls._table_source.map(cls, self._session_manager) - - # Log registration for transparency - logger.info(f"Registering endpoint {path} -> {cls.__name__}") - logger.debug(f" SQLAlchemy metadata: {cls._meta}") - - # Warn if endpoint defines async queryset but engine is sync - if self._mode == "sync": - qs = cls.__dict__.get("queryset") - if qs is None: - qs = getattr(cls, "queryset", None) - if qs is not None and asyncio.iscoroutinefunction(qs): - warnings.warn( - f"'{cls.__name__}.queryset' is async but engine is sync; " - "sync path will be used.", - RuntimeWarning, - stacklevel=2, - ) - app_context = AppContext( - self._middlewares, self._mode == "async", self._login_validator - ) - collection = EndpointHandler.for_collection(cls, app_context) - detail = EndpointHandler.for_detail(cls, app_context) - self._routes.append( - Route( - path, - endpoint=collection.handle, - methods=collection.methods, - name=f"{cls.__name__}_collection", - ) + self._register_endpoint(path, cls, app_context) + + def _register_endpoint(self, path: str, cls: type, app_context: AppContext) -> None: + cls._session_manager = self._session_manager + cls._table_source.map(cls, self._session_manager) + logger.info(f"Registering endpoint {path} -> {cls.__name__}") + logger.debug(f" SQLAlchemy metadata: {cls._meta}") + + collection = EndpointHandler.for_collection(cls, app_context) + detail = EndpointHandler.for_detail(cls, app_context) + self._routes.append( + Route( + path, + endpoint=collection.handle, + methods=collection.methods, + name=f"{cls.__name__}_collection", ) - self._routes.append( - Route( - path.rstrip("/") + "/{id:int}", - endpoint=detail.handle, - methods=detail.methods, - name=f"{cls.__name__}_detail", - ) + ) + self._routes.append( + Route( + path.rstrip("/") + "/{id:int}", + endpoint=detail.handle, + methods=detail.methods, + name=f"{cls.__name__}_detail", ) - self._endpoint_map[path] = cls - - self._register_login_routes() + ) + self._endpoint_map[path] = cls - def _register_login_routes(self) -> None: - """(Re)build /auth/login and /auth/token when an endpoint uses JWT or Basic.""" + def _login_routes(self) -> list[Route]: + """/auth/login and /auth/token, when an endpoint uses JWT or Basic auth.""" backend = self._login_backend() if backend is None: - return + return [] rate_limiter = self._rate_limiter_global or RateLimiter(**_LOGIN_RATE_LIMITS) login = LoginEndpoint(backend, self._login_validator, rate_limiter) auth_path = self._auth_path.rstrip("/") - login_paths = (f"{auth_path}/login", f"{auth_path}/token") - - self._routes = [ - route - for route in self._routes - if not (isinstance(route, Route) and route.path in login_paths) + return [ + Route(f"{auth_path}/{name}", login.handle, methods=["POST"]) + for name in ("login", "token") ] - for position, path in enumerate(login_paths): - self._routes.insert(position, Route(path, login.handle, methods=["POST"])) def _login_backend(self) -> BaseAuthentication | None: """The backend that answers the login routes: a JWT one if any, else Basic.""" @@ -267,30 +173,15 @@ def run( """Create tables, build the Starlette ASGI app and start uvicorn.""" if self._mode == "async": _validate_async_dependencies(self._engine) - self._create_tables() - self._check_cache_connections() - - on_startup = [self._create_tables] if self._mode == "async" else [] - app = Starlette(debug=debug, routes=self._asgi_routes(), on_startup=on_startup) - - if self._cors_origins: - app.add_middleware( - StarletteCORSMiddleware, - allow_origins=self._cors_origins, - allow_credentials=True, - allow_methods=["*"], - allow_headers=["*"], - ) - uvicorn.run( - app, + self.build_app(debug=debug), host=host, port=port, log_level="debug" if debug else "info", reload=reload, ) - def build_app(self) -> Starlette: + def build_app(self, *, debug: bool = False) -> Starlette: """Build and return the Starlette ASGI app without starting the server. Useful for testing with ``httpx.AsyncClient`` or ``starlette.testclient.TestClient``. @@ -300,7 +191,7 @@ def build_app(self) -> Starlette: self._create_tables() self._check_cache_connections() on_startup = [self._create_tables] if self._mode == "async" else [] - app = Starlette(routes=self._asgi_routes(), on_startup=on_startup) + app = Starlette(debug=debug, routes=self._asgi_routes(), on_startup=on_startup) if self._cors_origins: app.add_middleware( StarletteCORSMiddleware, @@ -312,8 +203,9 @@ def build_app(self) -> Starlette: return app def _asgi_routes(self) -> list[Route]: - """Registered endpoint routes plus the always-on ``/healthz`` probe.""" + """Login routes first, then the endpoints, then the always-on ``/healthz``.""" return [ + *self._login_routes(), *self._routes, Route(HEALTH_PATH, HealthCheckEndpoint), ] @@ -418,6 +310,69 @@ def _check_cache_connections(self) -> None: break +_ASYNC_HANDLER_NAMES = ("queryset", "get", "post", "put", "patch", "delete") + + +def _engine_from_url(database_url: str | None) -> Any: + url = database_url or os.environ.get("LIGHTAPI_DATABASE_URL") + if url is None: + raise ConfigurationError( + "No database configured. Provide engine=..., database_url=..., or set " + "LIGHTAPI_DATABASE_URL environment variable." + ) + return create_engine(url) + + +def _checked_mode(mode: str | None) -> str: + """'sync' unless told otherwise; register() may still switch it to 'async'.""" + if mode is None: + return "sync" + if mode not in ("sync", "async"): + raise ConfigurationError(f"mode must be 'sync' or 'async', got '{mode}'") + return mode + + +def _as_rate_limiter(value: RateLimiter | dict[str, int] | None) -> RateLimiter | None: + if isinstance(value, dict): + return RateLimiter( + **{ + limit: value.get(limit, default) + for limit, default in _LOGIN_RATE_LIMITS.items() + } + ) + return value if isinstance(value, RateLimiter) else None + + +def _require_rest_endpoint(path: str, cls: Any) -> None: + if not (isinstance(cls, type) and issubclass(cls, RestEndpoint)): + raise ConfigurationError( + f"register() value for '{path}' must be a RestEndpoint subclass, " + f"got {cls!r}." + ) + + +def _has_async_handlers(cls: type) -> bool: + # A queryset may be a Select, which is not callable, so check before asking. + return any( + callable(handler) and asyncio.iscoroutinefunction(handler) + for handler in (getattr(cls, name, None) for name in _ASYNC_HANDLER_NAMES) + ) + + +def _require_async_engine(engine: Any) -> None: + try: + from sqlalchemy.ext.asyncio import AsyncEngine + except ImportError: + raise ConfigurationError( + "mode='async' requires async dependencies. " + "Install: uv add 'lightapi[async]'" + ) + if not isinstance(engine, AsyncEngine): + raise ConfigurationError( + f"mode='async' requires AsyncEngine, got {type(engine).__name__}" + ) + + def _validate_async_dependencies(engine: Any) -> None: """Raise ConfigurationError if async SQLAlchemy extras or dialect driver are missing.""" try: From a7be4881c85d1306846618f396c4f0ca2df5c85c Mon Sep 17 00:00:00 2001 From: iklo Date: Fri, 18 Sep 2026 11:21:30 -0300 Subject: [PATCH 3/5] refactor: move row persistence out of RestEndpoint into Repository _update was 100 lines that validated input, built a pydantic model, inspected the mapper, ran three SQL statements and chose between four HTTP responses. _retrieve, _create and _destroy mixed SQL and HTTP the same way. Repository(model, session) now owns storage: get, add, update, delete, the created_at/updated_at timestamps and the optimistic lock on `version`. It knows nothing about HTTP and reports problems as RowNotFound and VersionConflict. RestEndpoint validates, calls it, and turns those two errors into 404 and 409 through one table, so the status and wording of each error exist once. The PATCH schema (every field optional) was rebuilt with pydantic.create_model on every PATCH request. patch_schema() builds it once per create schema and caches it. Public CRUD methods and the _*_async helpers are untouched; they still call _retrieve, _create, _update and _destroy. tests/test_repository.py covers the lock directly: a stale version raises and leaves the row alone, a current one bumps the version. It fails if the version check or the cache is removed. rest.py: 819 lines on dev, 505 now. Tests: 350 passed, 1 xfailed, 2 xpassed. --- lightapi/repository.py | 97 +++++++++++++++++++ lightapi/rest.py | 202 +++++++++++++-------------------------- lightapi/schema.py | 21 +++- tests/test_repository.py | 75 +++++++++++++++ 4 files changed, 256 insertions(+), 139 deletions(-) create mode 100644 lightapi/repository.py create mode 100644 tests/test_repository.py diff --git a/lightapi/repository.py b/lightapi/repository.py new file mode 100644 index 0000000..cf3bed5 --- /dev/null +++ b/lightapi/repository.py @@ -0,0 +1,97 @@ +"""Row persistence for one mapped endpoint class.""" + +from __future__ import annotations + +import datetime +from typing import Any, Protocol + +from sqlalchemy import delete, select, update +from sqlalchemy.orm import Session, class_mapper + +FIRST_VERSION = 1 + + +class RowNotFound(Exception): + """No row has the requested primary key.""" + + +class VersionConflict(Exception): + """The row exists, but at a different ``version`` than the caller sent.""" + + +class VersionedRow(Protocol): + """What every LightAPI table has: a primary key and a version counter.""" + + id: Any + version: Any + + def __init__(self, **columns: Any) -> None: ... + + +class Repository: + """Reads and writes rows through an open Session. + + Owns the two storage rules of a LightAPI table: the ``created_at`` / + ``updated_at`` timestamps and optimistic locking on ``version``. + """ + + def __init__(self, model: type[VersionedRow], session: Session) -> None: + self._model = model + self._session = session + + @property + def nullable_columns(self) -> set[str]: + return { + attr.key + for attr in class_mapper(self._model).column_attrs + if any(column.nullable for column in attr.columns) + } + + def get(self, pk: int) -> Any: + model = self._model + row = ( + self._session.execute(select(model).where(model.id == pk)).scalars().first() + ) + if row is None: + raise RowNotFound(pk) + return row + + def add(self, values: dict[str, Any]) -> Any: + now = _utcnow() + row = self._model( + **values, created_at=now, updated_at=now, version=FIRST_VERSION + ) + self._session.add(row) + self._session.flush() # executes INSERT, populates auto-increment id + self._session.refresh(row) # re-loads DB-generated columns + return row + + def update(self, pk: int, expected_version: int, values: dict[str, Any]) -> Any: + """Write ``values`` only if the row is still at ``expected_version``.""" + model = self._model + result = self._session.execute( + update(model) + .where(model.id == pk, model.version == expected_version) + .values(**values, version=expected_version + 1, updated_at=_utcnow()) + ) + if result.rowcount == 0: + exists = self._session.execute( + select(model.id).where(model.id == pk) + ).first() + self._session.rollback() + raise VersionConflict(pk) if exists else RowNotFound(pk) + # Re-fetch so every column, updated_at and version included, is current. + return self.get(pk) + + def delete(self, pk: int) -> None: + model = self._model + deleted = self._session.execute( + delete(model).where(model.id == pk).returning(model.id) + ).first() + if deleted is None: + raise RowNotFound(pk) + + +def _utcnow() -> datetime.datetime: + """Naive UTC, which is what the DateTime columns store.""" + return datetime.datetime.now(datetime.timezone.utc).replace(tzinfo=None) diff --git a/lightapi/rest.py b/lightapi/rest.py index 069f193..6579df3 100644 --- a/lightapi/rest.py +++ b/lightapi/rest.py @@ -14,6 +14,7 @@ from sqlalchemy.orm import Session from starlette.background import BackgroundTasks +from pydantic import ValidationError from sqlalchemy import ( Boolean, Column, @@ -24,8 +25,6 @@ Numeric, String, Uuid, - delete, - update, ) from sqlalchemy import select as sa_select from starlette.requests import Request @@ -38,11 +37,13 @@ ) from lightapi.exceptions import ConfigurationError from lightapi.pagination import NoPagination +from lightapi.repository import Repository, RowNotFound, VersionConflict from lightapi.schema import ( SchemaFactory, _apply_fields, _row_to_dict, normalise_serializer, + patch_schema, resolve_fields, ) from lightapi.session import get_async_session, get_sync_session @@ -245,6 +246,23 @@ def _allowed_methods(cls: type) -> set[str]: return allowed or set(_ALL_METHODS) +_ERROR_RESPONSES: dict[type[Exception], tuple[HTTPStatus, str]] = { + RowNotFound: (HTTPStatus.NOT_FOUND, "not found"), + VersionConflict: (HTTPStatus.CONFLICT, "version conflict"), +} + + +def _error_response(error: Exception) -> Response: + status, detail = _ERROR_RESPONSES[type(error)] + return JSONResponse({RESPONSE_KEY_DETAIL: detail}, status_code=status) + + +def _unprocessable(errors: Any) -> Response: + return JSONResponse( + {RESPONSE_KEY_DETAIL: errors}, status_code=HTTPStatus.UNPROCESSABLE_ENTITY + ) + + class RestEndpoint(metaclass=RestEndpointMeta): """Base class for all LightAPI endpoints. @@ -338,164 +356,72 @@ def _list(self, session: Session, request: Request, qs: Any) -> Response: return JSONResponse(paginator.render(request, qs, session, serialize)) def _retrieve(self, session: Session, pk: int) -> Response: - cls = type(self) - instance = ( - session.execute( - sa_select(cls._model_class).where(cls._model_class.id == pk) - ) - .scalars() - .first() - ) - if instance is None: - return JSONResponse( - {RESPONSE_KEY_DETAIL: "not found"}, status_code=HTTPStatus.NOT_FOUND - ) - return JSONResponse(self._serialize_row(instance, "GET")) + try: + row = self._repository(session).get(pk) + except RowNotFound as exc: + return _error_response(exc) + return JSONResponse(self._serialize_row(row, "GET")) def _create(self, session: Session, data: dict[str, Any]) -> Response: - from pydantic import ValidationError - - cls = type(self) try: - validated = cls.__schema_create__.model_validate(data) + validated = type(self).__schema_create__.model_validate(data) except ValidationError as exc: - return JSONResponse( - {RESPONSE_KEY_DETAIL: exc.errors()}, - status_code=HTTPStatus.UNPROCESSABLE_ENTITY, - ) + return _unprocessable(exc.errors()) - now = datetime.datetime.now(datetime.timezone.utc).replace(tzinfo=None) - instance = cls._model_class( - **validated.model_dump(), - created_at=now, - updated_at=now, - version=1, - ) - session.add(instance) - session.flush() # executes INSERT, populates auto-increment id - session.refresh(instance) # re-loads DB-generated columns + row = self._repository(session).add(validated.model_dump()) return JSONResponse( - self._serialize_row(instance, "POST"), status_code=HTTPStatus.CREATED + self._serialize_row(row, "POST"), status_code=HTTPStatus.CREATED ) def _update( self, session: Session, data: dict[str, Any], pk: int, partial: bool ) -> Response: - """PUT/PATCH with optimistic locking on the ``version`` column.""" - from pydantic import ValidationError - + """PUT/PATCH; the client must send the ``version`` it last read.""" client_version = data.get("version") if client_version is None: - return JSONResponse( - { - RESPONSE_KEY_DETAIL: [ - {"loc": ["version"], "msg": "Field required", "type": "missing"} - ] - }, - status_code=HTTPStatus.UNPROCESSABLE_ENTITY, + return _unprocessable( + [{"loc": ["version"], "msg": "Field required", "type": "missing"}] ) - cls = type(self) - + repository = self._repository(session) try: - if partial: - # Build a one-shot model where every field is Optional so that - # PATCH can supply any subset of fields without validation errors. - from typing import Optional as _Opt - - from pydantic import ConfigDict as _CD - from pydantic import create_model as _cm - - patch_fields: dict[str, Any] = {} - for fname, finfo in cls.__schema_create__.model_fields.items(): - ann = finfo.annotation - patch_fields[fname] = (_Opt[ann], None) # type: ignore[valid-type] - PatchSchema = _cm( - f"{cls.__name__}PatchSchema", - __config__=_CD(from_attributes=True), - **patch_fields, - ) - validated = PatchSchema.model_validate(data) - # Determine which columns are nullable so explicit null values - # can clear Optional fields (non-nullable fields still skip None). - from sqlalchemy import inspect as _sa_inspect - - nullable_cols: set[str] = { - attr.key - for attr in _sa_inspect(cls._model_class).mapper.column_attrs - if any(c.nullable for c in attr.columns) - } - update_data = { - k: v - for k, v in validated.model_dump(exclude_unset=True).items() - if k not in _AUTO_FIELDS and (v is not None or k in nullable_cols) - } - else: - validated = cls.__schema_create__.model_validate(data) - update_data = { - k: v - for k, v in validated.model_dump().items() - if k not in _AUTO_FIELDS - } + values = self._update_values(data, partial, repository.nullable_columns) except ValidationError as exc: - return JSONResponse( - {RESPONSE_KEY_DETAIL: exc.errors()}, - status_code=HTTPStatus.UNPROCESSABLE_ENTITY, - ) - - update_data.pop("version", None) + return _unprocessable(exc.errors()) - result = session.execute( - update(cls._model_class) - .where( - cls._model_class.id == pk, - cls._model_class.version == client_version, - ) - .values( - **update_data, - version=client_version + 1, - updated_at=datetime.datetime.now(datetime.timezone.utc).replace( - tzinfo=None - ), - ) - ) - if result.rowcount == 0: - exists = session.execute( - sa_select(cls._model_class.id).where(cls._model_class.id == pk) - ).first() - session.rollback() - if not exists: - return JSONResponse( - {RESPONSE_KEY_DETAIL: "not found"}, - status_code=HTTPStatus.NOT_FOUND, - ) - return JSONResponse( - {RESPONSE_KEY_DETAIL: "version conflict"}, - status_code=HTTPStatus.CONFLICT, - ) - # Re-fetch so all columns (including updated_at/version) are current - instance = ( - session.execute( - sa_select(cls._model_class).where(cls._model_class.id == pk) - ) - .scalars() - .first() - ) - return JSONResponse(self._serialize_row(instance, "PUT")) + try: + row = repository.update(pk, client_version, values) + except (RowNotFound, VersionConflict) as exc: + return _error_response(exc) + return JSONResponse(self._serialize_row(row, "PUT")) + + def _update_values( + self, data: dict[str, Any], partial: bool, nullable_columns: set[str] + ) -> dict[str, Any]: + """Validated column values for an update, without the auto-managed fields.""" + schema_create = type(self).__schema_create__ + if not partial: + values = schema_create.model_validate(data).model_dump() + return {k: v for k, v in values.items() if k not in _AUTO_FIELDS} + + sent = patch_schema(schema_create).model_validate(data) + # An explicit null clears a nullable column; on any other column it is ignored. + return { + k: v + for k, v in sent.model_dump(exclude_unset=True).items() + if k not in _AUTO_FIELDS and (v is not None or k in nullable_columns) + } def _destroy(self, session: Session, pk: int) -> Response: - cls = type(self) - stmt = ( - delete(cls._model_class) - .where(cls._model_class.id == pk) - .returning(cls._model_class.id) - ) - if session.execute(stmt).first() is None: - return JSONResponse( - {RESPONSE_KEY_DETAIL: "not found"}, status_code=HTTPStatus.NOT_FOUND - ) + try: + self._repository(session).delete(pk) + except RowNotFound as exc: + return _error_response(exc) return Response(status_code=HTTPStatus.NO_CONTENT) + def _repository(self, session: Session) -> Repository: + return Repository(type(self)._model_class, session) + # ── Sync CRUD: the core inside a sync Session ───────────────────────────── def list(self, request: Request) -> Response: diff --git a/lightapi/schema.py b/lightapi/schema.py index 6633961..aec0c44 100644 --- a/lightapi/schema.py +++ b/lightapi/schema.py @@ -3,10 +3,11 @@ import datetime import logging from decimal import Decimal +from functools import lru_cache from typing import Any, Optional from uuid import UUID -from pydantic import ConfigDict, create_model +from pydantic import BaseModel, ConfigDict, create_model from pydantic.fields import FieldInfo from lightapi.constants import AUTO_FIELDS @@ -84,6 +85,24 @@ def _apply_fields(d: dict[str, Any], fields: list[str] | None) -> dict[str, Any] return {k: v for k, v in d.items() if k in fields} +@lru_cache(maxsize=None) +def patch_schema(schema_create: type[BaseModel]) -> type[BaseModel]: + """``schema_create`` with every field optional, so PATCH can send any subset. + + Cached per create schema: endpoints are a fixed set of classes, and building a + pydantic model on every PATCH request was measurable work for nothing. + """ + optional_fields: dict[str, Any] = { + name: (Optional[field.annotation], None) # type: ignore[valid-type] + for name, field in schema_create.model_fields.items() + } + return create_model( + schema_create.__name__.replace("CreateSchema", "PatchSchema"), + __config__=ConfigDict(from_attributes=True), + **optional_fields, + ) + + # Python types a reflected column is validated as; any other type stays ``Any``. _REFLECTED_PYTHON_TYPES = frozenset( { diff --git a/tests/test_repository.py b/tests/test_repository.py new file mode 100644 index 0000000..d7ba91f --- /dev/null +++ b/tests/test_repository.py @@ -0,0 +1,75 @@ +"""Repository owns optimistic locking; patch_schema is built once per endpoint.""" + +import pytest +from sqlalchemy import create_engine +from sqlalchemy.pool import StaticPool + +from lightapi import LightApi, RestEndpoint, get_sync_session +from lightapi.fields import Field as LField +from lightapi.repository import Repository, RowNotFound, VersionConflict +from lightapi.schema import patch_schema + +MISSING_PK = 9999 + + +class LedgerEntry(RestEndpoint): + memo: str = LField(min_length=1) + + +@pytest.fixture +def engine(): + engine = create_engine( + "sqlite:///:memory:", + connect_args={"check_same_thread": False}, + poolclass=StaticPool, + ) + app = LightApi(engine=engine) + app.register({"/ledgerentries": LedgerEntry}) + app.build_app() + return engine + + +def _added_pk(engine) -> int: + with get_sync_session(engine) as session: + return Repository(LedgerEntry, session).add({"memo": "opening"}).id + + +def test_update_at_current_version_writes_and_bumps_version(engine): + pk = _added_pk(engine) + + with get_sync_session(engine) as session: + row = Repository(LedgerEntry, session).update(pk, 1, {"memo": "changed"}) + + assert row.memo == "changed" + assert row.version == 2 + + +def test_update_at_stale_version_raises_conflict_and_keeps_the_row(engine): + pk = _added_pk(engine) + + with get_sync_session(engine) as session: + repository = Repository(LedgerEntry, session) + with pytest.raises(VersionConflict): + repository.update(pk, 7, {"memo": "lost update"}) + + assert repository.get(pk).memo == "opening" + + +def test_update_of_missing_row_raises_not_found(engine): + with get_sync_session(engine) as session: + with pytest.raises(RowNotFound): + Repository(LedgerEntry, session).update(MISSING_PK, 1, {"memo": "x"}) + + +def test_delete_of_missing_row_raises_not_found(engine): + with get_sync_session(engine) as session: + with pytest.raises(RowNotFound): + Repository(LedgerEntry, session).delete(MISSING_PK) + + +def test_patch_schema_is_built_once_per_create_schema(): + first = patch_schema(LedgerEntry.__schema_create__) + second = patch_schema(LedgerEntry.__schema_create__) + + assert first is second + assert first.model_validate({}).model_dump(exclude_unset=True) == {} From 831330df3fe5efe7bdf39c50e8652262b1f9ee54 Mon Sep 17 00:00:00 2001 From: iklo Date: Fri, 18 Sep 2026 11:26:01 -0300 Subject: [PATCH 4/5] refactor: split the YAML loader by job and merge auth settings once yaml_loader.py was 705 lines doing three jobs: resolving class names, describing the YAML document as pydantic models, and turning a validated document into endpoint classes. - yaml_names.py: the built-in name registry and dotted-path resolution. - yaml_schema.py: the pydantic models and the field type map. - yaml_loader.py: the translation and load_config (358 lines). The rule "endpoint settings win, defaults fill the gaps" was written out field by field twice, once in _make_authentication and again inside the per-method branch of _build_meta_class. AuthConfig.merged_over(defaults) states it once, on the object that owns the fields. _build_meta_class was 123 lines; its per-method permission logic is now _per_method_authentication and _permission_name. The branch that handled a default permission that was neither str nor dict is gone: the pydantic model only allows those two. The one test change is an import: _resolve_name is resolve_name in yaml_names. The 491-line test_yaml_config.py is otherwise untouched. Tests: 350 passed, 1 xfailed, 2 xpassed. --- lightapi/yaml_loader.py | 485 ++++++-------------------------------- lightapi/yaml_names.py | 120 ++++++++++ lightapi/yaml_schema.py | 186 +++++++++++++++ tests/test_yaml_config.py | 11 +- 4 files changed, 381 insertions(+), 421 deletions(-) create mode 100644 lightapi/yaml_names.py create mode 100644 lightapi/yaml_schema.py diff --git a/lightapi/yaml_loader.py b/lightapi/yaml_loader.py index 4041c53..52e0af4 100644 --- a/lightapi/yaml_loader.py +++ b/lightapi/yaml_loader.py @@ -27,363 +27,87 @@ from __future__ import annotations -import datetime -import importlib -import os -from decimal import Decimal -from typing import Any, Union - -from pydantic import BaseModel, field_validator, model_validator +from typing import Any from lightapi.exceptions import ConfigurationError +from lightapi.yaml_names import name_registry, resolve_callable, resolve_name +from lightapi.yaml_schema import ( + YAML_FIELD_TYPES, + AuthConfig, + DefaultsConfig, + EndpointConfig, + FilteringConfig, + LightAPIConfig, + MetaConfig, + PaginationConfig, +) -# ───────────────────────────────────────────────────────────────────────────── -# String → class registry (everything a YAML author would reference by name) -# ───────────────────────────────────────────────────────────────────────────── - - -def _build_name_registry() -> dict[str, type]: - from lightapi.auth import ( - AllowAny, - BasicAuthentication, - IsAdminUser, - IsAuthenticated, - JWTAuthentication, - ) - from lightapi.core import AuthenticationMiddleware, CORSMiddleware, Middleware - from lightapi.filters import FieldFilter, OrderingFilter, SearchFilter - from lightapi.methods import HttpMethod - - return { - # Auth backends - "JWTAuthentication": JWTAuthentication, - "BasicAuthentication": BasicAuthentication, - # Permissions - "AllowAny": AllowAny, - "IsAuthenticated": IsAuthenticated, - "IsAdminUser": IsAdminUser, - # Filter backends - "FieldFilter": FieldFilter, - "SearchFilter": SearchFilter, - "OrderingFilter": OrderingFilter, - # Middleware - "Middleware": Middleware, - "CORSMiddleware": CORSMiddleware, - "AuthenticationMiddleware": AuthenticationMiddleware, - # HttpMethod mixins (for bases resolution) - "GET": HttpMethod.GET, - "POST": HttpMethod.POST, - "PUT": HttpMethod.PUT, - "PATCH": HttpMethod.PATCH, - "DELETE": HttpMethod.DELETE, - } - - -def _resolve_callable(dotted_path: str) -> Any: - """Resolve a dotted path like 'myapp.validators.validate_login' to a callable.""" - if "." not in dotted_path: - raise ConfigurationError( - f"login_validator must be a dotted path (e.g. myapp.validators.check), " - f"got '{dotted_path}'" - ) - module_path, attr_name = dotted_path.rsplit(".", 1) - try: - mod = importlib.import_module(module_path) - fn = getattr(mod, attr_name) - except (ImportError, AttributeError) as exc: - raise ConfigurationError( - f"Cannot resolve login_validator '{dotted_path}': {exc}" - ) from exc - if not callable(fn): - raise ConfigurationError(f"login_validator '{dotted_path}' is not callable.") - - # Validate signature: must be sync function with exactly 2 positional args - import inspect - - if inspect.iscoroutinefunction(fn): - raise ConfigurationError( - f"login_validator '{dotted_path}' is async, but sync function required. " - f"Login validation must be a synchronous function." - ) - - try: - sig = inspect.signature(fn) - except ValueError: - # Some callables (e.g., builtins) don't have inspectable signatures - return fn - - # Count required positional parameters - required_params = 0 - for param in sig.parameters.values(): - if param.kind in ( - inspect.Parameter.POSITIONAL_ONLY, - inspect.Parameter.POSITIONAL_OR_KEYWORD, - ): - if param.default is inspect.Parameter.empty: - required_params += 1 - - if required_params != 2: - raise ConfigurationError( - f"login_validator '{dotted_path}' must accept exactly 2 required " - f"positional parameters (username, password), got {required_params}" - ) - - return fn - - -def _resolve_name(name: str) -> type: - """Resolve a class name string to a class. - - Tries the built-in registry first, then falls back to dotted import path - (e.g. 'myapp.middleware.RequestIdMiddleware'). - """ - registry = _build_name_registry() - if name in registry: - return registry[name] - # Dotted path fallback - if "." in name: - module_path, class_name = name.rsplit(".", 1) - try: - mod = importlib.import_module(module_path) - return getattr(mod, class_name) - except (ImportError, AttributeError) as exc: - raise ConfigurationError(f"Cannot resolve '{name}': {exc}") from exc - raise ConfigurationError( - f"Unknown class name '{name}'. " - "Use a fully dotted import path for custom classes." - ) - - -# ───────────────────────────────────────────────────────────────────────────── -# Type map for declarative field definitions -# ───────────────────────────────────────────────────────────────────────────── - -_YAML_TYPE_MAP: dict[str, type] = { - "str": str, - "int": int, - "float": float, - "bool": bool, - "datetime": datetime.datetime, - "Decimal": Decimal, - "decimal": Decimal, -} - -# ───────────────────────────────────────────────────────────────────────────── -# Pydantic v2 models — the YAML schema -# ───────────────────────────────────────────────────────────────────────────── - - -class DatabaseConfig(BaseModel): - """Nested database block: database: { url: ... }""" - - url: str - - @field_validator("url", mode="before") - @classmethod - def substitute_env(cls, v: str) -> str: - return _substitute_env(v) - - -class AuthLoginConfig(BaseModel): - """Login/auth block: auth: { auth_path: ..., login_validator: ... }. - - When using JWTAuthentication or BasicAuthentication, login_validator is required. - It can be specified as a dotted path (e.g. myapp.validators.validate_login) - or passed as an override to from_config(login_validator=...). - """ - - auth_path: str = "/auth" - login_validator: str | None = None - - -class AuthConfig(BaseModel): - """Authentication block used in defaults and per-endpoint meta.""" - - backend: str | None = None - permission: Union[str, dict[str, str], None] = None - jwt_expiration: int | None = None - jwt_extra_claims: list[str] | None = None - jwt_algorithm: str | None = None - - -class FilteringConfig(BaseModel): - """Filtering block inside meta.""" - backends: list[str] = [] - fields: list[str] = [] - search: list[str] = [] - ordering: list[str] = [] - - -class PaginationConfig(BaseModel): - """Pagination block used in defaults and per-endpoint meta.""" - - style: str = "page_number" - page_size: int = 20 - - -class DefaultsConfig(BaseModel): - """Global defaults applied to all endpoints unless overridden.""" - - authentication: AuthConfig | None = None - pagination: PaginationConfig | None = None - - -class MethodAuthConfig(BaseModel): - """Per-method authentication override inside meta.methods dict.""" - - authentication: AuthConfig | None = None - - -class CacheConfig(BaseModel): - """Cache block inside meta: cache: { ttl: 60 }""" - - ttl: int = 60 +def _make_authentication( + auth_cfg: AuthConfig | None, + defaults_auth: AuthConfig | None, +) -> Any: + """Endpoint settings win; the defaults fill whatever the endpoint left unset.""" + settings = (auth_cfg or AuthConfig()).merged_over(defaults_auth) + if settings.backend is None and settings.permission is None: + return None + return _authentication_from(settings, _resolved_permission(settings.permission)) -class SerializerConfig(BaseModel): - """Serializer block inside meta. +def _per_method_authentication(meta: MetaConfig, defaults: DefaultsConfig) -> Any: + """Authentication for ``methods: {GET: {...}, DELETE: {...}}``. - Use ``fields`` for a unified list, or ``read``/``write`` for per-verb lists. + Each method takes its own permission, else the endpoint's, else the default. """ + default_auth = defaults.authentication + default_permission = default_auth.permission if default_auth else None - fields: list[str] | None = None - read: list[str] | None = None - write: list[str] | None = None - - -class MetaConfig(BaseModel): - """meta: block inside a declarative endpoint entry.""" - - # methods can be a list ["GET", "POST"] or a dict {GET: {...}, DELETE: {...}} - methods: Union[list[str], dict[str, MethodAuthConfig | None]] = [] - authentication: AuthConfig | None = None - filtering: FilteringConfig | None = None - pagination: PaginationConfig | None = None - cache: CacheConfig | None = None - serializer: SerializerConfig | None = None - table: str | None = None # custom table name (required when reflect: true) - - -class FieldSpec(BaseModel): - """Single field definition inside fields:.""" - - type: str - optional: bool = False - # All remaining keys forwarded to Field() as pydantic constraints - model_config = {"extra": "allow"} - - @field_validator("type") - @classmethod - def type_must_be_known(cls, v: str) -> str: - if v not in _YAML_TYPE_MAP: - raise ValueError( - f"Unknown field type '{v}'. Valid types: {sorted(_YAML_TYPE_MAP)}" - ) - return v - - -class EndpointConfig(BaseModel): - """A single endpoint entry.""" - - route: str - fields: dict[str, FieldSpec] = {} - reflect: bool = False - meta: MetaConfig = MetaConfig() - - @model_validator(mode="after") - def require_route(self) -> "EndpointConfig": - if not self.route: - raise ValueError("Each endpoint must have a 'route'.") - return self - - @property - def effective_route(self) -> str: - return self.route.strip() - - -class LightAPIConfig(BaseModel): - """Root YAML document schema.""" - - database: DatabaseConfig | None = None - cors_origins: list[str] = [] - defaults: DefaultsConfig = DefaultsConfig() - endpoints: list[EndpointConfig] = [] - middleware: list[str] = [] - auth: AuthLoginConfig | None = None - mode: str | None = None # "sync" | "async" — auto-detected when omitted - - @property - def effective_database_url(self) -> str | None: - return self.database.url if self.database else None - - -# ───────────────────────────────────────────────────────────────────────────── -# Translation: validated Pydantic model → LightAPI objects -# ───────────────────────────────────────────────────────────────────────────── - + permissions: dict[str, type] = {} + for method, method_cfg in meta.methods.items(): + method_auth = (method_cfg.authentication if method_cfg else None) or ( + meta.authentication + ) + name = _permission_name(method, method_auth, default_permission) + if name: + permissions[method] = resolve_name(name) -def _substitute_env(value: str) -> str: - """Replace ${VAR} with the environment variable value.""" - if value.startswith("${") and value.endswith("}"): - var = value[2:-1] - resolved = os.environ.get(var) - if not resolved: - raise ConfigurationError( - f"Environment variable '{var}' is not set (required by lightapi.yaml)." - ) - return resolved - return value + if not permissions: + return None + settings = (meta.authentication or AuthConfig()).merged_over(default_auth) + return _authentication_from(settings, permissions) -def _make_authentication( - auth_cfg: AuthConfig | None, - defaults_auth: AuthConfig | None, -) -> Any: - # Merge: explicit cfg wins over defaults, defaults fill gaps - merged_backend = None - merged_permission = None - merged_jwt_expiration = None - merged_jwt_extra_claims = None - merged_jwt_algorithm = None - - for source in (defaults_auth, auth_cfg): - if source is None: - continue - if source.backend is not None: - merged_backend = source.backend - if source.permission is not None: - merged_permission = source.permission - if source.jwt_expiration is not None: - merged_jwt_expiration = source.jwt_expiration - if source.jwt_extra_claims is not None: - merged_jwt_extra_claims = source.jwt_extra_claims - if source.jwt_algorithm is not None: - merged_jwt_algorithm = source.jwt_algorithm - - if merged_backend is None and merged_permission is None: - return None +def _permission_name( + method: str, + method_auth: AuthConfig | None, + default_permission: str | dict[str, str] | None, +) -> str | None: + own = method_auth.permission if method_auth else None + if isinstance(own, str): + return own + if isinstance(default_permission, dict): + return default_permission.get(method) + return default_permission - backend_cls = _resolve_name(merged_backend) if merged_backend else None - if isinstance(merged_permission, dict): +def _resolved_permission(permission: str | dict[str, str] | None) -> Any: + if isinstance(permission, dict): # Per-method permission dict: {GET: IsAuthenticated, DELETE: IsAdminUser} - permission = { - method: _resolve_name(perm) for method, perm in merged_permission.items() - } - elif isinstance(merged_permission, str): - permission = _resolve_name(merged_permission) - else: - permission = None + return {method: resolve_name(name) for method, name in permission.items()} + if isinstance(permission, str): + return resolve_name(permission) + return None + +def _authentication_from(settings: AuthConfig, permission: Any) -> Any: from lightapi.config import Authentication return Authentication( - backend=backend_cls, + backend=resolve_name(settings.backend) if settings.backend else None, permission=permission, - jwt_expiration=merged_jwt_expiration, - jwt_extra_claims=merged_jwt_extra_claims, - jwt_algorithm=merged_jwt_algorithm, + jwt_expiration=settings.jwt_expiration, + jwt_extra_claims=settings.jwt_extra_claims, + jwt_algorithm=settings.jwt_algorithm, ) @@ -396,7 +120,7 @@ def _make_filtering(filtering_cfg: FilteringConfig | None) -> Any: # Auto-select backends based on which lists are populated backends: list[type] = list( - [_resolve_name(b) for b in filtering_cfg.backends] + [resolve_name(b) for b in filtering_cfg.backends] if filtering_cfg.backends else [] ) @@ -439,83 +163,12 @@ def _build_meta_class( attrs["table"] = meta.table return type("Meta", (), attrs) - # Authentication — handle per-method dict (methods as dict form) if isinstance(meta.methods, dict): - # Build per-method permission dict from the methods dict - permission_map: dict[str, type] = {} - method_auth_default = meta.authentication # endpoint-level auth override - - # Get default permission from defaults - default_permission = None - if defaults.authentication and defaults.authentication.permission: - default_permission = defaults.authentication.permission - - for method, method_cfg in meta.methods.items(): - cfg_auth = method_cfg.authentication if method_cfg else None - src_auth = cfg_auth or method_auth_default - - # Determine permission for this method - method_permission = None - - # First, check method-specific auth - if src_auth and src_auth.permission: - perm = src_auth.permission - if isinstance(perm, str): - method_permission = _resolve_name(perm) - # Note: per-method auth in YAML dict can't have dict permissions - # only string permissions are supported in this path - - # If no method-specific permission, check default permission - if method_permission is None and default_permission is not None: - if isinstance(default_permission, dict): - # Default permission is a dict: check for method-specific default - if method in default_permission: - perm = default_permission[method] - if isinstance(perm, str): - method_permission = _resolve_name(perm) - else: - # Default permission is a string or class - if isinstance(default_permission, str): - method_permission = _resolve_name(default_permission) - else: - method_permission = default_permission - - if method_permission is not None: - permission_map[method] = method_permission - - if permission_map: - # Merge auth settings similar to _make_authentication - merged_backend = None - merged_jwt_expiration = None - merged_jwt_extra_claims = None - merged_jwt_algorithm = None - - for source in (defaults.authentication, meta.authentication): - if source is None: - continue - if source.backend is not None: - merged_backend = source.backend - if source.jwt_expiration is not None: - merged_jwt_expiration = source.jwt_expiration - if source.jwt_extra_claims is not None: - merged_jwt_extra_claims = source.jwt_extra_claims - if source.jwt_algorithm is not None: - merged_jwt_algorithm = source.jwt_algorithm - - from lightapi.config import Authentication - - attrs["authentication"] = Authentication( - backend=_resolve_name(merged_backend) if merged_backend else None, - permission=permission_map, - jwt_expiration=merged_jwt_expiration, - jwt_extra_claims=merged_jwt_extra_claims, - jwt_algorithm=merged_jwt_algorithm, - ) + auth = _per_method_authentication(meta, defaults) else: - # Simple auth: endpoint overrides defaults auth = _make_authentication(meta.authentication, defaults.authentication) - if auth is not None: - attrs["authentication"] = auth + if auth is not None: + attrs["authentication"] = auth filtering = _make_filtering(meta.filtering) if filtering is not None: @@ -565,7 +218,7 @@ def _resolve_methods_bases(meta: MetaConfig) -> tuple[type, ...]: if not method_list: return (RestEndpoint,) - registry = _build_name_registry() + registry = name_registry() bases: list[type] = [RestEndpoint] for m in method_list: mixin = registry.get(m) @@ -590,7 +243,7 @@ def _build_endpoint_class(entry: EndpointConfig, defaults: DefaultsConfig) -> ty class_attrs: dict[str, Any] = {"__annotations__": annotations} for field_name, spec in entry.fields.items(): - py_type = _YAML_TYPE_MAP[spec.type] + py_type = YAML_FIELD_TYPES[spec.type] if spec.optional: py_type = Optional[py_type] # type: ignore[assignment] @@ -670,7 +323,7 @@ def load_config(app_cls: type, config_path: str, **overrides: Any) -> Any: ) from exc db_url = cfg.effective_database_url - middlewares: list[type] = [_resolve_name(name) for name in cfg.middleware] + middlewares: list[type] = [resolve_name(name) for name in cfg.middleware] constructor_kwargs: dict[str, Any] = { "database_url": db_url or None, @@ -685,7 +338,7 @@ def load_config(app_cls: type, config_path: str, **overrides: Any) -> Any: if cfg.auth: constructor_kwargs["auth_path"] = cfg.auth.auth_path if cfg.auth.login_validator: - constructor_kwargs["login_validator"] = _resolve_callable( + constructor_kwargs["login_validator"] = resolve_callable( cfg.auth.login_validator ) diff --git a/lightapi/yaml_names.py b/lightapi/yaml_names.py new file mode 100644 index 0000000..e8d9468 --- /dev/null +++ b/lightapi/yaml_names.py @@ -0,0 +1,120 @@ +"""Names a YAML config may use: built-ins by short name, the rest by dotted path.""" + +from __future__ import annotations + +import importlib +from typing import Any + +from lightapi.exceptions import ConfigurationError + + +def name_registry() -> dict[str, type]: + from lightapi.auth import ( + AllowAny, + BasicAuthentication, + IsAdminUser, + IsAuthenticated, + JWTAuthentication, + ) + from lightapi.core import AuthenticationMiddleware, CORSMiddleware, Middleware + from lightapi.filters import FieldFilter, OrderingFilter, SearchFilter + from lightapi.methods import HttpMethod + + return { + # Auth backends + "JWTAuthentication": JWTAuthentication, + "BasicAuthentication": BasicAuthentication, + # Permissions + "AllowAny": AllowAny, + "IsAuthenticated": IsAuthenticated, + "IsAdminUser": IsAdminUser, + # Filter backends + "FieldFilter": FieldFilter, + "SearchFilter": SearchFilter, + "OrderingFilter": OrderingFilter, + # Middleware + "Middleware": Middleware, + "CORSMiddleware": CORSMiddleware, + "AuthenticationMiddleware": AuthenticationMiddleware, + # HttpMethod mixins (for bases resolution) + "GET": HttpMethod.GET, + "POST": HttpMethod.POST, + "PUT": HttpMethod.PUT, + "PATCH": HttpMethod.PATCH, + "DELETE": HttpMethod.DELETE, + } + + +def resolve_callable(dotted_path: str) -> Any: + """Resolve a dotted path like 'myapp.validators.validate_login' to a callable.""" + if "." not in dotted_path: + raise ConfigurationError( + f"login_validator must be a dotted path (e.g. myapp.validators.check), " + f"got '{dotted_path}'" + ) + module_path, attr_name = dotted_path.rsplit(".", 1) + try: + mod = importlib.import_module(module_path) + fn = getattr(mod, attr_name) + except (ImportError, AttributeError) as exc: + raise ConfigurationError( + f"Cannot resolve login_validator '{dotted_path}': {exc}" + ) from exc + if not callable(fn): + raise ConfigurationError(f"login_validator '{dotted_path}' is not callable.") + + # Validate signature: must be sync function with exactly 2 positional args + import inspect + + if inspect.iscoroutinefunction(fn): + raise ConfigurationError( + f"login_validator '{dotted_path}' is async, but sync function required. " + f"Login validation must be a synchronous function." + ) + + try: + sig = inspect.signature(fn) + except ValueError: + # Some callables (e.g., builtins) don't have inspectable signatures + return fn + + # Count required positional parameters + required_params = 0 + for param in sig.parameters.values(): + if param.kind in ( + inspect.Parameter.POSITIONAL_ONLY, + inspect.Parameter.POSITIONAL_OR_KEYWORD, + ): + if param.default is inspect.Parameter.empty: + required_params += 1 + + if required_params != 2: + raise ConfigurationError( + f"login_validator '{dotted_path}' must accept exactly 2 required " + f"positional parameters (username, password), got {required_params}" + ) + + return fn + + +def resolve_name(name: str) -> Any: + """Resolve a class name string to a class. + + Tries the built-in registry first, then falls back to dotted import path + (e.g. 'myapp.middleware.RequestIdMiddleware'). + """ + registry = name_registry() + if name in registry: + return registry[name] + # Dotted path fallback + if "." in name: + module_path, class_name = name.rsplit(".", 1) + try: + mod = importlib.import_module(module_path) + return getattr(mod, class_name) + except (ImportError, AttributeError) as exc: + raise ConfigurationError(f"Cannot resolve '{name}': {exc}") from exc + raise ConfigurationError( + f"Unknown class name '{name}'. " + "Use a fully dotted import path for custom classes." + ) diff --git a/lightapi/yaml_schema.py b/lightapi/yaml_schema.py new file mode 100644 index 0000000..68bafe0 --- /dev/null +++ b/lightapi/yaml_schema.py @@ -0,0 +1,186 @@ +"""Pydantic models of the declarative YAML document (see yaml_loader for the format).""" + +from __future__ import annotations + +import datetime +import os +from decimal import Decimal +from typing import Union + +from pydantic import BaseModel, field_validator, model_validator + +from lightapi.exceptions import ConfigurationError + +YAML_FIELD_TYPES: dict[str, type] = { + "str": str, + "int": int, + "float": float, + "bool": bool, + "datetime": datetime.datetime, + "Decimal": Decimal, + "decimal": Decimal, +} + + +def _substitute_env(value: str) -> str: + """Replace ${VAR} with the environment variable value.""" + if value.startswith("${") and value.endswith("}"): + var = value[2:-1] + resolved = os.environ.get(var) + if not resolved: + raise ConfigurationError( + f"Environment variable '{var}' is not set (required by lightapi.yaml)." + ) + return resolved + return value + + +class DatabaseConfig(BaseModel): + """Nested database block: database: { url: ... }""" + + url: str + + @field_validator("url", mode="before") + @classmethod + def substitute_env(cls, v: str) -> str: + return _substitute_env(v) + + +class AuthLoginConfig(BaseModel): + """Login/auth block: auth: { auth_path: ..., login_validator: ... }. + + When using JWTAuthentication or BasicAuthentication, login_validator is required. + It can be specified as a dotted path (e.g. myapp.validators.validate_login) + or passed as an override to from_config(login_validator=...). + """ + + auth_path: str = "/auth" + login_validator: str | None = None + + +class AuthConfig(BaseModel): + """Authentication block used in defaults and per-endpoint meta.""" + + backend: str | None = None + permission: Union[str, dict[str, str], None] = None + jwt_expiration: int | None = None + jwt_extra_claims: list[str] | None = None + jwt_algorithm: str | None = None + + def merged_over(self, defaults: AuthConfig | None) -> AuthConfig: + """These settings, with every one left unset taken from ``defaults``.""" + if defaults is None: + return self + return defaults.model_copy(update=self.model_dump(exclude_none=True)) + + +class FilteringConfig(BaseModel): + """Filtering block inside meta.""" + + backends: list[str] = [] + fields: list[str] = [] + search: list[str] = [] + ordering: list[str] = [] + + +class PaginationConfig(BaseModel): + """Pagination block used in defaults and per-endpoint meta.""" + + style: str = "page_number" + page_size: int = 20 + + +class DefaultsConfig(BaseModel): + """Global defaults applied to all endpoints unless overridden.""" + + authentication: AuthConfig | None = None + pagination: PaginationConfig | None = None + + +class MethodAuthConfig(BaseModel): + """Per-method authentication override inside meta.methods dict.""" + + authentication: AuthConfig | None = None + + +class CacheConfig(BaseModel): + """Cache block inside meta: cache: { ttl: 60 }""" + + ttl: int = 60 + + +class SerializerConfig(BaseModel): + """Serializer block inside meta. + + Use ``fields`` for a unified list, or ``read``/``write`` for per-verb lists. + """ + + fields: list[str] | None = None + read: list[str] | None = None + write: list[str] | None = None + + +class MetaConfig(BaseModel): + """meta: block inside a declarative endpoint entry.""" + + # methods can be a list ["GET", "POST"] or a dict {GET: {...}, DELETE: {...}} + methods: Union[list[str], dict[str, MethodAuthConfig | None]] = [] + authentication: AuthConfig | None = None + filtering: FilteringConfig | None = None + pagination: PaginationConfig | None = None + cache: CacheConfig | None = None + serializer: SerializerConfig | None = None + table: str | None = None # custom table name (required when reflect: true) + + +class FieldSpec(BaseModel): + """Single field definition inside fields:.""" + + type: str + optional: bool = False + # All remaining keys forwarded to Field() as pydantic constraints + model_config = {"extra": "allow"} + + @field_validator("type") + @classmethod + def type_must_be_known(cls, v: str) -> str: + if v not in YAML_FIELD_TYPES: + raise ValueError( + f"Unknown field type '{v}'. Valid types: {sorted(YAML_FIELD_TYPES)}" + ) + return v + + +class EndpointConfig(BaseModel): + """A single endpoint entry.""" + + route: str + fields: dict[str, FieldSpec] = {} + reflect: bool = False + meta: MetaConfig = MetaConfig() + + @model_validator(mode="after") + def require_route(self) -> "EndpointConfig": + if not self.route: + raise ValueError("Each endpoint must have a 'route'.") + return self + + @property + def effective_route(self) -> str: + return self.route.strip() + + +class LightAPIConfig(BaseModel): + """Root YAML document schema.""" + + database: DatabaseConfig | None = None + cors_origins: list[str] = [] + defaults: DefaultsConfig = DefaultsConfig() + endpoints: list[EndpointConfig] = [] + middleware: list[str] = [] + auth: AuthLoginConfig | None = None + mode: str | None = None # "sync" | "async" — auto-detected when omitted + + @property + def effective_database_url(self) -> str | None: + return self.database.url if self.database else None diff --git a/tests/test_yaml_config.py b/tests/test_yaml_config.py index a96b914..16829c9 100644 --- a/tests/test_yaml_config.py +++ b/tests/test_yaml_config.py @@ -10,7 +10,8 @@ from lightapi.exceptions import ConfigurationError from lightapi.lightapi import LightApi -from lightapi.yaml_loader import LightAPIConfig, _resolve_name +from lightapi.yaml_loader import LightAPIConfig +from lightapi.yaml_names import resolve_name # ───────────────────────────────────────────────────────────────────────────── # Helpers @@ -474,18 +475,18 @@ class TestResolveName: def test_builtin_names_resolve(self): from lightapi.auth import IsAuthenticated - assert _resolve_name("IsAuthenticated") is IsAuthenticated + assert resolve_name("IsAuthenticated") is IsAuthenticated def test_dotted_path_resolves(self): - cls = _resolve_name("lightapi.auth.AllowAny") + cls = resolve_name("lightapi.auth.AllowAny") from lightapi.auth import AllowAny assert cls is AllowAny def test_unknown_name_raises(self): with pytest.raises(ConfigurationError, match="Unknown class name"): - _resolve_name("CompletlyUnknownClass") + resolve_name("CompletlyUnknownClass") def test_bad_dotted_path_raises(self): with pytest.raises(ConfigurationError): - _resolve_name("non.existent.module.Foo") + resolve_name("non.existent.module.Foo") From b583efc1531dc1f9998cb409e26f884d90d65ea4 Mon Sep 17 00:00:00 2001 From: iklo Date: Fri, 18 Sep 2026 11:28:17 -0300 Subject: [PATCH 5/5] refactor: build the create/read schema pair in one place SchemaFactory.build and build_from_reflected_table ended with the same twenty lines: add the four auto-managed read fields, create the two pydantic models, rebuild them. build also imported Optional, datetime and ConfigDict a second time inside the function, and set read_fields identically in both arms of an if/else. _schema_pair does that tail once. Field order is unchanged (the endpoint's fields first, then any auto field not already present), and that order is the key order of every JSON response. build: 73 lines to 40. Tests unchanged: 350 passed, 1 xfailed, 2 xpassed. --- lightapi/schema.py | 125 ++++++++++++++++++--------------------------- 1 file changed, 51 insertions(+), 74 deletions(-) diff --git a/lightapi/schema.py b/lightapi/schema.py index aec0c44..f9d8224 100644 --- a/lightapi/schema.py +++ b/lightapi/schema.py @@ -135,6 +135,40 @@ def _reflected_annotation(column: Any) -> Any: return Any +# Every read schema carries the auto-managed columns, after the endpoint's own fields. +_AUTO_READ_FIELDS: dict[str, Any] = { + "id": (Optional[int], None), + "created_at": (Optional[datetime.datetime], None), + "updated_at": (Optional[datetime.datetime], None), + "version": (Optional[int], None), +} + + +def _schema_pair( + endpoint_name: str, create_fields: dict[str, Any], read_fields: dict[str, Any] +) -> tuple[type, type]: + """(create schema, read schema) for an endpoint. + + create: input validation for POST/PUT/PATCH. + read: response serialization; ``extra="allow"`` lets join labels through. + """ + missing_auto = {k: v for k, v in _AUTO_READ_FIELDS.items() if k not in read_fields} + schema_create = create_model( + f"{endpoint_name}CreateSchema", + __config__=ConfigDict(from_attributes=True), + **create_fields, + ) + schema_read = create_model( + f"{endpoint_name}ReadSchema", + __config__=ConfigDict(from_attributes=True, extra="allow"), + **read_fields, + **missing_auto, + ) + schema_create.model_rebuild() + schema_read.model_rebuild() + return schema_create, schema_read + + class SchemaFactory: """Builds Pydantic validation models from a RestEndpoint class.""" @@ -161,57 +195,26 @@ def build(cls: type) -> tuple[type, type]: } ) - field_infos: dict[str, FieldInfo] = {} - for name in user_annotations: - val = cls.__dict__.get(name) or getattr(cls, name, None) - if isinstance(val, FieldInfo): - field_infos[name] = val - create_fields: dict[str, Any] = {} read_fields: dict[str, Any] = {} - - from typing import Optional - for name, annotation in user_annotations.items(): if name in _AUTO_FIELDS: continue - fi = field_infos.get(name) - extra = (fi.json_schema_extra or {}) if fi else {} + value = cls.__dict__.get(name) or getattr(cls, name, None) + field_info = value if isinstance(value, FieldInfo) else None + extra = ( + (field_info.json_schema_extra or {}) if field_info is not None else {} + ) if extra.get("exclude"): continue - if fi is not None: - # create: FieldInfo with constraints for INPUT validation - create_fields[name] = (annotation, fi) - # read: Optional[T] so serializer can project out any field - read_fields[name] = (Optional[annotation], None) # type: ignore[valid-type] - else: - create_fields[name] = (annotation, ...) - read_fields[name] = (Optional[annotation], None) # type: ignore[valid-type] - - import datetime - from typing import Optional - - read_fields["id"] = (Optional[int], None) - read_fields["created_at"] = (Optional[datetime.datetime], None) - read_fields["updated_at"] = (Optional[datetime.datetime], None) - read_fields["version"] = (Optional[int], None) - - from pydantic import ConfigDict - - schema_create = create_model( - f"{cls.__name__}CreateSchema", - __config__=ConfigDict(from_attributes=True), - **create_fields, - ) - schema_read = create_model( - f"{cls.__name__}ReadSchema", - __config__=ConfigDict(from_attributes=True, extra="allow"), - **read_fields, - ) - schema_create.model_rebuild() - schema_read.model_rebuild() - return schema_create, schema_read + # create keeps the Field() constraints; read is Optional so a + # serializer can project any field out. + default = field_info if field_info is not None else ... + create_fields[name] = (annotation, default) + read_fields[name] = (Optional[annotation], None) # type: ignore[valid-type] + + return _schema_pair(cls.__name__, create_fields, read_fields) @staticmethod def build_from_reflected_table(cls: type, table: Any) -> tuple[type, type]: @@ -224,37 +227,11 @@ def build_from_reflected_table(cls: type, table: Any) -> tuple[type, type]: """ create_fields: dict[str, Any] = {} read_fields: dict[str, Any] = {} - for col in table.c: - name = col.key annotation = _reflected_annotation(col) - use_optional = col.nullable - ann = Optional[annotation] if use_optional else annotation # type: ignore[valid-type] + if col.key not in _AUTO_FIELDS: + required = Optional[annotation] if col.nullable else annotation # type: ignore[valid-type] + create_fields[col.key] = (required, ...) + read_fields[col.key] = (Optional[annotation], None) # type: ignore[valid-type] - if name not in _AUTO_FIELDS: - create_fields[name] = (ann, ...) - - read_fields[name] = (Optional[annotation], None) # type: ignore[valid-type] - - if "id" not in read_fields: - read_fields["id"] = (Optional[int], None) - if "created_at" not in read_fields: - read_fields["created_at"] = (Optional[datetime.datetime], None) - if "updated_at" not in read_fields: - read_fields["updated_at"] = (Optional[datetime.datetime], None) - if "version" not in read_fields: - read_fields["version"] = (Optional[int], None) - - schema_create = create_model( - f"{cls.__name__}CreateSchema", - __config__=ConfigDict(from_attributes=True), - **create_fields, - ) - schema_read = create_model( - f"{cls.__name__}ReadSchema", - __config__=ConfigDict(from_attributes=True, extra="allow"), - **read_fields, - ) - schema_create.model_rebuild() - schema_read.model_rebuild() - return schema_create, schema_read + return _schema_pair(cls.__name__, create_fields, read_fields)