Reference

Core Module

The core functionality of sqlalchemy-tenants.

get_table_policy

get_table_policy(
    *, table_name: str, column_type: Type[TenantIdentifier]
) -> str

Returns the SQL policy for a given table name.

Source code in src/sqlalchemy_tenants/core.py

def get_table_policy(*, table_name: str, column_type: Type[TenantIdentifier]) -> str:
    """
    Returns the SQL policy for a given table name.
    """
    if column_type is str:
        sql_type = "varchar"
    elif column_type is int:
        sql_type = "integer"
    elif column_type is UUID:
        sql_type = "uuid"
    else:
        raise TypeError(f"Unknown column type {column_type}")
    policy = _POLICY_TEMPLATE.format(
        table_name=table_name,
        get_tenant_fn=GET_TENANT_FUNCTION_NAME,
        policy_name=_POLICY_NAME,
        sql_type=sql_type,
    )
    return normalize_whitespace(policy)

get_tenant_role_name

get_tenant_role_name(tenant: TenantIdentifier) -> str

Get the Postgres role name for the given tenant.

Parameters:

Name	Type	Description	Default
tenant	TenantIdentifier	the tenant slug.	required

Returns:

Type	Description
str	The Postgres role name for the tenant.

Source code in src/sqlalchemy_tenants/core.py

def get_tenant_role_name(tenant: TenantIdentifier) -> str:
    """
    Get the Postgres role name for the given tenant.

    Args:
        tenant: the tenant slug.

    Returns:
        The Postgres role name for the tenant.
    """
    return f"{TENANT_ROLE_PREFIX}{str(tenant)}"

with_rls

with_rls(cls: Type[T]) -> Type[T]

Decorator to apply RLS (Row Level Security) to a SQLAlchemy model. Validates that the model includes a 'tenant' column.

Source code in src/sqlalchemy_tenants/core.py

def with_rls(cls: Type[T]) -> Type[T]:
    """
    Decorator to apply RLS (Row Level Security) to a SQLAlchemy model.
    Validates that the model includes a 'tenant' column.
    """
    mapper = inspect(cls, raiseerr=False)
    if mapper is None:
        raise TypeError(
            f"@with_rls must be applied to a SQLAlchemy ORM model class, got: {cls}"
        )

    if "tenant" not in mapper.columns:
        raise TypeError(
            f"Model '{cls.__name__}' is marked for RLS but is missing a required "
            f"'tenant' column."
            "\nHint: you can use 'sqlalchemy_tenant TenantMixin' class to add it "
            "easily."
        )

    tenant_column = mapper.columns["tenant"]
    if tenant_column.type.python_type not in TENANT_SUPPORTED_TYPES:
        raise TypeError(
            f"Model '{cls.__name__}' is marked for RLS but 'tenant' "
            f"has type '{tenant_column.type.python_type}', expected one "
            f"of the following: {', '.join(map(str, TENANT_SUPPORTED_TYPES))}."
        )

    setattr(cls.__table__, _ATTRIBUTE_RLS_ENABLED, True)
    setattr(
        cls.__table__, _ATTRIBUTE_TENANT_COLUMN_TYPE, tenant_column.type.python_type
    )
    return cls

Managers

sqlalchemy_tenants.managers.DBManager

Bases: Protocol

create_tenant abstractmethod

create_tenant(tenant: TenantIdentifier) -> None

Create a new tenant with the specified identifier.

Parameters:

Name	Type	Description	Default
tenant	TenantIdentifier	The identifier (slug or ID) of the tenant to create.	required

Source code in src/sqlalchemy_tenants/managers.py

@abstractmethod
def create_tenant(self, tenant: TenantIdentifier) -> None:
    """
    Create a new tenant with the specified identifier.

    Args:
        tenant: The identifier (slug or ID) of the tenant to create.
    """

delete_tenant abstractmethod

delete_tenant(tenant: TenantIdentifier) -> None

Delete a tenant and all its associated roles and privileges, reassigning owned objects to the current user.

No data will be deleted, only the role and privileges.

Parameters:

Name	Type	Description	Default
tenant	TenantIdentifier	The identifier of the tenant to delete.	required

Source code in src/sqlalchemy_tenants/managers.py

@abstractmethod
def delete_tenant(self, tenant: TenantIdentifier) -> None:
    """
    Delete a tenant and all its associated roles and privileges,
    reassigning owned objects to the current user.

    No data will be deleted, only the role and privileges.

    Args:
        tenant: The identifier of the tenant to delete.
    """

