Class tigrbl.app.tigrbl_app.TigrblApp

tigrbl.app.tigrbl_app.TigrblApp

TigrblApp(
    *,
    engine=None,
    jsonrpc_prefix="/rpc",
    system_prefix="/system",
    api_hooks=None,
    **fastapi_kwargs,
)

Bases: App

Monolithic facade that owns

• containers (models, schemas, handlers, hooks, rpc, rest, routers, columns, table_config, core proxies) • model inclusion (REST + RPC wiring) • JSON-RPC / diagnostics mounting • (optional) legacy-friendly helpers (transactional decorator, auth flags)

It composes v3 primitives; you can still use the functions directly if you prefer.

Source code in tigrbl/app/tigrbl_app.py

def __init__(
    self,
    *,
    engine: EngineCfg | None = None,
    jsonrpc_prefix: str = "/rpc",
    system_prefix: str = "/system",
    api_hooks: Mapping[str, Iterable[Callable]]
    | Mapping[str, Mapping[str, Iterable[Callable]]]
    | None = None,
    **fastapi_kwargs: Any,
) -> None:
    title = fastapi_kwargs.pop("title", None)
    if title is not None:
        self.TITLE = title
    version = fastapi_kwargs.pop("version", None)
    if version is not None:
        self.VERSION = version
    lifespan = fastapi_kwargs.pop("lifespan", None)
    if lifespan is not None:
        self.LIFESPAN = lifespan
    super().__init__(engine=engine, **fastapi_kwargs)
    # capture initial routes so refreshes retain FastAPI defaults
    self._base_routes = list(self.router.routes)
    self.jsonrpc_prefix = jsonrpc_prefix
    self.system_prefix = system_prefix

    # public containers (mirrors used by bindings.api)
    self.models: Dict[str, type] = {}
    self.schemas = SimpleNamespace()
    self.handlers = SimpleNamespace()
    self.hooks = SimpleNamespace()
    self.rpc = SimpleNamespace()
    self.rest = SimpleNamespace()
    self.routers: Dict[str, Any] = {}
    self.tables = AttrDict()
    self.columns: Dict[str, Tuple[str, ...]] = {}
    self.table_config: Dict[str, Dict[str, Any]] = {}
    self.core = SimpleNamespace()
    self.core_raw = SimpleNamespace()

    # API-level hooks map (merged into each model at include-time; precedence handled in bindings.hooks)
    self._api_hooks_map = copy.deepcopy(api_hooks) if api_hooks else None

TITLE class-attribute instance-attribute

TITLE = 'TigrblApp'

VERSION class-attribute instance-attribute

VERSION = '0.1.0'

LIFESPAN class-attribute instance-attribute

LIFESPAN = None

MIDDLEWARES class-attribute instance-attribute

MIDDLEWARES = ()

APIS class-attribute instance-attribute

APIS = ()

MODELS class-attribute instance-attribute

MODELS = ()

jsonrpc_prefix instance-attribute

jsonrpc_prefix = jsonrpc_prefix

system_prefix instance-attribute

system_prefix = system_prefix

models instance-attribute

models = {}

schemas instance-attribute

schemas = SimpleNamespace()

handlers instance-attribute

handlers = SimpleNamespace()

hooks instance-attribute

hooks = SimpleNamespace()

rpc instance-attribute

rpc = SimpleNamespace()

rest instance-attribute

rest = SimpleNamespace()

routers instance-attribute

routers = {}

tables instance-attribute

tables = AttrDict()

columns instance-attribute

columns = {}

table_config instance-attribute

table_config = {}

core instance-attribute

core = SimpleNamespace()

core_raw instance-attribute

core_raw = SimpleNamespace()

initialize class-attribute instance-attribute

initialize = initialize

title instance-attribute

title = TITLE

version instance-attribute

version = VERSION

engine instance-attribute

engine = (
    engine
    if engine is not None
    else getattr(self, "ENGINE", None)
)

apis instance-attribute

apis = tuple(getattr(self, 'APIS', ()))

ops instance-attribute

ops = tuple(getattr(self, 'OPS', ()))

security_deps instance-attribute

security_deps = tuple(getattr(self, 'SECURITY_DEPS', ()))

deps instance-attribute

deps = tuple(getattr(self, 'DEPS', ()))

response instance-attribute

response = getattr(self, 'RESPONSE', None)

middlewares instance-attribute

middlewares = tuple(getattr(self, 'MIDDLEWARES', ()))

lifespan instance-attribute

lifespan = LIFESPAN

include_model

include_model(model, *, prefix=None, mount_router=True)

Bind a model, mount its REST router, and attach all namespaces to this facade.

Source code in tigrbl/app/tigrbl_app.py

def include_model(
    self, model: type, *, prefix: str | None = None, mount_router: bool = True
) -> Tuple[type, Any]:
    """
    Bind a model, mount its REST router, and attach all namespaces to this facade.
    """
    # inject API-level hooks so the binder merges them
    self._merge_api_hooks_into_model(model, self._api_hooks_map)
    return _include_model(self, model, prefix=prefix, mount_router=mount_router)

include_models

include_models(
    models, *, base_prefix=None, mount_router=True
)

Source code in tigrbl/app/tigrbl_app.py

