Class tigrbl.engine.engine_spec.EngineSpec

tigrbl.engine.engine_spec.EngineSpec dataclass

EngineSpec(
    kind=None,
    async_=False,
    dsn=None,
    mapping=None,
    path=None,
    memory=False,
    user=None,
    pwd=None,
    host=None,
    port=None,
    name=None,
    pool_size=10,
    max=20,
)

Canonical, normalized engine spec → Provider factory.

Input comes from @engine_ctx attached to an App/API/Table/Op: • DSN string: "sqlite://:memory:" , "sqlite:///./file.db" , "sqlite+aiosqlite:///./file.db" , "postgresql://user:pwd@host:5432/db" , "postgresql+asyncpg://user:pwd@host:5432/db" • Mapping (recommended for clarity/portability): {"kind":"sqlite","async":True,"path":"./file.db"} {"kind":"postgres","async":True,"host":"db","db":"app_db",...} { ...} # for plugin engines

kind class-attribute instance-attribute

kind = None

async_ class-attribute instance-attribute

async_ = False

dsn class-attribute instance-attribute

dsn = None

mapping class-attribute instance-attribute

mapping = None

path class-attribute instance-attribute

path = None

memory class-attribute instance-attribute

memory = False

user class-attribute instance-attribute

user = None

pwd class-attribute instance-attribute

pwd = field(default=None, repr=False)

host class-attribute instance-attribute

host = None

port class-attribute instance-attribute

port = None

name class-attribute instance-attribute

name = None

pool_size class-attribute instance-attribute

pool_size = 10

max class-attribute instance-attribute

max = 20

from_any staticmethod

from_any(x)

Parse DSN/Mapping/Provider/Engine into an :class:EngineSpec.

Source code in tigrbl/engine/engine_spec.py

@staticmethod
def from_any(x: EngineCfg | None) -> Optional["EngineSpec"]:
    """Parse DSN/Mapping/Provider/Engine into an :class:`EngineSpec`."""
    if x is None:
        return None

    if isinstance(x, EngineSpec):
        return x

    if isinstance(x, Provider):
        return x.spec

    if isinstance(x, Engine):
        return x.spec

    # DSN string
    if isinstance(x, str):
        s = x.strip()
        # sqlite async
        if s.startswith("sqlite+aiosqlite://") or s.startswith("sqlite+aiosqlite:"):
            path = s.split("sqlite+aiosqlite://")[-1]
            path = path.lstrip("/") or None
            mem = (path is None) or (path == ":memory:")
            return EngineSpec(kind="sqlite", async_=True, path=path, memory=mem, dsn=s)
        # sqlite sync
        if s.startswith("sqlite://") or s.startswith("sqlite:"):
            # handle sqlite://:memory: and sqlite:///file.db
            if s.startswith("sqlite://:memory:") or s.endswith(":memory:"):
                return EngineSpec(kind="sqlite", async_=False, path=None, memory=True, dsn=s)
            # Take the path part after scheme; urlsplit handles both sqlite:// and sqlite:/// forms
            p = urlsplit(s).path.lstrip("/") or None
            mem = (p is None)
            return EngineSpec(kind="sqlite", async_=False, path=p, memory=mem, dsn=s)

        # postgres async
        if s.startswith("postgresql+asyncpg://") or s.startswith("postgres+asyncpg://"):
            u = urlsplit(s)
            # Extract netloc parts user:pwd@host:port
            user = u.username or None
            pwd = u.password or None
            host = u.hostname or None
            port = u.port or None
            name = (u.path or "/").lstrip("/") or None
            return EngineSpec(kind="postgres", async_=True, dsn=s, user=user, pwd=pwd, host=host, port=port, name=name)
        # postgres sync
        if s.startswith("postgresql://") or s.startswith("postgres://"):
            u = urlsplit(s)
            user = u.username or None
            pwd = u.password or None
            host = u.hostname or None
            port = u.port or None
            name = (u.path or "/").lstrip("/") or None
            return EngineSpec(kind="postgres", async_=False, dsn=s, user=user, pwd=pwd, host=host, port=port, name=name)

        raise ValueError(f"Unsupported DSN: {s}")

    # Mapping
    m = x  # type: ignore[assignment]

    # Helpers
    def _get_bool(key: str, *aliases: str, default: bool = False) -> bool:
        for k in (key, *aliases):
            if k in m:
                return bool(m[k])  # type: ignore[index]
        return default

    def _get_str(key: str, *aliases: str, default: Optional[str] = None) -> Optional[str]:
        for k in (key, *aliases):
            if k in m and m[k] is not None:
                return str(m[k])  # type: ignore[index]
        return default

    def _get_int(key: str, *aliases: str, default: Optional[int] = None) -> Optional[int]:
        for k in (key, *aliases):
            if k in m and m[k] is not None:
                try:
                    return int(m[k])  # type: ignore[index]
                except Exception:
                    return default
        return default

    k = str(m.get("kind", m.get("engine", ""))).lower()  # type: ignore[index]
    if k == "sqlite":
        async_ = _get_bool("async", "async_", default=False)
        path = _get_str("path")
        memory = _get_bool("memory", default=False) or (str(m.get("mode", "")).lower() == "memory") or (path is None)
        return EngineSpec(
            kind="sqlite",
            async_=async_,
            path=path,
            memory=memory,
            dsn=_get_str("dsn", "url"),
            mapping=m,
        )

    if k == "postgres":
        async_ = _get_bool("async", "async_", default=False)
        return EngineSpec(
            kind="postgres",
            async_=async_,
            user=_get_str("user"),
            pwd=_get_str("pwd", "password"),
            host=_get_str("host"),
            port=_get_int("port"),
            name=_get_str("db", "name"),
            pool_size=_get_int("pool_size", default=10) or 10,
            max=_get_int("max", "max_overflow", "max_size", default=20) or 20,
            dsn=_get_str("dsn", "url"),
            mapping=m,
        )

    # External / unknown kinds – keep mapping and defer to registry at build()
    return EngineSpec(
        kind=k or None,
        async_=_get_bool("async", "async_", default=False),
        dsn=_get_str("dsn", "url"),
        mapping=m,
    )