list_tenants abstractmethod

list_tenants() -> Set[TenantIdentifier]

Get all the available tenants.

Returns:

Type	Description
Set[TenantIdentifier]	A set with all the available tenants.

Source code in src/sqlalchemy_tenants/managers.py

@abstractmethod
def list_tenants(self) -> Set[TenantIdentifier]:
    """
    Get all the available tenants.

    Returns:
        A set with all the available tenants.
    """

new_tenant_session abstractmethod

new_tenant_session(
    tenant: TenantIdentifier, create_if_missing: bool = True
) -> ContextManager[TenantSession]

Create a new SQLAlchemy session scoped to a specific tenant.

The session uses the tenant's PostgreSQL role and is subject to Row-Level Security (RLS) policies. All queries and writes are automatically restricted to data belonging to the specified tenant.

Parameters:

Name	Type	Description	Default
tenant	TenantIdentifier	The identifier of the tenant.	required
create_if_missing	bool	Whether to create the tenant role if it doesn't exist.	True

Yields:

Type	Description
ContextManager[TenantSession]	A SQLAlchemy session restricted to the tenant's data via RLS.

Raises:

Type	Description
TenantNotFound	If the tenant role doesn't exist and create_if_missing is False.

Source code in src/sqlalchemy_tenants/managers.py

@abstractmethod
def new_tenant_session(
    self,
    tenant: TenantIdentifier,
    create_if_missing: bool = True,
) -> ContextManager[TenantSession]:
    """
    Create a new SQLAlchemy session scoped to a specific tenant.

    The session uses the tenant's PostgreSQL role and is subject to Row-Level
    Security (RLS) policies. All queries and writes are automatically restricted
    to data belonging to the specified tenant.

    Args:
        tenant: The identifier of the tenant.
        create_if_missing: Whether to create the tenant role if it doesn't exist.

    Yields:
        A SQLAlchemy session restricted to the tenant's data via RLS.

    Raises:
        TenantNotFound: If the tenant role doesn't exist and `create_if_missing`
            is False.
    """

new_session abstractmethod

new_session() -> ContextManager[Session]

Create a new admin session with unrestricted access to all tenant data.

This session is not bound to any tenant role and is not subject to RLS policies.

Yields:

Type	Description
ContextManager[Session]	An asynchronous SQLAlchemy session with full database access.

Source code in src/sqlalchemy_tenants/managers.py

@abstractmethod
def new_session(self) -> ContextManager[Session]:
    """
    Create a new admin session with unrestricted access to all tenant data.

    This session is not bound to any tenant role and is not subject to
    RLS policies.

    Yields:
        An asynchronous SQLAlchemy session with full database access.
    """

sqlalchemy_tenants.managers.PostgresManager

PostgresManager(
    schema_name: str,
    engine: Engine,
    session_maker: sessionmaker[Session],
)

Bases: DBManager

Source code in src/sqlalchemy_tenants/managers.py

def __init__(
    self,
    schema_name: str,
    engine: Engine,
    session_maker: sessionmaker[Session],
) -> None:
    self.engine = engine
    self.schema = schema_name
    self.session_maker = session_maker

Managers [async]

sqlalchemy_tenants.aio.managers.DBManager

Bases: Protocol

create_tenant abstractmethod async

create_tenant(tenant: TenantIdentifier) -> None

Create a new tenant with the specified identifier.

Parameters:

Name	Type	Description	Default
tenant	TenantIdentifier	The identifier of the tenant to create.	required

Source code in src/sqlalchemy_tenants/aio/managers.py

@abstractmethod
async def create_tenant(self, tenant: TenantIdentifier) -> None:
    """
    Create a new tenant with the specified identifier.

    Args:
        tenant: The identifier of the tenant to create.
    """

delete_tenant abstractmethod async

delete_tenant(tenant: TenantIdentifier) -> None

Delete a tenant and all its associated roles and privileges, reassigning owned objects to the current user.

No data will be deleted, only the role and privileges.

Parameters:

Name	Type	Description	Default
tenant	TenantIdentifier	The identifier of the tenant to delete.	required

Source code in src/sqlalchemy_tenants/aio/managers.py

@abstractmethod
async def delete_tenant(self, tenant: TenantIdentifier) -> None:
    """
    Delete a tenant and all its associated roles and privileges,
    reassigning owned objects to the current user.

    No data will be deleted, only the role and privileges.

    Args:
        tenant: The identifier of the tenant to delete.
    """

