Dependencies

make_tenant_db_dependency

Python

make_tenant_db_dependency(manager: TenancyManager) -> Any

Create a FastAPI dependency that yields a tenant-scoped AsyncSession.

The returned async generator function captures manager in its closure. This is the correct pattern — no app.state lookup, no circular imports, and the dependency works regardless of application startup order.

Parameters:

Name	Type	Description	Default
manager	TenancyManager	The configured :class:~fastapi_tenancy.manager.TenancyManager.	required

Returns:

Type	Description
Any	An async generator function suitable for use as a FastAPI Depends.

Example::

Text Only

get_tenant_db = make_tenant_db_dependency(manager)

@app.get("/orders")
async def list_orders(
    session: Annotated[AsyncSession, Depends(get_tenant_db)],
):
    result = await session.execute(select(Order))

Source code in src/fastapi_tenancy/dependencies.py

def make_tenant_db_dependency(
    manager: TenancyManager,
) -> Any:
    """Create a FastAPI dependency that yields a tenant-scoped ``AsyncSession``.

    The returned async generator function captures *manager* in its closure.
    This is the correct pattern — no ``app.state`` lookup, no circular
    imports, and the dependency works regardless of application startup order.

    Args:
        manager: The configured :class:`~fastapi_tenancy.manager.TenancyManager`.

    Returns:
        An async generator function suitable for use as a FastAPI ``Depends``.

    Example::

        get_tenant_db = make_tenant_db_dependency(manager)

        @app.get("/orders")
        async def list_orders(
            session: Annotated[AsyncSession, Depends(get_tenant_db)],
        ):
            result = await session.execute(select(Order))
    """

    async def _get_tenant_db(
        tenant: Annotated[Tenant, Depends(get_current_tenant)],
    ) -> AsyncIterator[AsyncSession]:
        """Yield a database session scoped to the current tenant.

        Args:
            tenant: The current request's tenant (injected by FastAPI).

        Yields:
            An :class:`~sqlalchemy.ext.asyncio.AsyncSession` configured for
            *tenant*'s isolation namespace.

        Raises:
            IsolationError: When the session cannot be opened.
        """
        async with manager.isolation_provider.get_session(tenant) as session:
            yield session

    return _get_tenant_db

make_tenant_config_dependency

Python

make_tenant_config_dependency(
    manager: TenancyManager,
) -> Any

Create a FastAPI dependency that yields the current tenant's config.

Reads tenant.metadata and constructs a TenantConfig with typed quota and feature-flag fields. Falls back to defaults when fields are absent from the metadata.

Parameters:

Name	Type	Description	Default
manager	TenancyManager	The configured :class:~fastapi_tenancy.manager.TenancyManager.	required

Returns:

Type	Description
Any	An async function returning a TenantConfig instance.

Example::

Text Only

get_tenant_config = make_tenant_config_dependency(manager)

@app.get("/status")
async def status(
    config: Annotated[TenantConfig, Depends(get_tenant_config)],
):
    return {"max_users": config.max_users}

Source code in src/fastapi_tenancy/dependencies.py

def make_tenant_config_dependency(
    manager: TenancyManager,
) -> Any:
    """Create a FastAPI dependency that yields the current tenant's config.

    Reads ``tenant.metadata`` and constructs a ``TenantConfig`` with typed
    quota and feature-flag fields.  Falls back to defaults when fields are
    absent from the metadata.

    Args:
        manager: The configured :class:`~fastapi_tenancy.manager.TenancyManager`.

    Returns:
        An async function returning a ``TenantConfig`` instance.

    Example::

        get_tenant_config = make_tenant_config_dependency(manager)

        @app.get("/status")
        async def status(
            config: Annotated[TenantConfig, Depends(get_tenant_config)],
        ):
            return {"max_users": config.max_users}
    """

    async def _get_tenant_config(
        tenant: Annotated[Tenant, Depends(get_current_tenant)],
    ) -> TenantConfig:
        """Return the ``TenantConfig`` parsed from the current tenant's metadata.

        Args:
            tenant: The current request's tenant.

        Returns:
            Validated :class:`~fastapi_tenancy.core.types.TenantConfig`.
        """
        return TenantConfig.model_validate(tenant.metadata)

    return _get_tenant_config

make_audit_log_dependency

Python

make_audit_log_dependency(manager: TenancyManager) -> Any

Create a FastAPI dependency that provides an audit-log writer function.

Returns a callable that the route handler uses to record operations::

Text Only

get_audit_logger = make_audit_log_dependency(manager)

@app.delete("/orders/{order_id}")
async def delete_order(
    order_id: str,
    audit: Annotated[..., Depends(get_audit_logger)],
    tenant: TenantDep,
):
    ...
    await audit(action="delete", resource="order", resource_id=order_id)

Parameters:

Name	Type	Description	Default
manager	TenancyManager	The configured :class:~fastapi_tenancy.manager.TenancyManager.	required

Returns:

Type	Description
Any	An async function that returns a write-audit-log callable.

Source code in src/fastapi_tenancy/dependencies.py

def make_audit_log_dependency(
    manager: TenancyManager,
) -> Any:
    """Create a FastAPI dependency that provides an audit-log writer function.

    Returns a callable that the route handler uses to record operations::

        get_audit_logger = make_audit_log_dependency(manager)

        @app.delete("/orders/{order_id}")
        async def delete_order(
            order_id: str,
            audit: Annotated[..., Depends(get_audit_logger)],
            tenant: TenantDep,
        ):
            ...
            await audit(action="delete", resource="order", resource_id=order_id)

    Args:
        manager: The configured :class:`~fastapi_tenancy.manager.TenancyManager`.

    Returns:
        An async function that returns a write-audit-log callable.
    """
    from fastapi_tenancy.core.types import AuditLog  # noqa: PLC0415

    async def _get_audit_logger(
        request: Request,
        tenant: Annotated[Tenant, Depends(get_current_tenant)],
    ) -> Any:
        """Return an async function that logs an audit entry for the current tenant.

        Args:
            request: The current HTTP request (used to extract client IP and
                user-agent for the audit record).
            tenant: Current tenant.

        Returns:
            An ``async def log(action, resource, **kwargs)`` callable.
        """
        # Capture request-level context once so the inner ``log``
        # closure does not need to re-read request headers on every call.
        client_ip: str | None = request.client.host if request.client else None
        user_agent: str | None = request.headers.get("user-agent")

        async def log(
            action: str,
            resource: str,
            resource_id: str | None = None,
            metadata: dict[str, Any] | None = None,
            user_id: str | None = None,
        ) -> None:
            """Record an audit log entry.

            Args:
                action: Verb (e.g. ``"create"``, ``"delete"``).
                resource: Resource type (e.g. ``"order"``, ``"user"``).
                resource_id: Optional resource identifier.
                metadata: Optional supplementary context.
                user_id: Optional authenticated user ID.
            """
            from datetime import UTC, datetime  # noqa: PLC0415

            entry = AuditLog(
                tenant_id=tenant.id,
                user_id=user_id,
                action=action,
                resource=resource,
                resource_id=resource_id,
                metadata=metadata or {},
                ip_address=client_ip,
                user_agent=user_agent,
                timestamp=datetime.now(UTC),
            )
            # Delegate to manager's audit log writer if configured.
            if hasattr(manager, "write_audit_log"):
                await manager.write_audit_log(entry)

        return log

    return _get_audit_logger