def include_models(
    self,
    models: Sequence[type],
    *,
    base_prefix: str | None = None,
    mount_router: bool = True,
) -> Dict[str, Any]:
    for m in models:
        self._merge_api_hooks_into_model(m, self._api_hooks_map)
    return _include_models(
        self,
        models,
        base_prefix=base_prefix,
        mount_router=mount_router,
    )

rpc_call async

rpc_call(
    model_or_name,
    method,
    payload=None,
    *,
    db,
    request=None,
    ctx=None,
)

Source code in tigrbl/app/tigrbl_app.py

async def rpc_call(
    self,
    model_or_name: type | str,
    method: str,
    payload: Any = None,
    *,
    db: Any,
    request: Any = None,
    ctx: Optional[Dict[str, Any]] = None,
) -> Any:
    return await _rpc_call(
        self, model_or_name, method, payload, db=db, request=request, ctx=ctx
    )

mount_jsonrpc

mount_jsonrpc(*, prefix=None)

Mount JSON-RPC router onto this app.

Source code in tigrbl/app/tigrbl_app.py

def mount_jsonrpc(self, *, prefix: str | None = None) -> Any:
    """Mount JSON-RPC router onto this app."""
    px = prefix if prefix is not None else self.jsonrpc_prefix
    prov = _resolver.resolve_provider(api=self)
    get_db = prov.get_db if prov else None
    router = _mount_jsonrpc(
        self,
        self,
        prefix=px,
        get_db=get_db,
    )
    self._base_routes = list(self.router.routes)
    return router

attach_diagnostics

attach_diagnostics(*, prefix=None, app=None)

Mount diagnostics router onto this app or the provided app.

Source code in tigrbl/app/tigrbl_app.py

def attach_diagnostics(
    self, *, prefix: str | None = None, app: Any | None = None
) -> Any:
    """Mount diagnostics router onto this app or the provided ``app``."""
    px = prefix if prefix is not None else self.system_prefix
    prov = _resolver.resolve_provider(api=self)
    get_db = prov.get_db if prov else None
    router = _mount_diagnostics(self, get_db=get_db)
    include_self = getattr(self, "include_router", None)
    if callable(include_self):
        include_self(router, prefix=px)
    if app is not None and app is not self:
        include_other = getattr(app, "include_router", None)
        if callable(include_other):
            include_other(router, prefix=px)
    if app is None:
        self._base_routes = list(self.router.routes)
    return router

registry

registry(model)

Return the per-model OpspecRegistry.

Source code in tigrbl/app/tigrbl_app.py

def registry(self, model: type):
    """Return the per-model OpspecRegistry."""
    return get_registry(model)

bind

bind(model)

Bind/rebuild a model in place (without mounting).

Source code in tigrbl/app/tigrbl_app.py

def bind(self, model: type) -> Tuple[OpSpec, ...]:
    """Bind/rebuild a model in place (without mounting)."""
    self._merge_api_hooks_into_model(model, self._api_hooks_map)
    return _bind(model)

rebind

rebind(model, *, changed_keys=None)

Targeted rebuild of a bound model.

Source code in tigrbl/app/tigrbl_app.py

def rebind(
    self, model: type, *, changed_keys: Optional[set[tuple[str, str]]] = None
) -> Tuple[OpSpec, ...]:
    """Targeted rebuild of a bound model."""
    return _rebind(model, changed_keys=changed_keys)

transactional

transactional(*dargs, **dkw)

Legacy-friendly decorator: @api.transactional(...) Wraps a function as a v3 custom op with START_TX/END_TX.

Source code in tigrbl/app/tigrbl_app.py

def transactional(self, *dargs, **dkw):
    """
    Legacy-friendly decorator: @api.transactional(...)
    Wraps a function as a v3 custom op with START_TX/END_TX.
    """
    if _txn_decorator is None:
        raise RuntimeError("transactional decorator not available")
    return _txn_decorator(self, *dargs, **dkw)

set_auth

set_auth(
    *,
    authn=None,
    allow_anon=None,
    authorize=None,
    optional_authn_dep=None,
)

Source code in tigrbl/app/tigrbl_app.py

def set_auth(
    self,
    *,
    authn: Any = None,
    allow_anon: Optional[bool] = None,
    authorize: Any = None,
    optional_authn_dep: Any = None,
) -> None:
    if authn is not None:
        self._authn = authn
    if allow_anon is not None:
        self._allow_anon = bool(allow_anon)
    if authorize is not None:
        self._authorize = authorize
    if optional_authn_dep is not None:
        self._optional_authn_dep = optional_authn_dep

    # Refresh already-included models so routers pick up new auth settings
    if self.models:
        self._refresh_security()

install_engines

install_engines(*, api=None, models=None)

Source code in tigrbl/app/_app.py

def install_engines(
    self, *, api: Any = None, models: tuple[Any, ...] | None = None
) -> None:
    # If class declared APIS/MODELS, use them unless explicit args are passed.
    apis = (api,) if api is not None else self.APIS
    models = models if models is not None else self.MODELS
    if apis:
        for a in apis:
            install_from_objects(app=self, api=a, models=models)
    else:
        install_from_objects(app=self, api=None, models=models)