list_tenants abstractmethod async

list_tenants() -> Set[TenantIdentifier]

Get all the available tenants.

Returns:

Type	Description
Set[TenantIdentifier]	A set with all the available tenants.

Source code in src/sqlalchemy_tenants/aio/managers.py

@abstractmethod
async def list_tenants(self) -> Set[TenantIdentifier]:
    """
    Get all the available tenants.

    Returns:
        A set with all the available tenants.
    """

new_tenant_session abstractmethod

new_tenant_session(
    tenant: TenantIdentifier, create_if_missing: bool = True
) -> AsyncContextManager[AsyncTenantSession]

Create a new SQLAlchemy session scoped to a specific tenant.

The session uses the tenant's PostgreSQL role and is subject to Row-Level Security (RLS) policies. All queries and writes are automatically restricted to data belonging to the specified tenant.

Parameters:

Name	Type	Description	Default
tenant	TenantIdentifier	The tenant identifier, which must match a valid PostgreSQL role used for RLS enforcement.	required
create_if_missing	bool	Whether to create the tenant role if it doesn't exist.	True

Yields:

Type	Description
AsyncContextManager[AsyncTenantSession]	A SQLAlchemy session restricted to the tenant's data via RLS.

Raises:

Type	Description
TenantNotFound	If the tenant role doesn't exist and create_if_missing is False.

Source code in src/sqlalchemy_tenants/aio/managers.py

@abstractmethod
def new_tenant_session(
    self,
    tenant: TenantIdentifier,
    create_if_missing: bool = True,
) -> AsyncContextManager[AsyncTenantSession]:
    """
    Create a new SQLAlchemy session scoped to a specific tenant.

    The session uses the tenant's PostgreSQL role and is subject to Row-Level
    Security (RLS) policies. All queries and writes are automatically restricted
    to data belonging to the specified tenant.

    Args:
        tenant: The tenant identifier, which must match a valid PostgreSQL role
            used for RLS enforcement.
        create_if_missing: Whether to create the tenant role if it doesn't exist.

    Yields:
        A SQLAlchemy session restricted to the tenant's data via RLS.

    Raises:
        TenantNotFound: If the tenant role doesn't exist and `create_if_missing`
            is False.
    """

new_session abstractmethod

new_session() -> AsyncContextManager[AsyncSession]

Create a new admin session with unrestricted access to all tenant data.

This session is not bound to any tenant role and is not subject to RLS policies.

Yields:

Type	Description
AsyncContextManager[AsyncSession]	An asynchronous SQLAlchemy session with full database access.

Source code in src/sqlalchemy_tenants/aio/managers.py

@abstractmethod
def new_session(self) -> AsyncContextManager[AsyncSession]:
    """
    Create a new admin session with unrestricted access to all tenant data.

    This session is not bound to any tenant role and is not subject to
    RLS policies.

    Yields:
        An asynchronous SQLAlchemy session with full database access.
    """

sqlalchemy_tenants.aio.managers.PostgresManager

PostgresManager(
    schema_name: str,
    engine: AsyncEngine,
    session_maker: async_sessionmaker[AsyncSession],
)

Bases: DBManager

Source code in src/sqlalchemy_tenants/aio/managers.py

def __init__(
    self,
    schema_name: str,
    engine: AsyncEngine,
    session_maker: async_sessionmaker[AsyncSession],
) -> None:
    self.engine = engine
    self.schema = schema_name
    self.session_maker = session_maker

Exceptions

Exception classes used throughout the library.

SqlalchemyTenantErr

Bases: Exception

Base class for all exceptions raised by the tenants package.

TenantAlreadyExists

TenantAlreadyExists(tenant: TenantIdentifier)

Bases: SqlalchemyTenantErr

Raised when trying to create a tenant that already exists.

Source code in src/sqlalchemy_tenants/exceptions.py

def __init__(self, tenant: TenantIdentifier) -> None:
    super().__init__(f"Tenant '{tenant}' already exists.")

TenantNotFound

TenantNotFound(tenant: TenantIdentifier)

Bases: SqlalchemyTenantErr

Raised when a tenant is not found.

Source code in src/sqlalchemy_tenants/exceptions.py

def __init__(self, tenant: TenantIdentifier) -> None:
    super().__init__(f"Tenant '{tenant}' not found.")