from __future__ import annotations from typing import Type, Optional from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from .backend import make_backend_info, BackendInfo from .config import Config, get_config from ._sqlite import apply_sqlite_pragmas def build_engine(config_cls: Type[Config] | None = None): config_cls = config_cls or get_config(None) engine = create_engine(config_cls.DATABASE_URL, **config_cls.engine_kwargs()) apply_sqlite_pragmas(engine, config_cls.SQLITE_PRAGMAS) return engine def build_sessionmaker(config_cls: Type[Config] | None = None, engine=None): config_cls = config_cls or get_config(None) engine = engine or build_engine(config_cls) return sessionmaker(bind=engine, **config_cls.session_kwargs()) class CRUDKitRuntime: """ Lightweight container so CRUDKit can be given either: - prebuild engine/sessionmaker, or - a Config to build them lazily """ def __init__(self, *, engine=None, session_factory=None, config: Optional[Type[Config]] = None): if engine is None and session_factory is None and config is None: config = get_config(None) self._config = config self._engine = engine or (build_engine(config) if config else None) self._session_factory = session_factory or (build_sessionmaker(config, self._engine) if config else None) @property def engine(self): if self._engine is None and self._config: self._engine = build_engine(self._config) return self._engine @property def session_factory(self): if self._session_factory is None: if self._config and self._engine: self._session_factory = build_sessionmaker(self._config, self._engine) return self._session_factory @property def backend(self) -> BackendInfo: if not hasattr(self, "_backend_info") or self._backend_info is None: self._backend_info = make_backend_info(self.engine) return self._backend_info