build

build()

Construct the engine and sessionmaker for this spec.

Source code in tigrbl/engine/engine_spec.py

def build(self) -> Tuple[Any, SessionFactory]:
    """Construct the engine and sessionmaker for this spec."""
    if self.kind == "sqlite":
        if self.memory:
            if self.async_:
                return async_sqlite_engine(path=None)
            return blocking_sqlite_engine(path=None)
        if not self.path:
            raise ValueError("sqlite file requires 'path'")
        if self.async_:
            return async_sqlite_engine(path=self.path)
        return blocking_sqlite_engine(path=self.path)

    if self.kind == "postgres":
        if self.dsn:
            if self.async_:
                return async_postgres_engine(dsn=self.dsn)
            return blocking_postgres_engine(dsn=self.dsn)
        # keyword build
        kwargs: dict[str, Any] = {
            "user": self.user or "app",
            "host": self.host or "localhost",
            "port": self.port or 5432,
            "db": self.name or "app_db",
            "pool_size": int(self.pool_size or 10),
        }
        if self.pwd is not None:
            kwargs["pwd"] = self.pwd
        if self.async_:
            kwargs["max_size"] = int(self.max or 20)
            return async_postgres_engine(**kwargs)
        else:
            kwargs["max_overflow"] = int(self.max or 20)
            return blocking_postgres_engine(**kwargs)

    # External/registered engines
    try:
        from .plugins import load_engine_plugins
        from .registry import get_engine_registration, known_engine_kinds
        load_engine_plugins()
        reg = get_engine_registration(self.kind or "")
    except Exception:
        reg = None
    if reg:
        mapping = self.mapping or {}
        return reg.build(mapping=mapping, spec=self, dsn=self.dsn)

    # No registration found: helpful error
    try:
        from .registry import known_engine_kinds  # re-import defensive
        kinds = ", ".join(known_engine_kinds()) or "(none)"
    except Exception:
        kinds = "(unknown)"
    raise RuntimeError(
        f"Unknown or unavailable engine kind '{self.kind}'. Installed engine kinds: {kinds}. "
        f"If this is an optional extension, install its package (e.g., 'pip install tigrbl_engine_{self.kind}')."
    )