from __future__ import annotations from dataclasses import dataclass from typing import Any, Dict, Optional, Type, TypeVar, cast from flask import Flask from sqlalchemy.orm import Session from crudkit.core.service import CRUDService from crudkit.api.flask_api import generate_crud_blueprint from crudkit.engines import CRUDKitRuntime T = TypeVar("T") @dataclass class Registered: model: Type[Any] service: CRUDService[Any] blueprint_name: str url_prefix: str class CRUDRegistry: """ Binds: - name -> model class - model class -> CRUDService (using CRUDKitRuntime.session_factory) - model class -> Flask blueprint (via generate_crud_blueprint) """ def __init__(self, runtime: CRUDKitRuntime): self._rt = runtime self._models_by_key: Dict[str, Type[Any]] = {} self._services_by_model: Dict[Type[Any], CRUDService[Any]] = {} self._bps_by_model: Dict[Type[Any], Registered] = {} @staticmethod def _key(model_or_name: Type[Any] | str) -> str: return model_or_name.lower() if isinstance(model_or_name, str) else model_or_name.__name__.lower() def get_model(self, key: str) -> Optional[Type[Any]]: return self._models_by_key.get(key.lower()) def get_service(self, model: Type[T]) -> Optional[CRUDService[T]]: return cast(Optional[CRUDService[T]], self._services_by_model.get(model)) def is_registered(self, model: Type[Any]) -> bool: return model in self._services_by_model def register_class( self, app: Flask, model: Type[Any], *, url_prefix: Optional[str] = None, blueprint_name: Optional[str] = None, polymorphic: bool = False, service_kwargs: Optional[dict] = None ) -> Registered: """ Register a model: - store name -> class - create a CRUDService bound to a Session from runtime.session_factory - attach backend into from runtime.backend - mount Flask blueprint at /api/ by default Idempotent for each model. """ key = self._key(model) self._models_by_key.setdefault(key, model) svc = self._services_by_model.get(model) if svc is None: SessionMaker = self._rt.session_factory if SessionMaker is None: raise RuntimeError("CRUDKitRuntime.session_factory is not initialized.") session: Session = SessionMaker() svc = CRUDService( model, session=session, polymorphic=polymorphic, backend=self._rt.backend, **(service_kwargs or {}), ) self._services_by_model[model] = svc reg = self._bps_by_model.get(model) if reg: return reg prefix = url_prefix or f"/api/{key}" bp_name = blueprint_name or f"crudkit.{key}" bp = generate_crud_blueprint(model, svc) bp.name = bp_name app.register_blueprint(bp, url_prefix=prefix) reg = Registered(model=model, service=svc, blueprint_name=bp_name, url_prefix=prefix) self._bps_by_model[model] = reg return reg def register_many( self, app: Flask, models: list[Type[Any]], *, base_prefix: str = "/api", polymorphic: bool = False, service_kwargs: Optional[dict] = None, ) -> list[Registered]: out: list[Registered] = [] for m in models: key = self._key(m) out.append( self.register_class( app, m, url_prefix=f"{base_prefix}/{key}", polymorphic=polymorphic, service_kwargs=service_kwargs, ) ) return out