API Reference¶

Auto-generated from source code docstrings.

Top-Level Exports¶

The most common symbols are available directly from django_rls_tenants:

from django_rls_tenants import (
    AddM2MRLSPolicy,
    RLSConstraint,
    RLSM2MConstraint,
    RLSManager,
    RLSProtectedModel,
    RLSTenantMiddleware,
    RLSTenantModelAdmin,
    TenantQuerySet,
    TenantUser,
    admin_context,
    async_admin_context,
    async_tenant_context,
    current_tenant_value_sql,
    safe_tenant_sql,
    tenant_context,
    with_rls_context,
)

Removed from top-level in v1.2.1

Raw state functions (get_current_tenant_id, set_current_tenant_id, reset_current_tenant_id, get_rls_context_active, set_rls_context_active, reset_rls_context_active) and exception classes (NoTenantContextError, RLSConfigurationError, RLSTenantError) are no longer re-exported from the top-level package. Import them from their actual modules instead:

from django_rls_tenants.tenants.state import get_current_tenant_id
from django_rls_tenants.exceptions import NoTenantContextError

Exceptions¶

Custom exception hierarchy for precise error handling. All exceptions live in django_rls_tenants.exceptions. Import them from that module directly.

django_rls_tenants.exceptions.RLSTenantError ¶

RLSTenantError(message: str, *, hint: str | None = None)

Bases: Exception

Base exception for all django-rls-tenants errors.

Catch this to handle any error raised by the library.

Accepts an optional, keyword-only hint describing how to fix the error. When supplied, it is appended to str(exc) after a blank line and a Hint: label, so the suggestion shows up in tracebacks and logs. The bare message and the hint are also exposed as attributes for programmatic access.

Parameters:

Name	Type	Description	Default
`message`	`str`	Human-readable description of what went wrong.	required
`hint`	`str \| None`	Optional actionable suggestion for fixing the error.	`None`

Attributes:

Name	Type	Description
`message`		The error description, without the hint.
`hint`		The actionable suggestion, or `None` when not provided.

Source code in django_rls_tenants/exceptions.py

def __init__(self, message: str, *, hint: str | None = None) -> None:
    self.message = message
    self.hint = hint
    super().__init__(f"{message}\n\nHint: {hint}" if hint else message)

django_rls_tenants.exceptions.NoTenantContextError ¶

NoTenantContextError(
    message: str, *, hint: str | None = None
)

Bases: RLSTenantError

Query or context operation attempted without an active tenant context.

Raised when STRICT_MODE=True and a queryset evaluation is attempted without an active tenant_context(), admin_context(), for_user(), or RLSTenantMiddleware context.

Also raised by tenant_context() and _resolve_user_guc_vars() when a non-admin user has rls_tenant_id=None.

Source code in django_rls_tenants/exceptions.py

def __init__(self, message: str, *, hint: str | None = None) -> None:
    self.message = message
    self.hint = hint
    super().__init__(f"{message}\n\nHint: {hint}" if hint else message)

django_rls_tenants.exceptions.RLSConfigurationError ¶

RLSConfigurationError(
    message: str, *, hint: str | None = None
)

Bases: RLSTenantError

Invalid or missing RLS configuration.

Raised when a required configuration key is missing from settings.RLS_TENANTS or when a configuration value is invalid.

Source code in django_rls_tenants/exceptions.py

def __init__(self, message: str, *, hint: str | None = None) -> None:
    self.message = message
    self.hint = hint
    super().__init__(f"{message}\n\nHint: {hint}" if hint else message)

RLS Layer¶

Generic PostgreSQL Row-Level Security primitives. This layer has zero imports from tenants/.

GUC Helpers¶

django_rls_tenants.rls.guc.set_guc ¶

set_guc(
    name: str,
    value: str,
    *,
    is_local: bool = False,
    using: str = "default",
) -> None

Set a PostgreSQL session variable (GUC).

Parameters:

Name	Type	Description	Default
`name`	`str`	Variable name (e.g., `"rls.current_tenant"`).	required
`value`	`str`	Variable value as string.	required
`is_local`	`bool`	If `True`, use `SET LOCAL` (transaction-scoped). If `False`, use `set_config` (session-scoped, persists until changed).	`False`
`using`	`str`	Database alias. Default: `"default"`.	`'default'`

Raises:

Type	Description
`ValueError`	If `name` contains invalid characters.
`RuntimeError`	If `is_local=True` outside `transaction.atomic()`.

Source code in django_rls_tenants/rls/guc.py

def set_guc(
    name: str,
    value: str,
    *,
    is_local: bool = False,
    using: str = "default",
) -> None:
    """Set a PostgreSQL session variable (GUC).

    Args:
        name: Variable name (e.g., ``"rls.current_tenant"``).
        value: Variable value as string.
        is_local: If ``True``, use ``SET LOCAL`` (transaction-scoped).
            If ``False``, use ``set_config`` (session-scoped, persists until changed).
        using: Database alias. Default: ``"default"``.

    Raises:
        ValueError: If ``name`` contains invalid characters.
        RuntimeError: If ``is_local=True`` outside ``transaction.atomic()``.
    """
    _validate_guc_name(name)
    conn = connections[using]
    if is_local and not conn.in_atomic_block:
        raise RuntimeError(
            f"Cannot use SET LOCAL for '{name}' outside a transaction. "
            f"Wrap your code in transaction.atomic() or use is_local=False."
        )
    with conn.cursor() as cursor:
        if is_local:
            # SET LOCAL is transaction-scoped; auto-clears at commit/rollback.
            # The variable name is developer-controlled, not user input.
            cursor.execute(f"SET LOCAL {name} TO %s", [value])
        else:
            cursor.execute("SELECT set_config(%s, %s, false)", [name, value])

django_rls_tenants.rls.guc.get_guc ¶

get_guc(name: str, *, using: str = 'default') -> str | None

Get a PostgreSQL session variable value.

Returns:

Type	Description
`str \| None`	The variable value, or `None` if unset or empty.

Raises:

Type	Description
`ValueError`	If `name` contains invalid characters.

Source code in django_rls_tenants/rls/guc.py

def get_guc(name: str, *, using: str = "default") -> str | None:
    """Get a PostgreSQL session variable value.

    Returns:
        The variable value, or ``None`` if unset or empty.

    Raises:
        ValueError: If ``name`` contains invalid characters.
    """
    _validate_guc_name(name)
    conn = connections[using]
    with conn.cursor() as cursor:
        cursor.execute("SELECT current_setting(%s, true)", [name])
        result = cursor.fetchone()
        value = result[0] if result else None
        return value if value != "" else None

django_rls_tenants.rls.guc.clear_guc ¶

clear_guc(
    name: str,
    *,
    is_local: bool = False,
    using: str = "default",
) -> None

Clear a GUC variable by setting it to an empty string.

Parameters:

Name	Type	Description	Default
`name`	`str`	Variable name to clear.	required
`is_local`	`bool`	If `True`, use `SET LOCAL` (transaction-scoped).	`False`
`using`	`str`	Database alias. Default: `"default"`.	`'default'`

Source code in django_rls_tenants/rls/guc.py

def clear_guc(name: str, *, is_local: bool = False, using: str = "default") -> None:
    """Clear a GUC variable by setting it to an empty string.

    Args:
        name: Variable name to clear.
        is_local: If ``True``, use ``SET LOCAL`` (transaction-scoped).
        using: Database alias. Default: ``"default"``.
    """
    set_guc(name, "", is_local=is_local, using=using)

django_rls_tenants.rls.guc.restore_guc ¶

restore_guc(
    name: str,
    previous: str | None,
    *,
    is_local: bool = False,
    using: str = "default",
) -> None

Put a GUC back to a value previously read with :func:get_guc.

None (unset or empty before) clears the variable. is_local must match the scope the value was changed with: a SET LOCAL change is undone with another SET LOCAL so the outer transaction-local value is reinstated, while a session-level change is undone with set_config.

Parameters:

Name	Type	Description	Default
`name`	`str`	Variable name to restore.	required
`previous`	`str \| None`	Value returned by `get_guc` before the change.	required
`is_local`	`bool`	If `True`, use `SET LOCAL` (transaction-scoped).	`False`
`using`	`str`	Database alias. Default: `"default"`.	`'default'`

Source code in django_rls_tenants/rls/guc.py

def restore_guc(
    name: str,
    previous: str | None,
    *,
    is_local: bool = False,
    using: str = "default",
) -> None:
    """Put a GUC back to a value previously read with :func:`get_guc`.

    ``None`` (unset or empty before) clears the variable. ``is_local`` must
    match the scope the value was changed with: a ``SET LOCAL`` change is
    undone with another ``SET LOCAL`` so the outer transaction-local value
    is reinstated, while a session-level change is undone with ``set_config``.

    Args:
        name: Variable name to restore.
        previous: Value returned by ``get_guc`` before the change.
        is_local: If ``True``, use ``SET LOCAL`` (transaction-scoped).
        using: Database alias. Default: ``"default"``.
    """
    if previous is not None:
        set_guc(name, previous, is_local=is_local, using=using)
    else:
        clear_guc(name, is_local=is_local, using=using)

Async GUC Helpers¶

Each runs the sync helper in one thread_sensitive sync_to_async hop. Django has no asynchronous database cursor, so these are not non-blocking calls; see Async Code.

django_rls_tenants.rls.guc.aset_guc `async` ¶

aset_guc(
    name: str,
    value: str,
    *,
    is_local: bool = False,
    using: str = "default",
) -> None

Async counterpart of :func:set_guc.

Runs set_guc in a thread_sensitive sync_to_async hop. Django has no asynchronous database cursor, so this is not a non-blocking call: it executes on the thread that owns the current task's database connection (under ASGI, the request's thread-sensitive thread, shared with the async ORM), which is what keeps the GUC on the same connection the queries use.

Parameters:

Name	Type	Description	Default
`name`	`str`	Variable name (e.g., `"rls.current_tenant"`).	required
`value`	`str`	Variable value as string.	required
`is_local`	`bool`	If `True`, use `SET LOCAL` (transaction-scoped). If `False`, use `set_config` (session-scoped, persists until changed).	`False`
`using`	`str`	Database alias. Default: `"default"`.	`'default'`

Raises:

Type	Description
`ValueError`	If `name` contains invalid characters.
`RuntimeError`	If `is_local=True` and the hop thread is not inside `transaction.atomic()`. A coroutine cannot open that transaction itself (`atomic()` on the event-loop thread raises `SynchronousOnlyOperation`); it has to be open in the synchronous code that called the coroutine through `async_to_sync`.

Source code in django_rls_tenants/rls/guc.py

async def aset_guc(
    name: str,
    value: str,
    *,
    is_local: bool = False,
    using: str = "default",
) -> None:
    """Async counterpart of :func:`set_guc`.

    Runs ``set_guc`` in a ``thread_sensitive`` ``sync_to_async`` hop. Django
    has no asynchronous database cursor, so this is not a non-blocking call:
    it executes on the thread that owns the current task's database
    connection (under ASGI, the request's thread-sensitive thread, shared
    with the async ORM), which is what keeps the GUC on the same connection
    the queries use.

    Args:
        name: Variable name (e.g., ``"rls.current_tenant"``).
        value: Variable value as string.
        is_local: If ``True``, use ``SET LOCAL`` (transaction-scoped).
            If ``False``, use ``set_config`` (session-scoped, persists until changed).
        using: Database alias. Default: ``"default"``.

    Raises:
        ValueError: If ``name`` contains invalid characters.
        RuntimeError: If ``is_local=True`` and the hop thread is not inside
            ``transaction.atomic()``. A coroutine cannot open that transaction
            itself (``atomic()`` on the event-loop thread raises
            ``SynchronousOnlyOperation``); it has to be open in the synchronous
            code that called the coroutine through ``async_to_sync``.
    """
    await sync_to_async(set_guc, thread_sensitive=True)(
        name,
        value,
        is_local=is_local,
        using=using,
    )

django_rls_tenants.rls.guc.aget_guc `async` ¶

aget_guc(
    name: str, *, using: str = "default"
) -> str | None

Async counterpart of :func:get_guc.

Runs get_guc in a thread_sensitive sync_to_async hop on the thread that owns the current task's database connection (see :func:aset_guc; this is not a non-blocking call).

Parameters:

Name	Type	Description	Default
`name`	`str`	Variable name (e.g., `"rls.current_tenant"`).	required
`using`	`str`	Database alias. Default: `"default"`.	`'default'`

Returns:

Type	Description
`str \| None`	The variable value, or `None` if unset or empty.

Raises:

Type	Description
`ValueError`	If `name` contains invalid characters.

Source code in django_rls_tenants/rls/guc.py

async def aget_guc(name: str, *, using: str = "default") -> str | None:
    """Async counterpart of :func:`get_guc`.

    Runs ``get_guc`` in a ``thread_sensitive`` ``sync_to_async`` hop on the
    thread that owns the current task's database connection (see
    :func:`aset_guc`; this is not a non-blocking call).

    Args:
        name: Variable name (e.g., ``"rls.current_tenant"``).
        using: Database alias. Default: ``"default"``.

    Returns:
        The variable value, or ``None`` if unset or empty.

    Raises:
        ValueError: If ``name`` contains invalid characters.
    """
    return await sync_to_async(get_guc, thread_sensitive=True)(name, using=using)

django_rls_tenants.rls.guc.aclear_guc `async` ¶

aclear_guc(
    name: str,
    *,
    is_local: bool = False,
    using: str = "default",
) -> None

Async counterpart of :func:clear_guc.

Runs clear_guc in a thread_sensitive sync_to_async hop on the thread that owns the current task's database connection (see :func:aset_guc; this is not a non-blocking call).

Parameters:

Name	Type	Description	Default
`name`	`str`	Variable name to clear.	required
`is_local`	`bool`	If `True`, use `SET LOCAL` (transaction-scoped).	`False`
`using`	`str`	Database alias. Default: `"default"`.	`'default'`

Raises:

Type	Description
`ValueError`	If `name` contains invalid characters.
`RuntimeError`	If `is_local=True` and the hop thread is not inside `transaction.atomic()` (see :func:`aset_guc`).

Source code in django_rls_tenants/rls/guc.py

async def aclear_guc(name: str, *, is_local: bool = False, using: str = "default") -> None:
    """Async counterpart of :func:`clear_guc`.

    Runs ``clear_guc`` in a ``thread_sensitive`` ``sync_to_async`` hop on the
    thread that owns the current task's database connection (see
    :func:`aset_guc`; this is not a non-blocking call).

    Args:
        name: Variable name to clear.
        is_local: If ``True``, use ``SET LOCAL`` (transaction-scoped).
        using: Database alias. Default: ``"default"``.

    Raises:
        ValueError: If ``name`` contains invalid characters.
        RuntimeError: If ``is_local=True`` and the hop thread is not inside
            ``transaction.atomic()`` (see :func:`aset_guc`).
    """
    await sync_to_async(clear_guc, thread_sensitive=True)(name, is_local=is_local, using=using)

django_rls_tenants.rls.guc.arestore_guc `async` ¶

arestore_guc(
    name: str,
    previous: str | None,
    *,
    is_local: bool = False,
    using: str = "default",
) -> None

Async counterpart of :func:restore_guc.

Runs restore_guc in a thread_sensitive sync_to_async hop on the thread that owns the current task's database connection (see :func:aset_guc; this is not a non-blocking call).

Parameters:

Name	Type	Description	Default
`name`	`str`	Variable name to restore.	required
`previous`	`str \| None`	Value returned by `get_guc` / `aget_guc` before the change.	required
`is_local`	`bool`	If `True`, use `SET LOCAL` (transaction-scoped).	`False`
`using`	`str`	Database alias. Default: `"default"`.	`'default'`

Raises:

Type	Description
`ValueError`	If `name` contains invalid characters.
`RuntimeError`	If `is_local=True` and the hop thread is not inside `transaction.atomic()` (see :func:`aset_guc`).

Source code in django_rls_tenants/rls/guc.py

async def arestore_guc(
    name: str,
    previous: str | None,
    *,
    is_local: bool = False,
    using: str = "default",
) -> None:
    """Async counterpart of :func:`restore_guc`.

    Runs ``restore_guc`` in a ``thread_sensitive`` ``sync_to_async`` hop on
    the thread that owns the current task's database connection (see
    :func:`aset_guc`; this is not a non-blocking call).

    Args:
        name: Variable name to restore.
        previous: Value returned by ``get_guc`` / ``aget_guc`` before the change.
        is_local: If ``True``, use ``SET LOCAL`` (transaction-scoped).
        using: Database alias. Default: ``"default"``.

    Raises:
        ValueError: If ``name`` contains invalid characters.
        RuntimeError: If ``is_local=True`` and the hop thread is not inside
            ``transaction.atomic()`` (see :func:`aset_guc`).
    """
    await sync_to_async(restore_guc, thread_sensitive=True)(
        name,
        previous,
        is_local=is_local,
        using=using,
    )

Constraints¶

django_rls_tenants.rls.constraints.RLSConstraint ¶

RLSConstraint(
    *,
    field: str,
    name: str,
    guc_tenant_var: str = "rls.current_tenant",
    guc_admin_var: str = "rls.is_admin",
    tenant_pk_type: str = "int",
    extra_bypass_flags: list[str] | None = None,
)

Bases: BaseConstraint

Django constraint that generates PostgreSQL RLS policies during migrations.

When Django applies a migration containing this constraint, it:

Enables RLS on the table (ALTER TABLE ... ENABLE ROW LEVEL SECURITY).
Forces RLS for the table owner (ALTER TABLE ... FORCE ROW LEVEL SECURITY).
Creates an isolation policy with configurable USING and WITH CHECK.

Parameters:

Name	Type	Description	Default
`field`	`str`	FK field name for tenant identification (e.g., `"tenant"`). The policy checks `{field}_id` against the GUC variable.	required
`name`	`str`	Constraint name. Supports `%(app_label)s` and `%(class)s`.	required
`guc_tenant_var`	`str`	GUC variable holding the current tenant ID. Default: `"rls.current_tenant"`.	`'rls.current_tenant'`
`guc_admin_var`	`str`	GUC variable for admin bypass. Default: `"rls.is_admin"`.	`'rls.is_admin'`
`tenant_pk_type`	`str`	SQL cast type for tenant PK. Default: `"int"`. Options: `"int"`, `"uuid"`, `"bigint"`.	`'int'`
`extra_bypass_flags`	`list[str] \| None`	Additional GUC variables that bypass the `USING` clause (but NOT `WITH CHECK`). Useful for auth edge cases.	`None`

Source code in django_rls_tenants/rls/constraints.py

def __init__(
    self,
    *,
    field: str,
    name: str,
    guc_tenant_var: str = "rls.current_tenant",
    guc_admin_var: str = "rls.is_admin",
    tenant_pk_type: str = "int",
    extra_bypass_flags: list[str] | None = None,
) -> None:
    super().__init__(name=name)
    _validate_field_name(field)
    _validate_pk_type(tenant_pk_type)
    _validate_guc_name_for_ddl(guc_tenant_var, "guc_tenant_var")
    _validate_guc_name_for_ddl(guc_admin_var, "guc_admin_var")
    for flag in extra_bypass_flags or []:
        _validate_guc_name_for_ddl(flag, "extra_bypass_flags")
    self.field = field
    self.guc_tenant_var = guc_tenant_var
    self.guc_admin_var = guc_admin_var
    self.tenant_pk_type = tenant_pk_type
    self.extra_bypass_flags = extra_bypass_flags or []

constraint_sql ¶

constraint_sql(
    model: type[Model],
    schema_editor: BaseDatabaseSchemaEditor,
) -> str

No inline constraint SQL; defer RLS DDL to after CREATE TABLE.

Django calls constraint_sql during CREATE TABLE for inline constraints. RLS policies require the table to exist first, so we defer the actual DDL and return an empty string (filtered out by Django's if statement guard).

Source code in django_rls_tenants/rls/constraints.py

def constraint_sql(  # type: ignore[override]
    self,
    model: type[Model],
    schema_editor: BaseDatabaseSchemaEditor,
) -> str:
    """No inline constraint SQL; defer RLS DDL to after ``CREATE TABLE``.

    Django calls ``constraint_sql`` during ``CREATE TABLE`` for inline
    constraints.  RLS policies require the table to exist first, so we
    defer the actual DDL and return an empty string (filtered out by
    Django's ``if statement`` guard).
    """
    schema_editor.deferred_sql.append(self.create_sql(model, schema_editor))
    return ""

create_sql ¶

create_sql(
    model: type[Model] | None,
    schema_editor: BaseDatabaseSchemaEditor | None,
) -> Statement

Generate SQL to enable RLS and create the isolation policy.

Source code in django_rls_tenants/rls/constraints.py

def create_sql(  # type: ignore[override]
    self,
    model: type[Model] | None,
    schema_editor: BaseDatabaseSchemaEditor | None,  # noqa: ARG002  -- required by BaseConstraint
) -> Statement:
    """Generate SQL to enable RLS and create the isolation policy."""
    table = model._meta.db_table  # type: ignore[union-attr]  # noqa: SLF001  -- Django's standard public API
    policy_name = f"{table}_tenant_isolation_policy"

    tenant_match = tenant_match_sql(
        f"{self.field}_id", self.guc_tenant_var, self.tenant_pk_type
    )
    admin_check = bool_flag_sql(self.guc_admin_var)

    # Build bypass conditions for CASE WHEN: admin is always first,
    # extra bypass flags are appended (USING only, NOT WITH CHECK).
    bypass_conditions_using = [admin_check]
    bypass_conditions_using.extend(bool_flag_sql(flag) for flag in self.extra_bypass_flags)

    bypass_clause_using = "\n                              OR ".join(bypass_conditions_using)
    bypass_clause_check = admin_check  # only admin in WITH CHECK

    # Safety: all interpolated values come from model._meta (developer-defined)
    # and constructor args, not from user input. SQL injection is not possible.
    return Statement(
        template="%(sql)s",
        sql=f"""
            DO $$
            BEGIN
                IF NOT EXISTS (
                    SELECT 1 FROM pg_policies
                    WHERE policyname = '{policy_name}'
                    AND tablename = '{table}'
                    AND schemaname = current_schema()
                ) THEN
                    EXECUTE $BODY$
                        ALTER TABLE "{table}" ENABLE ROW LEVEL SECURITY;
                        ALTER TABLE "{table}" FORCE ROW LEVEL SECURITY;
                        CREATE POLICY "{policy_name}"
                        ON "{table}"
                        USING (
                            CASE WHEN {bypass_clause_using}
                                 THEN true
                                 ELSE {tenant_match}
                            END
                        )
                        WITH CHECK (
                            CASE WHEN {bypass_clause_check}
                                 THEN true
                                 ELSE {tenant_match}
                            END
                        );
                    $BODY$;
                END IF;
            END
            $$;
        """,
    )

remove_sql ¶

remove_sql(
    model: type[Model] | None,
    schema_editor: BaseDatabaseSchemaEditor | None,
) -> Statement

Generate SQL to drop the policy and disable RLS.

Source code in django_rls_tenants/rls/constraints.py

def remove_sql(  # type: ignore[override]
    self,
    model: type[Model] | None,
    schema_editor: BaseDatabaseSchemaEditor | None,  # noqa: ARG002  -- required by BaseConstraint
) -> Statement:
    """Generate SQL to drop the policy and disable RLS."""
    table = model._meta.db_table  # type: ignore[union-attr]  # noqa: SLF001  -- Django's standard public API
    policy_name = f"{table}_tenant_isolation_policy"
    # Safety: values come from model._meta, not user input.
    return Statement(
        template="%(sql)s",
        sql=f"""
        DROP POLICY IF EXISTS "{policy_name}" ON "{table}";
        ALTER TABLE "{table}" NO FORCE ROW LEVEL SECURITY;
        ALTER TABLE "{table}" DISABLE ROW LEVEL SECURITY;
        """,
    )

validate ¶

validate(
    model: type[Model],
    instance: Any,
    exclude: Any = None,
    using: str | None = None,
) -> None

No-op: RLS is enforced at the database level, not in Django validation.

Source code in django_rls_tenants/rls/constraints.py

def validate(
    self,
    model: type[Model],
    instance: Any,
    exclude: Any = None,
    using: str | None = None,
) -> None:
    """No-op: RLS is enforced at the database level, not in Django validation."""

deconstruct ¶

deconstruct() -> tuple[str, tuple[()], dict[str, Any]]

Return a 3-tuple for Django's migration serializer.

Source code in django_rls_tenants/rls/constraints.py

def deconstruct(self) -> tuple[str, tuple[()], dict[str, Any]]:
    """Return a 3-tuple for Django's migration serializer."""
    path, _, kwargs = super().deconstruct()
    kwargs["field"] = self.field
    if self.guc_tenant_var != "rls.current_tenant":
        kwargs["guc_tenant_var"] = self.guc_tenant_var
    if self.guc_admin_var != "rls.is_admin":
        kwargs["guc_admin_var"] = self.guc_admin_var
    if self.tenant_pk_type != "int":
        kwargs["tenant_pk_type"] = self.tenant_pk_type
    if self.extra_bypass_flags:
        kwargs["extra_bypass_flags"] = self.extra_bypass_flags
    return (path, (), kwargs)

django_rls_tenants.rls.constraints.RLSM2MConstraint ¶

RLSM2MConstraint(
    *,
    name: str,
    from_model: str,
    to_model: str,
    from_fk: str,
    to_fk: str,
    from_tenant_fk: str | None = "tenant",
    to_tenant_fk: str | None = "tenant",
    guc_tenant_var: str = "rls.current_tenant",
    guc_admin_var: str = "rls.is_admin",
    tenant_pk_type: str = "int",
)

Bases: BaseConstraint

Constraint generating subquery-based RLS policies for M2M join tables.

Unlike RLSConstraint which checks a direct {field}_id column, this generates policies that verify both FK references point to rows belonging to the current tenant via EXISTS (SELECT 1 ...) subqueries. EXISTS is used rather than IN (SELECT ...) because it gives the planner more optimisation freedom; see _build_m2m_conditions.

For join tables where only one side is RLS-protected, only that side's FK is checked. For both sides protected, both FKs are checked.

Parameters:

Name	Type	Description	Default
`name`	`str`	Constraint name.	required
`from_model`	`str`	Dotted path to the "from" model (e.g., `"myapp.Project"`).	required
`to_model`	`str`	Dotted path to the "to" model (e.g., `"myapp.User"`).	required
`from_fk`	`str`	FK column name for the "from" side (e.g., `"project_id"`).	required
`to_fk`	`str`	FK column name for the "to" side (e.g., `"user_id"`).	required
`from_tenant_fk`	`str \| None`	Tenant FK field name on the "from" model, or `None` if the "from" side is not RLS-protected. Default: `"tenant"`.	`'tenant'`
`to_tenant_fk`	`str \| None`	Tenant FK field name on the "to" model, or `None` if the "to" side is not RLS-protected. Default: `"tenant"`.	`'tenant'`
`guc_tenant_var`	`str`	GUC variable for current tenant. Default: `"rls.current_tenant"`.	`'rls.current_tenant'`
`guc_admin_var`	`str`	GUC variable for admin bypass. Default: `"rls.is_admin"`.	`'rls.is_admin'`
`tenant_pk_type`	`str`	SQL cast type for tenant PK. Default: `"int"`.	`'int'`

Source code in django_rls_tenants/rls/constraints.py

def __init__(
    self,
    *,
    name: str,
    from_model: str,
    to_model: str,
    from_fk: str,
    to_fk: str,
    from_tenant_fk: str | None = "tenant",
    to_tenant_fk: str | None = "tenant",
    guc_tenant_var: str = "rls.current_tenant",
    guc_admin_var: str = "rls.is_admin",
    tenant_pk_type: str = "int",
) -> None:
    super().__init__(name=name)
    _validate_model_path(from_model, "from_model")
    _validate_model_path(to_model, "to_model")
    _validate_field_name(from_fk, "from_fk")
    _validate_field_name(to_fk, "to_fk")
    if from_tenant_fk is not None:
        _validate_field_name(from_tenant_fk, "from_tenant_fk")
    if to_tenant_fk is not None:
        _validate_field_name(to_tenant_fk, "to_tenant_fk")
    if from_tenant_fk is None and to_tenant_fk is None:
        msg = (
            "At least one side of the M2M relationship must be "
            "RLS-protected (from_tenant_fk or to_tenant_fk must be set)."
        )
        raise ValueError(msg)
    _validate_pk_type(tenant_pk_type)
    _validate_guc_name_for_ddl(guc_tenant_var, "guc_tenant_var")
    _validate_guc_name_for_ddl(guc_admin_var, "guc_admin_var")
    self.from_model = from_model
    self.to_model = to_model
    self.from_fk = from_fk
    self.to_fk = to_fk
    self.from_tenant_fk = from_tenant_fk
    self.to_tenant_fk = to_tenant_fk
    self.guc_tenant_var = guc_tenant_var
    self.guc_admin_var = guc_admin_var
    self.tenant_pk_type = tenant_pk_type

constraint_sql ¶

constraint_sql(
    model: type[Model],
    schema_editor: BaseDatabaseSchemaEditor,
) -> str

No inline constraint SQL; defer RLS DDL to after CREATE TABLE.

Source code in django_rls_tenants/rls/constraints.py

def constraint_sql(  # type: ignore[override]
    self,
    model: type[Model],
    schema_editor: BaseDatabaseSchemaEditor,
) -> str:
    """No inline constraint SQL; defer RLS DDL to after ``CREATE TABLE``."""
    schema_editor.deferred_sql.append(self.create_sql(model, schema_editor))
    return ""

create_sql ¶

create_sql(
    model: type[Model] | None,
    schema_editor: BaseDatabaseSchemaEditor | None,
) -> Statement

Generate SQL to enable RLS and create the M2M isolation policy.

Source code in django_rls_tenants/rls/constraints.py

def create_sql(  # type: ignore[override]
    self,
    model: type[Model] | None,
    schema_editor: BaseDatabaseSchemaEditor | None,  # noqa: ARG002  -- required by BaseConstraint
) -> Statement:
    """Generate SQL to enable RLS and create the M2M isolation policy."""
    table = model._meta.db_table  # type: ignore[union-attr]  # noqa: SLF001
    admin_check = bool_flag_sql(self.guc_admin_var)
    subquery_clause = self._build_subquery_clause()
    return Statement(
        template="%(sql)s",
        sql=_build_m2m_create_sql(
            table=table, admin_check=admin_check, subquery_clause=subquery_clause
        ),
    )

remove_sql ¶

remove_sql(
    model: type[Model] | None,
    schema_editor: BaseDatabaseSchemaEditor | None,
) -> Statement

Generate SQL to drop the M2M policy and disable RLS.

Source code in django_rls_tenants/rls/constraints.py

def remove_sql(  # type: ignore[override]
    self,
    model: type[Model] | None,
    schema_editor: BaseDatabaseSchemaEditor | None,  # noqa: ARG002  -- required by BaseConstraint
) -> Statement:
    """Generate SQL to drop the M2M policy and disable RLS."""
    table = model._meta.db_table  # type: ignore[union-attr]  # noqa: SLF001
    return Statement(
        template="%(sql)s",
        sql=_build_m2m_drop_sql(table=table),
    )

validate ¶

validate(
    model: type[Model],
    instance: Any,
    exclude: Any = None,
    using: str | None = None,
) -> None

No-op: RLS is enforced at the database level, not in Django validation.

Source code in django_rls_tenants/rls/constraints.py

def validate(
    self,
    model: type[Model],
    instance: Any,
    exclude: Any = None,
    using: str | None = None,
) -> None:
    """No-op: RLS is enforced at the database level, not in Django validation."""

deconstruct ¶

deconstruct() -> tuple[str, tuple[()], dict[str, Any]]

Return a 3-tuple for Django's migration serializer.

Source code in django_rls_tenants/rls/constraints.py

def deconstruct(self) -> tuple[str, tuple[()], dict[str, Any]]:
    """Return a 3-tuple for Django's migration serializer."""
    path, _, kwargs = super().deconstruct()
    kwargs["from_model"] = self.from_model
    kwargs["to_model"] = self.to_model
    kwargs["from_fk"] = self.from_fk
    kwargs["to_fk"] = self.to_fk
    if self.from_tenant_fk != "tenant":
        kwargs["from_tenant_fk"] = self.from_tenant_fk
    if self.to_tenant_fk != "tenant":
        kwargs["to_tenant_fk"] = self.to_tenant_fk
    if self.guc_tenant_var != "rls.current_tenant":
        kwargs["guc_tenant_var"] = self.guc_tenant_var
    if self.guc_admin_var != "rls.is_admin":
        kwargs["guc_admin_var"] = self.guc_admin_var
    if self.tenant_pk_type != "int":
        kwargs["tenant_pk_type"] = self.tenant_pk_type
    return (path, (), kwargs)

Migration Operations¶

django_rls_tenants.operations.AddM2MRLSPolicy ¶

AddM2MRLSPolicy(
    m2m_table: str,
    from_model: str,
    to_model: str,
    from_fk: str,
    to_fk: str,
    from_tenant_fk: str | None = "tenant",
    to_tenant_fk: str | None = "tenant",
    guc_tenant_var: str = "rls.current_tenant",
    guc_admin_var: str = "rls.is_admin",
    tenant_pk_type: str = "int",
)

Bases: Operation

Migration operation to add an RLS policy to an M2M through table.

Generates and executes subquery-based CREATE POLICY SQL that checks both FK sides of the M2M join table belong to the current tenant. Supports tables where only one side is RLS-protected.

This operation is reversible: database_backwards drops the policy and disables RLS on the table.

Parameters:

Name	Type	Description	Default
`m2m_table`	`str`	The database table name of the M2M through table.	required
`from_model`	`str`	Dotted path to the "from" model (e.g., `"myapp.Project"`).	required
`to_model`	`str`	Dotted path to the "to" model (e.g., `"myapp.User"`).	required
`from_fk`	`str`	FK column on the through table for the "from" side.	required
`to_fk`	`str`	FK column on the through table for the "to" side.	required
`from_tenant_fk`	`str \| None`	Tenant FK on the "from" model, or `None`. Default: `"tenant"`.	`'tenant'`
`to_tenant_fk`	`str \| None`	Tenant FK on the "to" model, or `None`. Default: `"tenant"`.	`'tenant'`
`guc_tenant_var`	`str`	GUC variable for current tenant. Default: `"rls.current_tenant"`.	`'rls.current_tenant'`
`guc_admin_var`	`str`	GUC variable for admin bypass. Default: `"rls.is_admin"`.	`'rls.is_admin'`
`tenant_pk_type`	`str`	SQL cast type for tenant PK. Default: `"int"`.	`'int'`

Source code in django_rls_tenants/operations.py

def __init__(
    self,
    m2m_table: str,
    from_model: str,
    to_model: str,
    from_fk: str,
    to_fk: str,
    from_tenant_fk: str | None = "tenant",
    to_tenant_fk: str | None = "tenant",
    guc_tenant_var: str = "rls.current_tenant",
    guc_admin_var: str = "rls.is_admin",
    tenant_pk_type: str = "int",
) -> None:
    _validate_field_name(m2m_table, "m2m_table")
    _validate_model_path(from_model, "from_model")
    _validate_model_path(to_model, "to_model")
    _validate_field_name(from_fk, "from_fk")
    _validate_field_name(to_fk, "to_fk")
    if from_tenant_fk is not None:
        _validate_field_name(from_tenant_fk, "from_tenant_fk")
    if to_tenant_fk is not None:
        _validate_field_name(to_tenant_fk, "to_tenant_fk")
    _validate_pk_type(tenant_pk_type)
    _validate_guc_name_for_ddl(guc_tenant_var, "guc_tenant_var")
    _validate_guc_name_for_ddl(guc_admin_var, "guc_admin_var")
    self.m2m_table = m2m_table
    self.from_model = from_model
    self.to_model = to_model
    self.from_fk = from_fk
    self.to_fk = to_fk
    self.from_tenant_fk = from_tenant_fk
    self.to_tenant_fk = to_tenant_fk
    self.guc_tenant_var = guc_tenant_var
    self.guc_admin_var = guc_admin_var
    self.tenant_pk_type = tenant_pk_type

state_forwards ¶

state_forwards(app_label: str, state: ProjectState) -> None

No model state changes -- this is a database-only operation.

Source code in django_rls_tenants/operations.py

def state_forwards(
    self,
    app_label: str,
    state: ProjectState,
) -> None:
    """No model state changes -- this is a database-only operation."""

database_forwards ¶

database_forwards(
    app_label: str,
    schema_editor: BaseDatabaseSchemaEditor,
    from_state: ProjectState,
    to_state: ProjectState,
) -> None

Create the M2M RLS policy.

Source code in django_rls_tenants/operations.py

def database_forwards(
    self,
    app_label: str,  # noqa: ARG002
    schema_editor: BaseDatabaseSchemaEditor,
    from_state: ProjectState,  # noqa: ARG002
    to_state: ProjectState,
) -> None:
    """Create the M2M RLS policy."""
    table = self.m2m_table
    admin_check = bool_flag_sql(self.guc_admin_var)
    subquery_clause = _build_m2m_conditions(
        from_fk=self.from_fk,
        from_table=self._resolve_table(self.from_model, to_state.apps),
        from_tenant_fk=self.from_tenant_fk,
        to_fk=self.to_fk,
        to_table=self._resolve_table(self.to_model, to_state.apps),
        to_tenant_fk=self.to_tenant_fk,
        guc_tenant_var=self.guc_tenant_var,
        tenant_pk_type=self.tenant_pk_type,
    )
    sql = _build_m2m_create_sql(
        table=table, admin_check=admin_check, subquery_clause=subquery_clause
    )
    schema_editor.execute(sql)

database_backwards ¶

database_backwards(
    app_label: str,
    schema_editor: BaseDatabaseSchemaEditor,
    from_state: ProjectState,
    to_state: ProjectState,
) -> None

Drop the M2M RLS policy and disable RLS.

Source code in django_rls_tenants/operations.py

def database_backwards(
    self,
    app_label: str,  # noqa: ARG002
    schema_editor: BaseDatabaseSchemaEditor,
    from_state: ProjectState,  # noqa: ARG002
    to_state: ProjectState,  # noqa: ARG002
) -> None:
    """Drop the M2M RLS policy and disable RLS."""
    schema_editor.execute(_build_m2m_drop_sql(table=self.m2m_table))

describe ¶

describe() -> str

Return a human-readable description.

Source code in django_rls_tenants/operations.py

def describe(self) -> str:
    """Return a human-readable description."""
    return f"Add M2M RLS policy to {self.m2m_table}"

deconstruct ¶

deconstruct() -> tuple[str, list[Any], dict[str, Any]]

Return args for Django's migration serializer.

Source code in django_rls_tenants/operations.py

def deconstruct(self) -> tuple[str, list[Any], dict[str, Any]]:
    """Return args for Django's migration serializer."""
    kwargs: dict[str, Any] = {
        "m2m_table": self.m2m_table,
        "from_model": self.from_model,
        "to_model": self.to_model,
        "from_fk": self.from_fk,
        "to_fk": self.to_fk,
    }
    if self.from_tenant_fk != "tenant":
        kwargs["from_tenant_fk"] = self.from_tenant_fk
    if self.to_tenant_fk != "tenant":
        kwargs["to_tenant_fk"] = self.to_tenant_fk
    if self.guc_tenant_var != "rls.current_tenant":
        kwargs["guc_tenant_var"] = self.guc_tenant_var
    if self.guc_admin_var != "rls.is_admin":
        kwargs["guc_admin_var"] = self.guc_admin_var
    if self.tenant_pk_type != "int":
        kwargs["tenant_pk_type"] = self.tenant_pk_type
    return (
        f"{self.__class__.__module__}.{self.__class__.__qualname__}",
        [],
        kwargs,
    )

Context Managers¶

django_rls_tenants.rls.context.rls_context ¶

rls_context(
    variables: dict[str, str],
    *,
    is_local: bool = False,
    using: str = "default",
) -> Iterator[None]

Set multiple GUC variables for the duration of a block.

Saves and restores previous values on exit (supports nesting) in both GUC modes. With is_local=True the previous values are re-issued with SET LOCAL so an enclosing block keeps its scope until the transaction ends.

Parameters:

Name	Type	Description	Default
`variables`	`dict[str, str]`	Dict of GUC variable names to values.	required
`is_local`	`bool`	If `True`, use `SET LOCAL` (transaction-scoped).	`False`
`using`	`str`	Database alias. Default: `"default"`.	`'default'`

Source code in django_rls_tenants/rls/context.py

@contextmanager
def rls_context(
    variables: dict[str, str],
    *,
    is_local: bool = False,
    using: str = "default",
) -> Iterator[None]:
    """Set multiple GUC variables for the duration of a block.

    Saves and restores previous values on exit (supports nesting) in both
    GUC modes. With ``is_local=True`` the previous values are re-issued with
    ``SET LOCAL`` so an enclosing block keeps its scope until the
    transaction ends.

    Args:
        variables: Dict of GUC variable names to values.
        is_local: If ``True``, use ``SET LOCAL`` (transaction-scoped).
        using: Database alias. Default: ``"default"``.
    """
    previous_values: dict[str, str | None] = {
        name: get_guc(name, using=using) for name in variables
    }

    exc_pending = False
    try:
        for name, value in variables.items():
            set_guc(name, value, is_local=is_local, using=using)
        yield
    except BaseException:
        exc_pending = True
        raise
    finally:
        restore_gucs(previous_values, is_local=is_local, using=using, exc_pending=exc_pending)

django_rls_tenants.rls.context.bypass_flag ¶

bypass_flag(
    flag_name: str,
    *,
    is_local: bool = False,
    using: str = "default",
) -> Iterator[None]

Temporarily set a GUC bypass flag to 'true'.

Saves and restores the previous value on exit (supports nesting) in both GUC modes.

Usage::

with bypass_flag("rls.is_login_request"):
    user = User.objects.get(email=email)

Source code in django_rls_tenants/rls/context.py

@contextmanager
def bypass_flag(
    flag_name: str,
    *,
    is_local: bool = False,
    using: str = "default",
) -> Iterator[None]:
    """Temporarily set a GUC bypass flag to ``'true'``.

    Saves and restores the previous value on exit (supports nesting) in
    both GUC modes.

    Usage::

        with bypass_flag("rls.is_login_request"):
            user = User.objects.get(email=email)
    """
    previous = get_guc(flag_name, using=using)

    set_guc(flag_name, "true", is_local=is_local, using=using)
    exc_pending = False
    try:
        yield
    except BaseException:
        exc_pending = True
        raise
    finally:
        restore_gucs(
            {flag_name: previous},
            is_local=is_local,
            using=using,
            exc_pending=exc_pending,
        )

django_rls_tenants.rls.context.async_rls_context `async` ¶

async_rls_context(
    variables: dict[str, str],
    *,
    is_local: bool = False,
    using: str = "default",
) -> AsyncIterator[None]

Async counterpart of :func:rls_context, for async with.

Snapshots and sets the variables in one thread_sensitive sync_to_async hop on entry and restores them in one hop on exit (see :func:django_rls_tenants.rls.guc.aset_guc for what a hop is and why it is not a non-blocking call). Nesting, the restore in both GUC modes and the aborted-transaction behaviour are the same as for :func:rls_context.

The restore runs even when the task is cancelled while entering or inside the block. Only a second cancellation delivered during the restore itself can skip it; wrap long-lived tasks in your own cleanup if that matters.

Parameters:

Name	Type	Description	Default
`variables`	`dict[str, str]`	Dict of GUC variable names to values.	required
`is_local`	`bool`	If `True`, use `SET LOCAL` (transaction-scoped). The hop thread must then be inside `transaction.atomic()`, otherwise the `RuntimeError` from :func:`set_guc` propagates.	`False`
`using`	`str`	Database alias. Default: `"default"`.	`'default'`

Source code in django_rls_tenants/rls/context.py

@asynccontextmanager
async def async_rls_context(
    variables: dict[str, str],
    *,
    is_local: bool = False,
    using: str = "default",
) -> AsyncIterator[None]:
    """Async counterpart of :func:`rls_context`, for ``async with``.

    Snapshots and sets the variables in one ``thread_sensitive``
    ``sync_to_async`` hop on entry and restores them in one hop on exit (see
    :func:`django_rls_tenants.rls.guc.aset_guc` for what a hop is and why it
    is not a non-blocking call). Nesting, the restore in both GUC modes and
    the aborted-transaction behaviour are the same as for :func:`rls_context`.

    The restore runs even when the task is cancelled while entering or
    inside the block. Only a second cancellation delivered during the
    restore itself can skip it; wrap long-lived tasks in your own cleanup if
    that matters.

    Args:
        variables: Dict of GUC variable names to values.
        is_local: If ``True``, use ``SET LOCAL`` (transaction-scoped). The hop
            thread must then be inside ``transaction.atomic()``, otherwise the
            ``RuntimeError`` from :func:`set_guc` propagates.
        using: Database alias. Default: ``"default"``.
    """
    previous = await _run_enter_hop(_enter_gucs, variables, is_local=is_local, using=using)
    exc_pending = False
    try:
        yield
    except BaseException:
        exc_pending = True
        raise
    finally:
        await sync_to_async(restore_gucs, thread_sensitive=True)(
            previous,
            is_local=is_local,
            using=using,
            exc_pending=exc_pending,
        )

django_rls_tenants.rls.context.async_bypass_flag `async` ¶

async_bypass_flag(
    flag_name: str,
    *,
    is_local: bool = False,
    using: str = "default",
) -> AsyncIterator[None]

Async counterpart of :func:bypass_flag, for async with.

Sets the flag to 'true' in one thread_sensitive hop and restores the previous value in one hop on exit, with the same semantics as :func:async_rls_context.

Usage::

async with async_bypass_flag("rls.is_login_request"):
    user = await User.objects.aget(email=email)

Parameters:

Name	Type	Description	Default
`flag_name`	`str`	GUC variable name (e.g., `"rls.is_login_request"`).	required
`is_local`	`bool`	If `True`, use `SET LOCAL` (transaction-scoped).	`False`
`using`	`str`	Database alias. Default: `"default"`.	`'default'`

Source code in django_rls_tenants/rls/context.py

@asynccontextmanager
async def async_bypass_flag(
    flag_name: str,
    *,
    is_local: bool = False,
    using: str = "default",
) -> AsyncIterator[None]:
    """Async counterpart of :func:`bypass_flag`, for ``async with``.

    Sets the flag to ``'true'`` in one ``thread_sensitive`` hop and restores
    the previous value in one hop on exit, with the same semantics as
    :func:`async_rls_context`.

    Usage::

        async with async_bypass_flag("rls.is_login_request"):
            user = await User.objects.aget(email=email)

    Args:
        flag_name: GUC variable name (e.g., ``"rls.is_login_request"``).
        is_local: If ``True``, use ``SET LOCAL`` (transaction-scoped).
        using: Database alias. Default: ``"default"``.
    """
    async with async_rls_context({flag_name: "true"}, is_local=is_local, using=using):
        yield

Tenants Layer¶

Django multitenancy built on top of the rls/ primitives.

Configuration¶

django_rls_tenants.tenants.conf.RLSTenantsConfig ¶

RLSTenantsConfig()

Read library configuration from settings.RLS_TENANTS.

All settings live under a single dict::

RLS_TENANTS = {
    "TENANT_MODEL": "myapp.Tenant",       # Required
    "DATABASES": ["default"],              # Default: ["default"]
    "GUC_PREFIX": "rls",                   # Default: "rls"
    "STRICT_MODE": False,                  # Default: False
    "TENANT_FK_FIELD": "tenant",           # Default: "tenant"
    "USER_PARAM_NAME": "as_user",          # Default: "as_user"
    "TENANT_PK_TYPE": "int",               # Default: "int"
    "USE_LOCAL_SET": False,                 # Default: False
}

Source code in django_rls_tenants/tenants/conf.py

def __init__(self) -> None:
    self._config_cache: dict[str, Any] | None = None
    self._unknown_keys_checked: bool = False

TENANT_MODEL `property` ¶

TENANT_MODEL: str

Dotted path to the Tenant model (e.g., "myapp.Tenant").

GUC_PREFIX `property` ¶

GUC_PREFIX: str

Prefix for GUC variable names. Default: "rls".

GUC_CURRENT_TENANT `property` ¶

GUC_CURRENT_TENANT: str

Derived: "{prefix}.current_tenant".

GUC_IS_ADMIN `property` ¶

GUC_IS_ADMIN: str

Derived: "{prefix}.is_admin".

TENANT_FK_FIELD `property` ¶

TENANT_FK_FIELD: str

FK field name on RLSProtectedModel. Default: "tenant".

USER_PARAM_NAME `property` ¶

USER_PARAM_NAME: str

Parameter name @with_rls_context looks for. Default: "as_user".

TENANT_PK_TYPE `property` ¶

TENANT_PK_TYPE: str

SQL cast type for tenant PK. Default: "int".

USE_LOCAL_SET `property` ¶

USE_LOCAL_SET: bool

Set the GUCs transaction-locally (SET LOCAL). Default: False.

The GUC affinity guard then sets the active context's values in each transaction, giving a statement run in autocommit inside a context a transaction of its own.

STRICT_MODE `property` ¶

STRICT_MODE: bool

Raise on queries without tenant context. Default: False.

When enabled, TenantQuerySet evaluation methods raise NoTenantContextError if no RLS context is active (no tenant_context(), admin_context(), for_user(), or RLSTenantMiddleware).

DATABASES `property` ¶

DATABASES: list[str]

Database aliases that follow the active RLS context. Default: ["default"].

The middleware sets the GUCs on each of them (with USE_LOCAL_SET=True on its synchronous path only), and the GUC affinity guard applies the active context on each of them before every statement, whichever alias the context was entered on (transaction-locally with USE_LOCAL_SET=True). With USE_LOCAL_SET=True an alias outside this list still needs transaction.atomic() around a context entered on it. In multi-database setups (e.g., read replicas), add all aliases that serve RLS-protected queries::

RLS_TENANTS = {
    "DATABASES": ["default", "replica"],
}

reload ¶

reload() -> None

Discard the cached RLS_TENANTS dict so the next read hits settings.

The app config connects this to Django's setting_changed signal, so override_settings(RLS_TENANTS=...), @override_settings on a test class and pytest-django's settings fixture all take effect without touching the cache by hand. The unknown-key check runs again on the next read as well, so a typo introduced by an override is still reported.

Source code in django_rls_tenants/tenants/conf.py

def reload(self) -> None:
    """Discard the cached ``RLS_TENANTS`` dict so the next read hits ``settings``.

    The app config connects this to Django's ``setting_changed`` signal, so
    ``override_settings(RLS_TENANTS=...)``, ``@override_settings`` on a test
    class and pytest-django's ``settings`` fixture all take effect without
    touching the cache by hand. The unknown-key check runs again on the
    next read as well, so a typo introduced by an override is still reported.
    """
    self._config_cache = None
    self._unknown_keys_checked = False

Models¶

django_rls_tenants.tenants.models.RLSProtectedModel ¶

Bases: Model

Abstract base model for tenant-scoped models.

Provides:

A tenant ForeignKey added dynamically via the class_prepared signal (target read from RLS_TENANTS["TENANT_MODEL"]).
RLSManager as the default manager (with for_user()).
RLSConstraint in Meta.constraints (generates RLS policy).

Usage::

class Order(RLSProtectedModel):
    product = models.CharField(max_length=255)
    amount = models.DecimalField(...)

    class Meta(RLSProtectedModel.Meta):
        db_table = "order"

To customize the tenant FK (e.g., nullable for admin users), declare the field directly on your model -- the class_prepared handler will not add a duplicate::

class User(AbstractUser, RLSProtectedModel):
    tenant = models.ForeignKey(
        Tenant, on_delete=models.CASCADE,
        null=True, blank=True,
    )

Managers¶

django_rls_tenants.tenants.managers.TenantQuerySet ¶

TenantQuerySet(*args: Any, **kwargs: Any)

Bases: QuerySet

QuerySet that applies a for_user() user's RLS context when it runs SQL.

for_user() stores the user and sets nothing. Every method that executes SQL then runs its statements in a scope that sets the user's GUC variables and tenant state on the connection the SQL uses and puts the previous values back afterwards (see "Lazy QuerySet Evaluation" in the architecture guide). That covers iteration (_fetch_all()), count(), exists(), aggregate(), iterator(), aiterator(), update(), delete(), create(), bulk_create(), bulk_update(), update_or_create() and explain(); everything else (get(), first(), get_or_create(), contains(), in_bulk(), the a-prefixed async methods, ...) runs its SQL through one of those.

When auto-scope is active (via tenant_context() or middleware), select_related() automatically adds WHERE related.tenant_id = X for joined RLS-protected tables, enabling index usage on both sides of the join.

Source code in django_rls_tenants/tenants/managers.py

def __init__(self, *args: Any, **kwargs: Any) -> None:
    super().__init__(*args, **kwargs)
    self._rls_user = None

for_user ¶

for_user(as_user: TenantUser) -> TenantQuerySet

Scope this queryset to the given user's tenant.

For admin users: returns all rows (RLS admin bypass at eval time). For tenant users: returns rows matching the user's tenant.

The queryset remains lazy and chainable. The user's GUC variables are set when the queryset runs SQL, not when for_user() is called, and only for the duration of that SQL: every method that executes a query applies them, and the values found on the connection before are restored afterwards (see "Lazy QuerySet Evaluation" in the architecture guide).

Parameters:

Name	Type	Description	Default
`as_user`	`TenantUser`	User object satisfying the `TenantUser` protocol.	required

Source code in django_rls_tenants/tenants/managers.py

def for_user(self, as_user: TenantUser) -> TenantQuerySet:
    """Scope this queryset to the given user's tenant.

    For admin users: returns all rows (RLS admin bypass at eval time).
    For tenant users: returns rows matching the user's tenant.

    The queryset remains lazy and chainable. The user's GUC variables
    are set when the queryset runs SQL, not when ``for_user()`` is
    called, and only for the duration of that SQL: every method that
    executes a query applies them, and the values found on the
    connection before are restored afterwards (see "Lazy QuerySet
    Evaluation" in the architecture guide).

    Args:
        as_user: User object satisfying the ``TenantUser`` protocol.
    """
    qs = self._clone()
    qs._rls_user = as_user  # noqa: SLF001
    if not as_user.is_tenant_admin:
        # Defense-in-depth: Django-level filter provides isolation
        # even if GUC is misconfigured. RLS provides isolation even
        # if the Django filter is bypassed (e.g., raw SQL).
        conf = rls_tenants_config
        qs = qs.filter(**{f"{conf.TENANT_FK_FIELD}_id": as_user.rls_tenant_id})
    return qs

select_related ¶

select_related(*fields: Any) -> TenantQuerySet

Override to add tenant filters on joined RLS-protected tables.

When a tenant context is active (or for_user() was called), adds WHERE related.tenant_id = X for each explicitly named relation that targets an RLS-protected model. This enables PostgreSQL to use composite indexes on joined tables instead of relying solely on per-row RLS current_setting() evaluation.

Falls back to super().select_related() when no tenant scope is active. Called with no arguments (select-all mode) it adds no tenant filters; Django 6.1 deprecates that form, and the RemovedInDjango2028Warning names the caller's line, not this method.

select_related(None) clears the selection without adding tenant filters. select_related(False) is treated the same way as a convenience kept for existing callers; Django itself has never accepted False.

Source code in django_rls_tenants/tenants/managers.py

def select_related(self, *fields: Any) -> TenantQuerySet:
    """Override to add tenant filters on joined RLS-protected tables.

    When a tenant context is active (or ``for_user()`` was called),
    adds ``WHERE related.tenant_id = X`` for each explicitly named
    relation that targets an RLS-protected model. This enables
    PostgreSQL to use composite indexes on joined tables instead of
    relying solely on per-row RLS ``current_setting()`` evaluation.

    Falls back to ``super().select_related()`` when no tenant scope
    is active. Called with no arguments (select-all mode) it adds no
    tenant filters; Django 6.1 deprecates that form, and the
    ``RemovedInDjango2028Warning`` names the caller's line, not this
    method.

    ``select_related(None)`` clears the selection without adding
    tenant filters. ``select_related(False)`` is treated the same way
    as a convenience kept for existing callers; Django itself has
    never accepted ``False``.
    """
    # Django never accepted False (it reaches add_select_related() and
    # fails with 'bool' object has no attribute 'split'); normalize it
    # to None, which clears the selection.
    if fields in ((False,), (None,)):
        return super().select_related(None)
    if not fields:
        if not _SELECT_RELATED_NO_ARGS_DEPRECATED:
            return super().select_related()
        # These guards (and _chain() below) mirror QuerySet.select_related();
        # all three are internal Django APIs that django-stubs does not declare.
        self._not_support_combined_queries("select_related")  # type: ignore[attr-defined]
        if self._fields is not None:  # type: ignore[attr-defined]
            raise TypeError("Cannot call select_related() after .values() or .values_list()")
        # Django 6.1 deprecates the no-argument form through
        # warn_about_external_use(), which blames the first frame outside
        # django.db.models -- this override. Emit the same warning
        # ourselves with this package skipped, so the user's call site is
        # named, and set the flag the way QuerySet.select_related() does,
        # so Django does not warn a second time. Django-internal callers
        # (the admin changelist) get no warning from us; Django deprecates
        # list_select_related=True on its own.
        # RemovedInDjango2028Warning: when Django removes the no-argument
        # form, raise TypeError here too.
        warn_about_external_use(
            "Calling select_related() with no arguments is deprecated. "
            "Specify the fields to fetch instead.",
            category=RemovedInDjango2028Warning,
            skip_name_prefixes=("django.db.models", "django_rls_tenants."),
        )
        everything: TenantQuerySet = self._chain()  # type: ignore[attr-defined]
        everything.query.select_related = True
        return everything
    tenant_id = self._get_active_tenant_id()
    qs: TenantQuerySet = super().select_related(*fields)
    if tenant_id is None:
        return qs
    conf = rls_tenants_config
    fk_field_id = f"{conf.TENANT_FK_FIELD}_id"
    for field_path in fields:
        related_model, nullable = _resolve_related_path(self.model, field_path)
        if related_model is not None and _is_rls_protected(related_model):
            tenant_filter = Q(**{f"{field_path}__{fk_field_id}": tenant_id})
            # Preserve LEFT OUTER JOIN semantics when *any* hop of the
            # path is nullable: include rows whose chain ends in NULL.
            # Without this, .filter() forces an INNER JOIN which
            # silently drops those rows. The check must cover every
            # segment, not just the first or the last one -- a nullable
            # intermediate FK breaks the chain just as well (#79).
            if nullable:
                tenant_filter = tenant_filter | Q(**{f"{field_path}__isnull": True})
            qs = qs.filter(tenant_filter)
    return qs

count ¶

count() -> int

Run count() in the for_user() scope; guarded by strict mode.

A queryset with cached results answers from the cache, so it runs no SQL and needs neither the scope nor the strict mode check.

Source code in django_rls_tenants/tenants/managers.py

def count(self) -> int:
    """Run ``count()`` in the ``for_user()`` scope; guarded by strict mode.

    A queryset with cached results answers from the cache, so it runs
    no SQL and needs neither the scope nor the strict mode check.
    """
    if self._result_cache is not None:
        return super().count()
    self._check_strict_mode()
    with self._rls_scope():
        return super().count()

exists ¶

exists() -> bool

Run exists() in the for_user() scope; guarded by strict mode.

A queryset with cached results answers from the cache, so it runs no SQL and needs neither the scope nor the strict mode check.

Source code in django_rls_tenants/tenants/managers.py

def exists(self) -> bool:
    """Run ``exists()`` in the ``for_user()`` scope; guarded by strict mode.

    A queryset with cached results answers from the cache, so it runs
    no SQL and needs neither the scope nor the strict mode check.
    """
    if self._result_cache is not None:
        return super().exists()
    self._check_strict_mode()
    with self._rls_scope():
        return super().exists()

aggregate ¶

aggregate(*args: Any, **kwargs: Any) -> dict[str, Any]

Run aggregate() in the for_user() scope; guarded by strict mode.

Source code in django_rls_tenants/tenants/managers.py

def aggregate(self, *args: Any, **kwargs: Any) -> dict[str, Any]:
    """Run ``aggregate()`` in the ``for_user()`` scope; guarded by strict mode."""
    self._check_strict_mode()
    with self._rls_scope():
        return super().aggregate(*args, **kwargs)

update ¶

update(**kwargs: Any) -> int

Run update() in the for_user() scope; guarded by strict mode.

Source code in django_rls_tenants/tenants/managers.py

def update(self, **kwargs: Any) -> int:
    """Run ``update()`` in the ``for_user()`` scope; guarded by strict mode."""
    self._check_strict_mode()
    with self._rls_scope(for_write=True):
        return super().update(**kwargs)

delete ¶

delete() -> tuple[int, dict[str, int]]

Run delete() in the for_user() scope; guarded by strict mode.

One scope covers the whole deletion: collecting the rows, the cascades and SET_NULL updates on related models, the delete signals and the DELETE statements.

Source code in django_rls_tenants/tenants/managers.py

def delete(self) -> tuple[int, dict[str, int]]:
    """Run ``delete()`` in the ``for_user()`` scope; guarded by strict mode.

    One scope covers the whole deletion: collecting the rows, the
    cascades and ``SET_NULL`` updates on related models, the delete
    signals and the ``DELETE`` statements.
    """
    self._check_strict_mode()
    with self._rls_scope(for_write=True, own_transaction=True):
        return super().delete()

iterator ¶

iterator(chunk_size: int | None = None) -> Iterator[Any]

Run iterator() in the for_user() scope; guarded by strict mode.

Each chunk of chunk_size rows (Django's default of 2000 when None) is fetched, together with its prefetch_related() lookups, inside its own scope. The scope is not held while the caller's loop body runs, so the user's context neither leaks into that code nor stays on the connection when the loop stops early.

Source code in django_rls_tenants/tenants/managers.py

def iterator(self, chunk_size: int | None = None) -> Iterator[Any]:
    """Run ``iterator()`` in the ``for_user()`` scope; guarded by strict mode.

    Each chunk of ``chunk_size`` rows (Django's default of 2000 when
    ``None``) is fetched, together with its ``prefetch_related()``
    lookups, inside its own scope. The scope is not held while the
    caller's loop body runs, so the user's context neither leaks into
    that code nor stays on the connection when the loop stops early.
    """
    self._check_strict_mode()
    if self._rls_user is None:
        return super().iterator(chunk_size=chunk_size)
    # Resolved once: the cursor and every chunk's scope use the same alias.
    using = self.db
    with self._pinned_db(using):
        rows = super().iterator(chunk_size=chunk_size)
    return self._scoped_chunks(rows, chunk_size or _ITERATOR_CHUNK_SIZE, using)

aiterator `async` ¶

aiterator(
    chunk_size: int = _ITERATOR_CHUNK_SIZE,
) -> AsyncIterator[Any]

Run aiterator() in the for_user() scope; guarded by strict mode.

Django's aiterator() does not go through iterator(). With a for_user() user, each chunk is pulled from :meth:iterator in one thread-sensitive sync_to_async hop, so every scope is entered and left inside a hop, on the thread that owns the connection.

Source code in django_rls_tenants/tenants/managers.py

async def aiterator(self, chunk_size: int = _ITERATOR_CHUNK_SIZE) -> AsyncIterator[Any]:
    """Run ``aiterator()`` in the ``for_user()`` scope; guarded by strict mode.

    Django's ``aiterator()`` does not go through ``iterator()``. With a
    ``for_user()`` user, each chunk is pulled from :meth:`iterator` in
    one thread-sensitive ``sync_to_async`` hop, so every scope is
    entered and left inside a hop, on the thread that owns the
    connection.
    """
    if self._rls_user is None:
        self._check_strict_mode()
        async for item in super().aiterator(chunk_size=chunk_size):
            yield item
        return
    if chunk_size <= 0:  # QuerySet.aiterator()'s check and message
        msg = "Chunk size must be strictly positive."
        raise ValueError(msg)
    # iterator() also runs the strict mode check, which for_user() satisfies.
    rows = self.iterator(chunk_size=chunk_size)
    while True:
        chunk = await sync_to_async(_next_chunk, thread_sensitive=True)(rows, chunk_size)
        for item in chunk:
            yield item
        if len(chunk) < chunk_size:
            return

create ¶

create(**kwargs: Any) -> Any

Run create() in the for_user() scope.

The INSERT must pass the RLS policy's WITH CHECK clause, which reads the GUCs. Unlike bulk_create(), which has been guarded since strict mode was added, create() is not: without a context the database rejects the row loudly instead of returning a wrong result, and guarding it now would break code that creates rows outside a context on a superuser connection.

Source code in django_rls_tenants/tenants/managers.py

def create(self, **kwargs: Any) -> Any:
    """Run ``create()`` in the ``for_user()`` scope.

    The ``INSERT`` must pass the RLS policy's ``WITH CHECK`` clause,
    which reads the GUCs. Unlike ``bulk_create()``, which has been
    guarded since strict mode was added, ``create()`` is not: without a
    context the database rejects the row loudly instead of returning a
    wrong result, and guarding it now would break code that creates
    rows outside a context on a superuser connection.
    """
    with self._rls_scope(for_write=True):
        return super().create(**kwargs)

bulk_create ¶

bulk_create(
    objs: Any,
    batch_size: int | None = None,
    ignore_conflicts: bool = False,
    update_conflicts: bool = False,
    update_fields: Any = None,
    unique_fields: Any = None,
) -> list[Any]

Run bulk_create() in the for_user() scope; guarded by strict mode.

Source code in django_rls_tenants/tenants/managers.py

def bulk_create(
    self,
    objs: Any,
    batch_size: int | None = None,
    ignore_conflicts: bool = False,
    update_conflicts: bool = False,
    update_fields: Any = None,
    unique_fields: Any = None,
) -> list[Any]:
    """Run ``bulk_create()`` in the ``for_user()`` scope; guarded by strict mode."""
    self._check_strict_mode()
    with self._rls_scope(for_write=True, own_transaction=True):
        return super().bulk_create(
            objs,
            batch_size=batch_size,
            ignore_conflicts=ignore_conflicts,
            update_conflicts=update_conflicts,
            update_fields=update_fields,
            unique_fields=unique_fields,
        )

bulk_update ¶

bulk_update(
    objs: Any, fields: Any, batch_size: int | None = None
) -> int

Run bulk_update() in one for_user() scope; guarded by strict mode.

Source code in django_rls_tenants/tenants/managers.py

def bulk_update(
    self,
    objs: Any,
    fields: Any,
    batch_size: int | None = None,
) -> int:
    """Run ``bulk_update()`` in one ``for_user()`` scope; guarded by strict mode."""
    self._check_strict_mode()
    with self._rls_scope(for_write=True, own_transaction=True):
        return super().bulk_update(objs, fields, batch_size=batch_size)

update_or_create ¶

update_or_create(
    defaults: Mapping[str, Any] | None = None,
    create_defaults: Mapping[str, Any] | None = None,
    **kwargs: Any,
) -> tuple[Any, bool]

Run update_or_create() in the for_user() scope.

Updating an existing row calls Model.save(), which writes through the model's base manager rather than this queryset, so one scope spans the whole call. The lookup runs through _fetch_all() and is guarded by strict mode there.

Source code in django_rls_tenants/tenants/managers.py

def update_or_create(
    self,
    defaults: Mapping[str, Any] | None = None,
    create_defaults: Mapping[str, Any] | None = None,
    **kwargs: Any,
) -> tuple[Any, bool]:
    """Run ``update_or_create()`` in the ``for_user()`` scope.

    Updating an existing row calls ``Model.save()``, which writes
    through the model's base manager rather than this queryset, so one
    scope spans the whole call. The lookup runs through ``_fetch_all()``
    and is guarded by strict mode there.
    """
    with self._rls_scope(for_write=True, own_transaction=True):
        return super().update_or_create(
            defaults=defaults,
            create_defaults=create_defaults,
            **kwargs,
        )

explain ¶

explain(
    *, format: str | None = None, **options: Any
) -> str

Run explain() in the for_user() scope.

With analyze=True the query runs, and the plan then reports the user's rows. Not guarded by strict mode: it is a diagnostic.

Source code in django_rls_tenants/tenants/managers.py

def explain(
    self,
    *,
    format: str | None = None,  # noqa: A002  -- QuerySet.explain()'s parameter name
    **options: Any,
) -> str:
    """Run ``explain()`` in the ``for_user()`` scope.

    With ``analyze=True`` the query runs, and the plan then reports the
    user's rows. Not guarded by strict mode: it is a diagnostic.
    """
    with self._rls_scope():
        return super().explain(format=format, **options)

get ¶

get(*args: Any, **kwargs: Any) -> Any

Guard get() with strict mode check.

Source code in django_rls_tenants/tenants/managers.py

def get(self, *args: Any, **kwargs: Any) -> Any:
    """Guard ``get()`` with strict mode check."""
    self._check_strict_mode()
    return super().get(*args, **kwargs)

first ¶

first() -> Any

Guard first() with strict mode check, unless the results are cached.

On a cached, ordered queryset Django answers from the cache. Any other case queries a clone, which _fetch_all() checks.

Source code in django_rls_tenants/tenants/managers.py

def first(self) -> Any:
    """Guard ``first()`` with strict mode check, unless the results are cached.

    On a cached, ordered queryset Django answers from the cache. Any
    other case queries a clone, which ``_fetch_all()`` checks.
    """
    if self._result_cache is None:
        self._check_strict_mode()
    return super().first()

last ¶

last() -> Any

Guard last() with strict mode check.

It always queries: Django reverses the ordering on a fresh clone.

Source code in django_rls_tenants/tenants/managers.py

def last(self) -> Any:
    """Guard ``last()`` with strict mode check.

    It always queries: Django reverses the ordering on a fresh clone.
    """
    self._check_strict_mode()
    return super().last()

django_rls_tenants.tenants.managers.RLSManager ¶

Bases: Manager

Manager for RLS-protected models.

Provides for_user() for scoped queries and prepare_tenant_in_model_data() for resolving tenant FKs.

get_queryset ¶

get_queryset() -> TenantQuerySet

Return a TenantQuerySet instance, auto-scoped if a tenant context is active.

When tenant_context(), admin_context(), or RLSTenantMiddleware has set a current tenant ID, the queryset is automatically filtered by WHERE tenant_id = X. This enables PostgreSQL to use composite indexes instead of relying solely on RLS current_setting() calls.

Source code in django_rls_tenants/tenants/managers.py

def get_queryset(self) -> TenantQuerySet:
    """Return a ``TenantQuerySet`` instance, auto-scoped if a tenant context is active.

    When ``tenant_context()``, ``admin_context()``, or ``RLSTenantMiddleware``
    has set a current tenant ID, the queryset is automatically filtered by
    ``WHERE tenant_id = X``. This enables PostgreSQL to use composite indexes
    instead of relying solely on RLS ``current_setting()`` calls.
    """
    # hints too, as BaseManager.get_queryset() passes them: a router reads
    # them (db_manager(hints=...), the hint Django adds for related objects).
    qs = TenantQuerySet(self.model, using=self._db, hints=self._hints)
    tenant_id = get_current_tenant_id()
    if tenant_id is not None:
        conf = rls_tenants_config
        qs = qs.filter(**{f"{conf.TENANT_FK_FIELD}_id": tenant_id})
    return qs

for_user ¶

for_user(as_user: TenantUser) -> TenantQuerySet

Return a queryset scoped to the given user's tenant.

Source code in django_rls_tenants/tenants/managers.py

def for_user(self, as_user: TenantUser) -> TenantQuerySet:
    """Return a queryset scoped to the given user's tenant."""
    return self.get_queryset().for_user(as_user=as_user)

prepare_tenant_in_model_data ¶

prepare_tenant_in_model_data(
    model_data: dict[str, Any], as_user: TenantUser
) -> None

Resolve a raw tenant ID for model creation.

If model_data contains a raw tenant ID (int/str) under the configured FK field name, sets the FK column directly ({field}_id) to avoid a SELECT query. Allows passing tenant=42 in creation data without N+1 overhead.

Parameters:

Name	Type	Description	Default
`model_data`	`dict[str, Any]`	Dict of field names to values.	required
`as_user`	`TenantUser`	User for context (unused here but part of API).	required

Source code in django_rls_tenants/tenants/managers.py

def prepare_tenant_in_model_data(
    self,
    model_data: dict[str, Any],
    # Unused but part of the public API contract.
    as_user: TenantUser,  # noqa: ARG002
) -> None:
    """Resolve a raw tenant ID for model creation.

    If ``model_data`` contains a raw tenant ID (int/str) under
    the configured FK field name, sets the FK column directly
    (``{field}_id``) to avoid a ``SELECT`` query. Allows passing
    ``tenant=42`` in creation data without N+1 overhead.

    Args:
        model_data: Dict of field names to values.
        as_user: User for context (unused here but part of API).
    """
    # Lazy import avoids a circular import.
    from django.apps import apps  # noqa: PLC0415

    conf = rls_tenants_config
    field_name = conf.TENANT_FK_FIELD
    tenant_model = apps.get_model(conf.TENANT_MODEL)

    tenant = model_data.get(field_name)
    if tenant is not None and not isinstance(tenant, tenant_model):
        # Set the FK column directly to avoid a model fetch query.
        # For bulk creates this eliminates N identical SELECTs.
        fk_column = f"{field_name}_id"
        model_data[fk_column] = tenant
        del model_data[field_name]

Context Managers¶

django_rls_tenants.tenants.context.tenant_context ¶

tenant_context(
    tenant_id: int | str, *, using: str = "default"
) -> Iterator[None]

Set RLS context to a specific tenant. Supports nesting.

The previous GUC values are saved on entry and restored on exit in both GUC modes. With USE_LOCAL_SET=True the restore is issued with SET LOCAL so an enclosing block keeps its database scope for the rest of the transaction (see :func:django_rls_tenants.rls.context.restore_gucs for the behaviour when the block exits with an aborted transaction). Outside transaction.atomic() that mode writes nothing on entry or exit: the GUC affinity guard sets the context's values transaction-locally for each statement, giving a statement run in autocommit a transaction of its own (see "GUC affinity guard" in the architecture guide).

Parameters:

Name	Type	Description	Default
`tenant_id`	`int \| str`	The tenant PK to scope queries to.	required
`using`	`str`	Database alias the GUCs are written to on entry and restored on exit. Default: `"default"`. Every alias in `RLS_TENANTS["DATABASES"]` follows the context too (the GUC affinity guard applies it before each statement there).	`'default'`

Raises:

Type	Description
`NoTenantContextError`	If `tenant_id` is `None`.
`RuntimeError`	If `USE_LOCAL_SET=True`, `using` is not in `RLS_TENANTS["DATABASES"]` and its connection is not inside `transaction.atomic()`.

Source code in django_rls_tenants/tenants/context.py

@contextmanager
def tenant_context(
    tenant_id: int | str,
    *,
    using: str = "default",
) -> Iterator[None]:
    """Set RLS context to a specific tenant. Supports nesting.

    The previous GUC values are saved on entry and restored on exit in both
    GUC modes. With ``USE_LOCAL_SET=True`` the restore is issued with
    ``SET LOCAL`` so an enclosing block keeps its database scope for the
    rest of the transaction (see :func:`django_rls_tenants.rls.context.restore_gucs`
    for the behaviour when the block exits with an aborted transaction).
    Outside ``transaction.atomic()`` that mode writes nothing on entry or
    exit: the GUC affinity guard sets the context's values
    transaction-locally for each statement, giving a statement run in
    autocommit a transaction of its own (see "GUC affinity guard" in the
    architecture guide).

    Args:
        tenant_id: The tenant PK to scope queries to.
        using: Database alias the GUCs are written to on entry and restored on
            exit. Default: ``"default"``. Every alias in
            ``RLS_TENANTS["DATABASES"]`` follows the context too (the GUC
            affinity guard applies it before each statement there).

    Raises:
        NoTenantContextError: If ``tenant_id`` is ``None``.
        RuntimeError: If ``USE_LOCAL_SET=True``, ``using`` is not in
            ``RLS_TENANTS["DATABASES"]`` and its connection is not inside
            ``transaction.atomic()``.
    """
    if tenant_id is None:
        msg = "tenant_id cannot be None."
        raise NoTenantContextError(msg, hint=HINT_TENANT_ID_NONE)

    conf = rls_tenants_config
    is_local = conf.USE_LOCAL_SET
    writes = _writes_gucs(using, is_local=is_local)

    # Save previous state for nesting support (both modes: a SET LOCAL made
    # by an enclosing block is visible here and must be re-issued on exit).
    previous = _snapshot_gucs(conf, using=using) if writes else None
    session = _session_of(using)
    enclosing = _current_live()

    gucs = ContextGucs.for_tenant(tenant_id)
    exc_pending = False
    try:
        with _rls_state(tenant_id, gucs):
            if writes:
                with writing(using, gucs, local=is_local):
                    set_guc(conf.GUC_IS_ADMIN, gucs.is_admin, is_local=is_local, using=using)
                    set_guc(
                        conf.GUC_CURRENT_TENANT,
                        gucs.current_tenant,
                        is_local=is_local,
                        using=using,
                    )
            logger.debug("tenant_context entered: tenant_id=%s, using=%s", tenant_id, using)
            yield
    except BaseException:
        exc_pending = True
        raise
    finally:
        _restore_tenant_gucs(
            previous,
            is_local=is_local,
            using=using,
            exc_pending=exc_pending,
            enclosing=enclosing,
            session=session,
        )
        logger.debug("tenant_context exited: tenant_id=%s, using=%s", tenant_id, using)

django_rls_tenants.tenants.context.admin_context ¶

admin_context(*, using: str = 'default') -> Iterator[None]

Set RLS context to admin mode. Supports nesting.

The previous GUC values are saved on entry and restored on exit in both GUC modes, so an inner admin_context() never leaks the admin bypass into the enclosing block (see :func:tenant_context, also for USE_LOCAL_SET=True outside transaction.atomic()).

Parameters:

Name	Type	Description	Default
`using`	`str`	Database alias the GUCs are written to on entry and restored on exit. Default: `"default"`. Every alias in `RLS_TENANTS["DATABASES"]` gets the admin bypass too, even when `using` names an alias outside that list.	`'default'`

Raises:

Type	Description
`RuntimeError`	If `USE_LOCAL_SET=True`, `using` is not in `RLS_TENANTS["DATABASES"]` and its connection is not inside `transaction.atomic()`.

Source code in django_rls_tenants/tenants/context.py

@contextmanager
def admin_context(
    *,
    using: str = "default",
) -> Iterator[None]:
    """Set RLS context to admin mode. Supports nesting.

    The previous GUC values are saved on entry and restored on exit in both
    GUC modes, so an inner ``admin_context()`` never leaks the admin bypass
    into the enclosing block (see :func:`tenant_context`, also for
    ``USE_LOCAL_SET=True`` outside ``transaction.atomic()``).

    Args:
        using: Database alias the GUCs are written to on entry and restored on
            exit. Default: ``"default"``. Every alias in
            ``RLS_TENANTS["DATABASES"]`` gets the admin bypass too, even when
            ``using`` names an alias outside that list.

    Raises:
        RuntimeError: If ``USE_LOCAL_SET=True``, ``using`` is not in
            ``RLS_TENANTS["DATABASES"]`` and its connection is not inside
            ``transaction.atomic()``.
    """
    conf = rls_tenants_config
    is_local = conf.USE_LOCAL_SET
    writes = _writes_gucs(using, is_local=is_local)

    # Save previous state for nesting support (both modes).
    previous = _snapshot_gucs(conf, using=using) if writes else None
    session = _session_of(using)
    enclosing = _current_live()

    exc_pending = False
    try:
        with _rls_state(None, ADMIN_GUCS):
            if writes:
                with writing(using, ADMIN_GUCS, local=is_local):
                    set_guc(
                        conf.GUC_IS_ADMIN,
                        ADMIN_GUCS.is_admin,
                        is_local=is_local,
                        using=using,
                    )
                    # Clear tenant GUC for admin mode; the admin_bypass clause in the
                    # RLS policy handles access independently. Avoids the old "-1"
                    # sentinel which could collide with integer PKs or fail UUID casts.
                    clear_guc(conf.GUC_CURRENT_TENANT, is_local=is_local, using=using)
            logger.debug("admin_context entered: using=%s", using)
            yield
    except BaseException:
        exc_pending = True
        raise
    finally:
        _restore_tenant_gucs(
            previous,
            is_local=is_local,
            using=using,
            exc_pending=exc_pending,
            enclosing=enclosing,
            session=session,
        )
        logger.debug("admin_context exited: using=%s", using)

django_rls_tenants.tenants.context.async_tenant_context `async` ¶

async_tenant_context(
    tenant_id: int | str, *, using: str = "default"
) -> AsyncIterator[None]

Async counterpart of :func:tenant_context, for async with. Supports nesting.

Usage::

async with async_tenant_context(tenant.pk):
    count = await Order.objects.acount()

The GUCs are snapshotted and set in one thread_sensitive sync_to_async hop on entry and restored in one hop on exit. Django has no asynchronous database cursor, so this is not a non-blocking call; the hop runs on the thread that owns the task's connection (under ASGI, the request's thread, shared with the async ORM). The tenant state ContextVar values are set and reset in the coroutine itself, so the ORM auto-scope filter and STRICT_MODE see them in async code. Nesting (with sync or async blocks) and the restore in both GUC modes behave as in :func:tenant_context.

USE_LOCAL_SET. A coroutine cannot open transaction.atomic(), so with USE_LOCAL_SET=True the hop thread is usually in autocommit: the block then writes nothing, and the GUC affinity guard sets its values transaction-locally for each statement, in a transaction of its own (see :func:tenant_context). When the coroutine is driven by async_to_sync from synchronous code inside atomic(), the block writes SET LOCAL values into that transaction, as the sync manager does. On an alias outside RLS_TENANTS["DATABASES"], which the guard leaves alone, it needs that transaction and raises :class:RLSConfigurationError without one.

Concurrent tasks. Sibling tasks of one request (for example under asyncio.gather) share the request's database connection, but each has its own ContextVar state. The GUC affinity guard applies the context of the task that runs a statement before that statement, so blocks for different tenants can run concurrently, raw SQL, safe_tenant_sql() and M2M through tables included. It does not cover bypass flags and other GUCs set with async_bypass_flag() / async_rls_context() (sibling tasks see them), or SQL that bypasses Django's cursor. Work that outlives the block (a task it started without awaiting it) runs without its context.

The restore runs even when the task is cancelled while entering or inside the block; only a second cancellation delivered during the restore itself can skip it (for requests RLSTenantMiddleware set a context for, its safety net clears the GUCs at the end of the request).

Parameters:

Name	Type	Description	Default
`tenant_id`	`int \| str`	The tenant PK to scope queries to.	required
`using`	`str`	Database alias the GUCs are written to first, as in :func:`tenant_context`. Default: `"default"`.	`'default'`

Raises:

Type	Description
`NoTenantContextError`	If `tenant_id` is `None`.
`RLSConfigurationError`	If `USE_LOCAL_SET=True`, `using` is not in `RLS_TENANTS["DATABASES"]` and no transaction is open on the hop thread.

Source code in django_rls_tenants/tenants/context.py

@asynccontextmanager
async def async_tenant_context(
    tenant_id: int | str,
    *,
    using: str = "default",
) -> AsyncIterator[None]:
    """Async counterpart of :func:`tenant_context`, for ``async with``. Supports nesting.

    Usage::

        async with async_tenant_context(tenant.pk):
            count = await Order.objects.acount()

    The GUCs are snapshotted and set in one ``thread_sensitive``
    ``sync_to_async`` hop on entry and restored in one hop on exit. Django
    has no asynchronous database cursor, so this is not a non-blocking call;
    the hop runs on the thread that owns the task's connection (under ASGI,
    the request's thread, shared with the async ORM). The tenant state
    ``ContextVar`` values are set and reset in the coroutine itself, so the
    ORM auto-scope filter and ``STRICT_MODE`` see them in async code. Nesting
    (with sync or async blocks) and the restore in both GUC modes behave as
    in :func:`tenant_context`.

    **USE_LOCAL_SET.** A coroutine cannot open ``transaction.atomic()``, so
    with ``USE_LOCAL_SET=True`` the hop thread is usually in autocommit: the
    block then writes nothing, and the GUC affinity guard sets its values
    transaction-locally for each statement, in a transaction of its own (see
    :func:`tenant_context`). When the coroutine is driven by
    ``async_to_sync`` from synchronous code inside ``atomic()``, the block
    writes ``SET LOCAL`` values into that transaction, as the sync manager
    does. On an alias outside ``RLS_TENANTS["DATABASES"]``, which the guard
    leaves alone, it needs that transaction and raises
    :class:`RLSConfigurationError` without one.

    **Concurrent tasks.** Sibling tasks of one request (for example under
    ``asyncio.gather``) share the request's database connection, but each
    has its own ``ContextVar`` state. The GUC affinity guard applies the
    context of the task that runs a statement before that statement, so
    blocks for different tenants can run concurrently, raw SQL,
    ``safe_tenant_sql()`` and M2M through tables included. It does not cover
    bypass flags and other GUCs set with ``async_bypass_flag()`` /
    ``async_rls_context()`` (sibling tasks see them), or SQL that bypasses
    Django's cursor. Work that outlives the block (a task it started without
    awaiting it) runs without its context.

    The restore runs even when the task is cancelled while entering or
    inside the block; only a second cancellation delivered during the
    restore itself can skip it (for requests ``RLSTenantMiddleware`` set a
    context for, its safety net clears the GUCs at the end of the request).

    Args:
        tenant_id: The tenant PK to scope queries to.
        using: Database alias the GUCs are written to first, as in
            :func:`tenant_context`. Default: ``"default"``.

    Raises:
        NoTenantContextError: If ``tenant_id`` is ``None``.
        RLSConfigurationError: If ``USE_LOCAL_SET=True``, ``using`` is not in
            ``RLS_TENANTS["DATABASES"]`` and no transaction is open on the hop
            thread.
    """
    if tenant_id is None:
        msg = "tenant_id cannot be None."
        raise NoTenantContextError(msg, hint=HINT_TENANT_ID_NONE)

    conf = rls_tenants_config
    # Read once so entry and exit agree even if settings change mid-block.
    is_local = conf.USE_LOCAL_SET

    # GUCs first, ContextVars second (the sync manager does the reverse):
    # a failed hop then leaves no ContextVar to unwind. The ContextVars are
    # set here in the coroutine, never inside the hop, because a token
    # created in the hop's context copy cannot be reset from here (#58).
    gucs = ContextGucs.for_tenant(tenant_id)
    enclosing = _current_live()
    entered_on: list[object] = []
    previous = await _run_enter_hop(
        functools.partial(_enter_tenant_gucs, entered_on=entered_on),
        {conf.GUC_IS_ADMIN: gucs.is_admin, conf.GUC_CURRENT_TENANT: gucs.current_tenant},
        is_local=is_local,
        using=using,
        # Runs in the Context the block enters in, so the session is not needed.
        undo=functools.partial(_restore_tenant_gucs, enclosing=enclosing, session=None),
    )
    exc_pending = False
    try:
        with _rls_state(tenant_id, gucs):
            logger.debug(
                "async_tenant_context entered: tenant_id=%s, using=%s",
                tenant_id,
                using,
            )
            yield
    except BaseException:
        exc_pending = True
        raise
    finally:
        await sync_to_async(_restore_tenant_gucs, thread_sensitive=True)(
            previous,
            is_local=is_local,
            using=using,
            exc_pending=exc_pending,
            enclosing=enclosing,
            session=next(iter(entered_on), None),
        )
        logger.debug("async_tenant_context exited: tenant_id=%s, using=%s", tenant_id, using)

django_rls_tenants.tenants.context.async_admin_context `async` ¶

async_admin_context(
    *, using: str = "default"
) -> AsyncIterator[None]

Async counterpart of :func:admin_context, for async with. Supports nesting.

Same hop model, USE_LOCAL_SET behaviour and concurrent-task behaviour as :func:async_tenant_context. An inner async_admin_context() never leaks the admin bypass into the enclosing block.

Parameters:

Name	Type	Description	Default
`using`	`str`	Database alias the GUCs are written to first, as in :func:`admin_context`. Default: `"default"`.	`'default'`

Raises:

Type	Description
`RLSConfigurationError`	If `USE_LOCAL_SET=True`, `using` is not in `RLS_TENANTS["DATABASES"]` and no transaction is open on the hop thread.

Source code in django_rls_tenants/tenants/context.py

@asynccontextmanager
async def async_admin_context(
    *,
    using: str = "default",
) -> AsyncIterator[None]:
    """Async counterpart of :func:`admin_context`, for ``async with``. Supports nesting.

    Same hop model, ``USE_LOCAL_SET`` behaviour and concurrent-task
    behaviour as :func:`async_tenant_context`. An inner
    ``async_admin_context()`` never leaks the admin bypass into the
    enclosing block.

    Args:
        using: Database alias the GUCs are written to first, as in
            :func:`admin_context`. Default: ``"default"``.

    Raises:
        RLSConfigurationError: If ``USE_LOCAL_SET=True``, ``using`` is not in
            ``RLS_TENANTS["DATABASES"]`` and no transaction is open on the hop
            thread.
    """
    conf = rls_tenants_config
    is_local = conf.USE_LOCAL_SET

    # Same mapping as admin_context(): clear_guc() is set_guc(name, "").
    enclosing = _current_live()
    entered_on: list[object] = []
    previous = await _run_enter_hop(
        functools.partial(_enter_tenant_gucs, entered_on=entered_on),
        {conf.GUC_IS_ADMIN: ADMIN_GUCS.is_admin, conf.GUC_CURRENT_TENANT: ""},
        is_local=is_local,
        using=using,
        # Runs in the Context the block enters in, so the session is not needed.
        undo=functools.partial(_restore_tenant_gucs, enclosing=enclosing, session=None),
    )
    exc_pending = False
    try:
        with _rls_state(None, ADMIN_GUCS):
            logger.debug("async_admin_context entered: using=%s", using)
            yield
    except BaseException:
        exc_pending = True
        raise
    finally:
        await sync_to_async(_restore_tenant_gucs, thread_sensitive=True)(
            previous,
            is_local=is_local,
            using=using,
            exc_pending=exc_pending,
            enclosing=enclosing,
            session=next(iter(entered_on), None),
        )
        logger.debug("async_admin_context exited: using=%s", using)

django_rls_tenants.tenants.context.with_rls_context ¶

with_rls_context(
    func: Callable[P, R], *, user_param: str | None = None
) -> Callable[P, R]

with_rls_context(
    func: None = None, *, user_param: str | None = None
) -> Callable[[Callable[P, R]], Callable[P, R]]

with_rls_context(
    func: Callable[P, R] | None = None,
    *,
    user_param: str | None = None,
) -> (
    Callable[P, R]
    | Callable[[Callable[P, R]], Callable[P, R]]
)

Decorator that extracts a user argument and sets RLS context.

Can be used bare or with an explicit user_param::

@with_rls_context
def my_view(request, as_user): ...

@with_rls_context(user_param="current_user")
def my_view(request, current_user): ...

async def functions are supported: the decorator detects them with :func:inspect.iscoroutinefunction and returns an async def wrapper that enters :func:async_tenant_context / :func:async_admin_context around the awaited body. The wrapper is itself a coroutine function, so Django and other async-aware decorators treat it as async::

@with_rls_context
async def order_count(as_user):
    return await Order.objects.acount()

In the async wrapper the user argument is read (is_tenant_admin, rls_tenant_id) in the coroutine, when the call is awaited, so those properties must not query the database; pass a loaded user (in an async view, await request.auser(), not request.user).

Parameters:

Name	Type	Description	Default
`func`	`Callable[P, R] \| None`	The function to decorate (when used without parentheses).	`None`
`user_param`	`str \| None`	Override the parameter name to look for. Defaults to `RLS_TENANTS["USER_PARAM_NAME"]` (default: `"as_user"`).	`None`

Returns:

Type	Description
`Callable[P, R] \| Callable[[Callable[P, R]], Callable[P, R]]`	The decorated function with the same signature as the original,
`Callable[P, R] \| Callable[[Callable[P, R]], Callable[P, R]]`	or a decorator factory when called with keyword arguments.

Raises:

Type	Description
`TypeError`	At decoration time, if the function is a generator or an async generator function (its body runs while the caller iterates, after the context has been exited). At call time, if a function decorated as synchronous returns a coroutine.
`NoTenantContextError`	At call time (when awaited, for `async def`), if a non-admin user has `rls_tenant_id=None`.

When the user argument is None, logs a warning and proceeds without context (fail-closed: RLS blocks all access).

Source code in django_rls_tenants/tenants/context.py

def with_rls_context[**P, R](
    func: Callable[P, R] | None = None,
    *,
    user_param: str | None = None,
) -> Callable[P, R] | Callable[[Callable[P, R]], Callable[P, R]]:
    """Decorator that extracts a user argument and sets RLS context.

    Can be used bare or with an explicit ``user_param``::

        @with_rls_context
        def my_view(request, as_user): ...

        @with_rls_context(user_param="current_user")
        def my_view(request, current_user): ...

    ``async def`` functions are supported: the decorator detects them with
    :func:`inspect.iscoroutinefunction` and returns an ``async def`` wrapper
    that enters :func:`async_tenant_context` / :func:`async_admin_context`
    around the awaited body. The wrapper is itself a coroutine function, so
    Django and other async-aware decorators treat it as async::

        @with_rls_context
        async def order_count(as_user):
            return await Order.objects.acount()

    In the async wrapper the user argument is read (``is_tenant_admin``,
    ``rls_tenant_id``) in the coroutine, when the call is awaited, so those
    properties must not query the database; pass a loaded user (in an async
    view, ``await request.auser()``, not ``request.user``).

    Args:
        func: The function to decorate (when used without parentheses).
        user_param: Override the parameter name to look for. Defaults to
            ``RLS_TENANTS["USER_PARAM_NAME"]`` (default: ``"as_user"``).

    Returns:
        The decorated function with the same signature as the original,
        or a decorator factory when called with keyword arguments.

    Raises:
        TypeError: At decoration time, if the function is a generator or an
            async generator function (its body runs while the caller
            iterates, after the context has been exited). At call time, if a
            function decorated as synchronous returns a coroutine.
        NoTenantContextError: At call time (when awaited, for ``async def``),
            if a non-admin user has ``rls_tenant_id=None``.

    When the user argument is ``None``, logs a warning and proceeds
    without context (fail-closed: RLS blocks all access).
    """

    def decorator(fn: Callable[P, R]) -> Callable[P, R]:
        # Callable objects have no __qualname__ of their own.
        qualname: str = getattr(fn, "__qualname__", type(fn).__qualname__)

        if inspect.isgeneratorfunction(fn) or inspect.isasyncgenfunction(fn):
            msg = (
                f"@with_rls_context cannot decorate the generator function {qualname}: its "
                "body runs while the caller iterates, after the decorator's RLS context has "
                "been exited. Enter the context inside the generator instead (tenant_context() "
                "/ admin_context(), or async_tenant_context() / async_admin_context() in an "
                "async generator)."
            )
            raise TypeError(msg)

        # Cache signature at decoration time (not per-invocation).
        sig = inspect.signature(fn)
        param_name = user_param if user_param is not None else rls_tenants_config.USER_PARAM_NAME

        if param_name not in sig.parameters:
            logger.warning(
                "with_rls_context: parameter %r not found in signature of %s. "
                "RLS context will never be set (fail-closed). "
                "Use @with_rls_context(user_param='your_param') to specify explicitly.",
                param_name,
                qualname,
            )

        def user_of(*args: P.args, **kwargs: P.kwargs) -> TenantUser | None:
            """Read the user argument of one call; log a warning when it is ``None``."""
            as_user: TenantUser | None = _get_arg_from_signature(sig, param_name, *args, **kwargs)
            if as_user is None:
                logger.warning(
                    "with_rls_context: %s is None in call to %s, no RLS context set (fail-closed)",
                    param_name,
                    qualname,
                )
            return as_user

        if inspect.iscoroutinefunction(fn):

            @functools.wraps(fn)
            async def async_wrapper(*args: P.args, **kwargs: P.kwargs) -> Any:
                as_user = user_of(*args, **kwargs)
                if as_user is None:
                    return await fn(*args, **kwargs)
                tenant_id = _decorated_call_tenant_id(as_user, qualname)
                ctx = (
                    async_admin_context() if tenant_id is None else async_tenant_context(tenant_id)
                )
                async with ctx:
                    return await fn(*args, **kwargs)

            # mypy cannot express "R is a coroutine" for the async branch.
            return cast("Callable[P, R]", async_wrapper)

        @functools.wraps(fn)
        def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
            as_user = user_of(*args, **kwargs)
            if as_user is None:
                return _reject_coroutine(fn(*args, **kwargs), qualname)
            tenant_id = _decorated_call_tenant_id(as_user, qualname)
            ctx = admin_context() if tenant_id is None else tenant_context(tenant_id)
            with ctx:
                return _reject_coroutine(fn(*args, **kwargs), qualname)

        return wrapper

    # Support both @with_rls_context and @with_rls_context(user_param="x")
    if func is not None:
        return decorator(func)
    return decorator

Raw SQL¶

Helpers for scoping hand-written SQL to the current tenant. See the Raw SQL guide for usage and safety notes.

django_rls_tenants.tenants.sql.safe_tenant_sql ¶

safe_tenant_sql(
    column: str = "tenant_id",
    *,
    table: str | None = None,
    include_admin: bool = True,
    extra_bypass_flags: list[str] | None = None,
) -> str

Return a WHERE-clause fragment scoping rows to the current tenant.

The fragment compares column against the current-tenant GUC, using the exact expression the RLS policies use (via :mod:~django_rls_tenants.rls.policy_sql). When include_admin is true it also lets rows through while the admin-bypass GUC is set, mirroring admin_context(). Splice it straight into a raw query::

sql = f"SELECT * FROM orders WHERE {safe_tenant_sql()} AND amount > %s"
cursor.execute(sql, [100])

There are deliberately no bind parameters: the tenant id is read inside PostgreSQL from the session GUC, so the fragment contains no Python-side user input.

Parameters:

Name	Type	Description	Default
`column`	`str`	Tenant foreign-key column on the target table. Defaults to `"tenant_id"` (Django's column name for a `tenant` FK).	`'tenant_id'`
`table`	`str \| None`	Optional table name/alias to qualify the column with, e.g. `safe_tenant_sql(table="orders")` yields `"orders".tenant_id`. Use it when the query joins several tables and `column` would be ambiguous.	`None`
`include_admin`	`bool`	When `True` (default), the fragment also matches every row while the admin-bypass GUC is active, so a query run inside `admin_context()` sees all tenants -- matching the RLS policy. Set it to `False` to scope strictly to the current tenant regardless of admin (or any other bypass) state.	`True`
`extra_bypass_flags`	`list[str] \| None`	Additional boolean bypass GUCs that should also let rows through, matching the `extra_bypass_flags` you passed to the model's :class:`~django_rls_tenants.rls.constraints.RLSConstraint`. Pass the same list here so the fragment mirrors the live policy's `USING` clause; otherwise a session with one of those flags set passes the policy but is filtered out by this fragment. Only applied when `include_admin=True` (these flags extend the bypass set); ignored when `include_admin=False`.	`None`

Returns:

Type	Description
`str`	A SQL `WHERE`-clause fragment. With `include_admin=True` it is
`str`	wrapped in parentheses (`(<match> OR <admin> [OR <flag>...])`) so it
`str`	composes safely with surrounding `AND` clauses. With
`str`	`include_admin=False` it is the bare `<match>` predicate.

Warning

The parentheses only make the fragment safe to combine with AND. Appending OR defeats tenant isolation -- WHERE {safe_tenant_sql()} OR is_public returns rows from every tenant. Always restrict further with AND, never OR.

Raises:

Type	Description
`ValueError`	If `column` or `table` is not a valid SQL identifier, or if the configured GUC names, the tenant PK type, or any `extra_bypass_flags` entry is invalid.

Example

safe_tenant_sql("tenant_id", include_admin=False) "tenant_id = nullif((SELECT current_setting('rls.current_tenant', true)), '')::int"

The default (include_admin=True) wraps that predicate as (<predicate> OR (SELECT current_setting('rls.is_admin', true)) = 'true').

Source code in django_rls_tenants/tenants/sql.py

def safe_tenant_sql(
    column: str = "tenant_id",
    *,
    table: str | None = None,
    include_admin: bool = True,
    extra_bypass_flags: list[str] | None = None,
) -> str:
    """Return a ``WHERE``-clause fragment scoping rows to the current tenant.

    The fragment compares ``column`` against the current-tenant GUC, using the
    exact expression the RLS policies use (via
    :mod:`~django_rls_tenants.rls.policy_sql`).  When ``include_admin`` is true
    it also lets rows through while the admin-bypass GUC is set, mirroring
    ``admin_context()``.  Splice it straight into a raw query::

        sql = f"SELECT * FROM orders WHERE {safe_tenant_sql()} AND amount > %s"
        cursor.execute(sql, [100])

    There are deliberately no bind parameters: the tenant id is read inside
    PostgreSQL from the session GUC, so the fragment contains no Python-side
    user input.

    Args:
        column: Tenant foreign-key column on the target table. Defaults to
            ``"tenant_id"`` (Django's column name for a ``tenant`` FK).
        table: Optional table name/alias to qualify the column with, e.g.
            ``safe_tenant_sql(table="orders")`` yields ``"orders".tenant_id``.
            Use it when the query joins several tables and ``column`` would be
            ambiguous.
        include_admin: When ``True`` (default), the fragment also matches every
            row while the admin-bypass GUC is active, so a query run inside
            ``admin_context()`` sees all tenants -- matching the RLS policy. Set
            it to ``False`` to scope strictly to the current tenant regardless
            of admin (or any other bypass) state.
        extra_bypass_flags: Additional boolean bypass GUCs that should also let
            rows through, matching the ``extra_bypass_flags`` you passed to the
            model's :class:`~django_rls_tenants.rls.constraints.RLSConstraint`.
            Pass the *same* list here so the fragment mirrors the live policy's
            ``USING`` clause; otherwise a session with one of those flags set
            passes the policy but is filtered out by this fragment. Only applied
            when ``include_admin=True`` (these flags extend the bypass set);
            ignored when ``include_admin=False``.

    Returns:
        A SQL ``WHERE``-clause fragment. With ``include_admin=True`` it is
        wrapped in parentheses (``(<match> OR <admin> [OR <flag>...])``) so it
        composes safely with surrounding ``AND`` clauses. With
        ``include_admin=False`` it is the bare ``<match>`` predicate.

    Warning:
        The parentheses only make the fragment safe to combine with ``AND``.
        Appending ``OR`` defeats tenant isolation -- ``WHERE {safe_tenant_sql()}
        OR is_public`` returns rows from *every* tenant. Always restrict
        further with ``AND``, never ``OR``.

    Raises:
        ValueError: If ``column`` or ``table`` is not a valid SQL identifier, or
            if the configured GUC names, the tenant PK type, or any
            ``extra_bypass_flags`` entry is invalid.

    Example:
        >>> safe_tenant_sql("tenant_id", include_admin=False)
        "tenant_id = nullif((SELECT current_setting('rls.current_tenant', true)), '')::int"

        The default (``include_admin=True``) wraps that predicate as
        ``(<predicate> OR (SELECT current_setting('rls.is_admin', true)) = 'true')``.
    """
    guc_tenant, pk_type = _validated_tenant_guc_and_pk_type()

    _validate_field_name(column, "column")
    if table is not None:
        _validate_field_name(table, "table")
        col = f'"{table}".{column}'
    else:
        col = column

    predicate = tenant_match_sql(col, guc_tenant, pk_type)
    if not include_admin:
        return predicate

    # Bypass set mirrors RLSConstraint's USING clause: admin first, then any
    # extra flags, each read via the same InitPlan-wrapped boolean-flag helper.
    guc_admin = rls_tenants_config.GUC_IS_ADMIN
    _validate_guc_name(guc_admin)
    bypass_conditions = [bool_flag_sql(guc_admin)]
    for flag in extra_bypass_flags or []:
        _validate_guc_name(flag)
        bypass_conditions.append(bool_flag_sql(flag))

    bypass_clause = " OR ".join(bypass_conditions)
    return f"({predicate} OR {bypass_clause})"

django_rls_tenants.tenants.sql.current_tenant_value_sql ¶

current_tenant_value_sql() -> str

Return the current-tenant value expression for use in raw SQL.

Produces the same cast GUC read the RLS policies compare against -- e.g. nullif((SELECT current_setting('rls.current_tenant', true)), '')::int -- suitable for an INSERT value list or a SELECT projection::

cursor.execute(
    f"INSERT INTO orders (product, tenant_id) VALUES (%s, {current_tenant_value_sql()})",
    ["Widget"],
)

An unset GUC evaluates to NULL (via nullif(..., '')), so the cast never fails on an empty string.

Returns:

Type	Description
`str`	A SQL value expression yielding the current tenant id (or `NULL` when
`str`	no tenant context is active).

Raises:

Type	Description
`ValueError`	If the configured GUC name or tenant PK type is invalid.

Example

current_tenant_value_sql() "nullif((SELECT current_setting('rls.current_tenant', true)), '')::int"

Source code in django_rls_tenants/tenants/sql.py

def current_tenant_value_sql() -> str:
    """Return the current-tenant value expression for use in raw SQL.

    Produces the same cast GUC read the RLS policies compare against -- e.g.
    ``nullif((SELECT current_setting('rls.current_tenant', true)), '')::int`` --
    suitable for an ``INSERT`` value list or a ``SELECT`` projection::

        cursor.execute(
            f"INSERT INTO orders (product, tenant_id) VALUES (%s, {current_tenant_value_sql()})",
            ["Widget"],
        )

    An unset GUC evaluates to ``NULL`` (via ``nullif(..., '')``), so the cast
    never fails on an empty string.

    Returns:
        A SQL value expression yielding the current tenant id (or ``NULL`` when
        no tenant context is active).

    Raises:
        ValueError: If the configured GUC name or tenant PK type is invalid.

    Example:
        >>> current_tenant_value_sql()
        "nullif((SELECT current_setting('rls.current_tenant', true)), '')::int"
    """
    guc_tenant, pk_type = _validated_tenant_guc_and_pk_type()
    return tenant_value_sql(guc_tenant, pk_type)

State¶

Internal helpers

State functions are internal helpers for custom middleware and advanced use cases. Prefer tenant_context() and admin_context() for managing RLS state. Import state functions from django_rls_tenants.tenants.state, not the top-level package.

Tenant ID¶

django_rls_tenants.tenants.state.get_current_tenant_id ¶

get_current_tenant_id() -> int | str | None

Return the current tenant ID, or None if no tenant context is active.

Returns:

Type	Description
`int \| str \| None`	The tenant ID set by the innermost active `tenant_context()` or
`int \| str \| None`	middleware, or `None` if in admin mode or no context is active.

Source code in django_rls_tenants/tenants/state.py

def get_current_tenant_id() -> int | str | None:
    """Return the current tenant ID, or ``None`` if no tenant context is active.

    Returns:
        The tenant ID set by the innermost active ``tenant_context()`` or
        middleware, or ``None`` if in admin mode or no context is active.
    """
    return _current_tenant_id.get()

django_rls_tenants.tenants.state.set_current_tenant_id ¶

set_current_tenant_id(
    tenant_id: int | str | None,
) -> Token[int | str | None]

Set the current tenant ID for automatic query scoping.

Returns a token that can be passed to reset_current_tenant_id() to restore the previous value (for nesting support).

Parameters:

Name	Type	Description	Default
`tenant_id`	`int \| str \| None`	The tenant PK, or `None` to clear (admin/no context).	required

Returns:

Type	Description
`Token[int \| str \| None]`	A `Token` for restoring the previous value.

Source code in django_rls_tenants/tenants/state.py

def set_current_tenant_id(tenant_id: int | str | None) -> Token[int | str | None]:
    """Set the current tenant ID for automatic query scoping.

    Returns a token that can be passed to ``reset_current_tenant_id()``
    to restore the previous value (for nesting support).

    Args:
        tenant_id: The tenant PK, or ``None`` to clear (admin/no context).

    Returns:
        A ``Token`` for restoring the previous value.
    """
    return _current_tenant_id.set(tenant_id)

django_rls_tenants.tenants.state.reset_current_tenant_id ¶

reset_current_tenant_id(
    token: Token[int | str | None],
) -> None

Restore the previous tenant ID using a token from set_current_tenant_id().

Parameters:

Name	Type	Description	Default
`token`	`Token[int \| str \| None]`	The token returned by the corresponding `set_current_tenant_id()` call.	required

Source code in django_rls_tenants/tenants/state.py

def reset_current_tenant_id(token: Token[int | str | None]) -> None:
    """Restore the previous tenant ID using a token from ``set_current_tenant_id()``.

    Args:
        token: The token returned by the corresponding ``set_current_tenant_id()`` call.
    """
    _current_tenant_id.reset(token)

RLS Context Active (Strict Mode)¶

django_rls_tenants.tenants.state.get_rls_context_active ¶

get_rls_context_active() -> bool

Return whether an RLS context is currently active.

An RLS context is active when tenant_context(), admin_context(), or RLSTenantMiddleware has established a context for the current execution scope. Used by strict mode to distinguish "no context" from "admin context".

Returns:

Type	Description
`bool`	`True` if an RLS context is active.

Source code in django_rls_tenants/tenants/state.py

def get_rls_context_active() -> bool:
    """Return whether an RLS context is currently active.

    An RLS context is active when ``tenant_context()``,
    ``admin_context()``, or ``RLSTenantMiddleware`` has established
    a context for the current execution scope. Used by strict mode
    to distinguish "no context" from "admin context".

    Returns:
        ``True`` if an RLS context is active.
    """
    return _rls_context_active.get()

django_rls_tenants.tenants.state.set_rls_context_active ¶

set_rls_context_active(active: bool) -> Token[bool]

Set the RLS context active flag.

Parameters:

Name	Type	Description	Default
`active`	`bool`	Whether an RLS context is active.	required

Returns:

Type	Description
`Token[bool]`	A `Token` for restoring the previous value.

Source code in django_rls_tenants/tenants/state.py

def set_rls_context_active(active: bool) -> Token[bool]:
    """Set the RLS context active flag.

    Args:
        active: Whether an RLS context is active.

    Returns:
        A ``Token`` for restoring the previous value.
    """
    return _rls_context_active.set(active)

django_rls_tenants.tenants.state.reset_rls_context_active ¶

reset_rls_context_active(token: Token[bool]) -> None

Restore the previous RLS context active state.

Parameters:

Name	Type	Description	Default
`token`	`Token[bool]`	The token returned by `set_rls_context_active()`.	required

Source code in django_rls_tenants/tenants/state.py

def reset_rls_context_active(token: Token[bool]) -> None:
    """Restore the previous RLS context active state.

    Args:
        token: The token returned by ``set_rls_context_active()``.
    """
    _rls_context_active.reset(token)

Middleware¶

django_rls_tenants.tenants.middleware.RLSTenantMiddleware ¶

RLSTenantMiddleware(get_response: Callable[..., Any])

Bases: MiddlewareMixin

Set RLS context for each authenticated request.

For authenticated users: sets tenant_context or admin_context based on the user's TenantUser protocol implementation.

For unauthenticated requests: no context is set. RLS policies block all access to protected tables (fail-closed).

This is API-agnostic -- works identically for REST, GraphQL, Django views, or any other request handler.

Add to MIDDLEWARE::

MIDDLEWARE = [
    ...
    "django_rls_tenants.tenants.middleware.RLSTenantMiddleware",
]

With USE_LOCAL_SET=True the middleware owns a database transaction for every authenticated request on the synchronous (WSGI) path: transaction.atomic() is opened on each alias in RLS_TENANTS["DATABASES"] around the rest of the middleware chain and the view, so the SET LOCAL GUCs are in force for the whole request and vanish at commit/rollback. ATOMIC_REQUESTS is not required (and is harmless -- it nests as a savepoint).

Under ASGI, __acall__ serves the request natively instead of MiddlewareMixin.__acall__: one thread-sensitive hop resolves the user and sets the GUCs, the tenant ContextVar values are set and reset with their own tokens in the request coroutine, and one hop clears the GUCs at the end. A subclass that overrides process_request or process_response keeps the hook-based path, so its hooks still run. With USE_LOCAL_SET=True a request on the async path gets no transaction (none can be held open across the awaited chain; under ASGI with a synchronous-only middleware after this one, Django runs it on the synchronous path, with the transaction) and the middleware writes no GUC: the GUC affinity guard sets the request's values transaction-locally for each statement, giving a statement run in autocommit a transaction of its own (#96).

Source code in django_rls_tenants/tenants/middleware.py

def __init__(self, get_response: Callable[..., Any]) -> None:
    super().__init__(get_response)
    # The native async path does not call these hooks, so a subclass that
    # customises them is served through MiddlewareMixin.__acall__ instead.
    self._sync_hooks_overridden = any(
        getattr(type(self), hook) is not getattr(RLSTenantMiddleware, hook)
        for hook in ("process_request", "process_response")
    )
    # Only the synchronous path opens the request transaction this protects.
    if rls_tenants_config.USE_LOCAL_SET and not self.async_mode:
        self._validate_exception_middleware_order()

call ¶

__call__(request: HttpRequest) -> Any

Serve the request: on the WSGI path here, on the ASGI path in __acall__.

The WSGI request holds a context holder for its duration (see _holder_for_the_request), which a coroutine the view runs with async_to_sync inherits.

Source code in django_rls_tenants/tenants/middleware.py

def __call__(self, request: HttpRequest) -> Any:
    """Serve the request: on the WSGI path here, on the ASGI path in ``__acall__``.

    The WSGI request holds a context holder for its duration (see
    ``_holder_for_the_request``), which a coroutine the view runs with
    ``async_to_sync`` inherits.
    """
    if self.async_mode:
        return self.__acall__(request)
    with _holder_for_the_request(request):
        return self._call_sync(request)

acall `async` ¶

__acall__(request: HttpRequest) -> HttpResponseBase

Serve the request on Django's async (ASGI) path.

Replaces MiddlewareMixin.__acall__, which ran process_request and process_response in separate sync_to_async hops. Here one thread-sensitive hop resolves request.user and sets the GUCs on every alias. The tenant ContextVar values are then set in the request coroutine, so the downstream chain and async views see them, and the finally block resets them with their own tokens, which restores any outer value. A second hop clears the GUCs, unless process_exception already ended the context for a failed view. Cleanup also runs when the request task is cancelled; only a second cancellation that arrives while the clear hop is still queued can skip it, and the request_finished safety net then retries where it can see the request's flag. As on the WSGI path, the context ends when the chain returns, before a streaming response body is iterated.

A subclass that overrides process_request or process_response is served by _acall_through_sync_hooks instead, so its hooks run as they did before this path existed.

With USE_LOCAL_SET=True the async path opens no transaction (it cannot hold a synchronous transaction.atomic() open across the awaited chain) and writes no GUC; the GUC affinity guard applies the request's context to each statement, transaction-locally.

Source code in django_rls_tenants/tenants/middleware.py

async def __acall__(self, request: HttpRequest) -> HttpResponseBase:
    """Serve the request on Django's async (ASGI) path.

    Replaces ``MiddlewareMixin.__acall__``, which ran ``process_request`` and
    ``process_response`` in separate ``sync_to_async`` hops. Here one
    thread-sensitive hop resolves ``request.user`` and sets the GUCs on every
    alias. The tenant ``ContextVar`` values are then set in the request
    coroutine, so the downstream chain and async views see them, and the
    ``finally`` block resets them with their own tokens, which restores any
    outer value. A second hop clears the GUCs, unless ``process_exception``
    already ended the context for a failed view. Cleanup also runs when the
    request task is cancelled; only a second cancellation that arrives while
    the clear hop is still queued can skip it, and the ``request_finished``
    safety net then retries where it can see the request's flag. As on the
    WSGI path, the context ends when the chain returns, before a streaming
    response body is iterated.

    A subclass that overrides ``process_request`` or ``process_response``
    is served by ``_acall_through_sync_hooks`` instead, so its hooks run as
    they did before this path existed.

    With ``USE_LOCAL_SET=True`` the async path opens no transaction (it
    cannot hold a synchronous ``transaction.atomic()`` open across the
    awaited chain) and writes no GUC; the GUC affinity guard applies the
    request's context to each statement, transaction-locally.
    """
    if self._sync_hooks_overridden:
        return await self._acall_through_sync_hooks(request)
    conf = rls_tenants_config
    scope = await self._aenter_request(request, conf)
    if scope is None:
        with _holder_for_the_request(request):
            return await self._aget_response(request)
    setattr(request, "_rls_found_context", _FoundContext(_current_live()))  # noqa: B010  -- dynamic attr on HttpRequest
    tenant_token = set_current_tenant_id(scope.tenant_id)
    active_token = set_rls_context_active(True)
    gucs_token = _set_context_gucs(scope.gucs)
    # For process_exception, which ends the context from a hop of its own.
    setattr(request, "_rls_context_gucs_token", gucs_token)  # noqa: B010  -- dynamic attr on HttpRequest
    # The hop's thread, not the event loop's, owns the GUCs.
    _mark_gucs_set(scope.thread_marker)
    try:
        return await self._aget_response(request)
    finally:
        try:
            setattr(request, "_rls_context_gucs_token", None)  # noqa: B010  -- dynamic attr on HttpRequest
            _end_context_gucs(gucs_token)
            reset_current_tenant_id(tenant_token)
            reset_rls_context_active(active_token)
        finally:
            # False when process_exception has already cleared the GUCs. Left
            # set if the clear fails, so the request_finished safety net can
            # retry. Runs even if a reset fails, as in _cleanup_rls_state.
            if _were_gucs_set():
                await sync_to_async(_clear_request_gucs, thread_sensitive=True)(conf)
                _clear_gucs_set_flag()

process_request ¶

process_request(request: HttpRequest) -> None

Set GUC variables on all configured databases.

Iterates over RLS_TENANTS["DATABASES"] and sets GUCs on each alias. If setting GUCs on one database fails, clears all GUCs on databases that were already set, then re-raises. Used on the synchronous path only; __acall__ uses _enter_request_gucs.

Source code in django_rls_tenants/tenants/middleware.py

def process_request(self, request: HttpRequest) -> None:
    """Set GUC variables on all configured databases.

    Iterates over ``RLS_TENANTS["DATABASES"]`` and sets GUCs on each
    alias. If setting GUCs on one database fails, clears all GUCs on
    databases that were already set, then re-raises. Used on the
    synchronous path only; ``__acall__`` uses ``_enter_request_gucs``.
    """
    if not _is_authenticated(request):
        return

    conf = rls_tenants_config
    user: Any = request.user

    # Track tokens so the except branch can reset via token (not
    # push a new value) when the ContextVar was already set before
    # the failure.
    tenant_token: Token[int | str | None] | None = None
    active_token: Token[bool] | None = None
    gucs_token: _ContextGucsToken | None = None
    try:
        guc_vars = _resolve_user_guc_vars(user, conf)
        self._set_gucs_on_all_databases(guc_vars, conf)
        logger.debug(
            "Middleware set GUCs on databases=%s for user=%s (admin=%s, tenant=%s)",
            conf.DATABASES,
            type(user).__name__,
            user.is_tenant_admin,
            getattr(user, "rls_tenant_id", None),
        )

        # Set auto-scope state for RLSManager.get_queryset().
        # Use the original tenant ID from the user (preserving type)
        # rather than the string GUC representation, so that
        # get_current_tenant_id() returns a consistent type.
        if not user.is_tenant_admin:
            tenant_token = set_current_tenant_id(user.rls_tenant_id)
        else:
            tenant_token = set_current_tenant_id(None)
        setattr(request, "_rls_tenant_token", tenant_token)  # noqa: B010  -- dynamic attr on HttpRequest

        # Mark RLS context as active for strict mode. This
        # distinguishes "middleware set context" from "no context".
        active_token = set_rls_context_active(True)
        setattr(request, "_rls_context_active_token", active_token)  # noqa: B010  -- dynamic attr on HttpRequest

        # The values the GUC affinity guard re-applies on other connections.
        # Not a block's holder: under ASGI, MiddlewareMixin runs this hook in a
        # sync_to_async hop of its own, which must hand the values back.
        gucs_token = _set_context_gucs(_context_gucs_of(guc_vars, conf), block=False)
        setattr(request, "_rls_context_gucs_token", gucs_token)  # noqa: B010  -- dynamic attr on HttpRequest
        _mark_gucs_set()
    except Exception:
        logger.exception("Failed to set RLS GUC variables, clearing to prevent leak")
        # Reset via token when available to avoid stacking unreset
        # ContextVar values; _reset_or_clear falls back to clearing when
        # the token was never created (error occurred before set call).
        _end_context_gucs(gucs_token)
        _reset_or_clear(
            tenant_token,
            reset_current_tenant_id,
            partial(set_current_tenant_id, None),
        )
        _reset_or_clear(
            active_token,
            reset_rls_context_active,
            partial(set_rls_context_active, False),
        )
        setattr(request, "_rls_state_reset", True)  # noqa: B010 -- tokens already reset
        _clear_gucs_on_all_databases(conf)
        raise

process_response ¶

process_response(
    request: HttpRequest, response: HttpResponse
) -> HttpResponse

Clear GUC variables and auto-scope state to prevent cross-request leaks.

Source code in django_rls_tenants/tenants/middleware.py

def process_response(
    self,
    request: HttpRequest,
    response: HttpResponse,
) -> HttpResponse:
    """Clear GUC variables and auto-scope state to prevent cross-request leaks."""
    self._cleanup_rls_state(request)
    return response

process_exception ¶

process_exception(
    request: HttpRequest, exception: Exception
) -> None

Record a view failure before Django converts it to an error response.

Defer set_rollback(True) until the chain returns to __call__. Setting it here would prevent error handlers and response middleware from querying even after a non-database exception such as Http404. All writes, including those made by error handlers, are rolled back when the middleware-owned transaction exits. An explicit error response without an exception still commits.

Without a middleware-owned transaction the hook ends the request context instead. Under ASGI, Django calls it in its own sync_to_async hop and then builds the error response on another thread and database connection; ending the context first keeps that error handling fail-closed, as on WSGI. __acall__ then only resets its tokens.

Either way, a context block the view left open (a generator suspended inside one) is ended, so the error response does not run in it. With a middleware-owned transaction the request's own context stays in force while the error response is built (the rollback is deferred, so error handlers can still query); without one the response runs with no context (fail-closed).

Source code in django_rls_tenants/tenants/middleware.py

def process_exception(
    self,
    request: HttpRequest,
    exception: Exception,  # noqa: ARG002  -- required by MiddlewareMixin
) -> None:
    """Record a view failure before Django converts it to an error response.

    Defer ``set_rollback(True)`` until the chain returns to ``__call__``.
    Setting it here would prevent error handlers and response middleware
    from querying even after a non-database exception such as ``Http404``.
    All writes, including those made by error handlers, are rolled back
    when the middleware-owned transaction exits. An explicit error
    response without an exception still commits.

    Without a middleware-owned transaction the hook ends the request
    context instead. Under ASGI, Django calls it in its own
    ``sync_to_async`` hop and then builds the error response on another
    thread and database connection; ending the context first keeps that
    error handling fail-closed, as on WSGI. ``__acall__`` then only resets
    its tokens.

    Either way, a context block the view left open (a generator suspended
    inside one) is ended, so the error response does not run in it. With
    a middleware-owned transaction the request's own context stays in
    force while the error response is built (the rollback is deferred, so
    error handlers can still query); without one the response runs with
    no context (fail-closed).
    """
    try:
        if getattr(request, "_rls_owns_transaction", False) is True:
            setattr(request, "_rls_rollback_transaction", True)  # noqa: B010 -- dynamic request marker
        else:
            self._cleanup_rls_state(request)
    finally:
        _end_blocks_left_open(request)

Admin¶

Tenant-aware Django admin. See the Admin guide for usage, the tenant switcher, and the middleware interaction.

django_rls_tenants.tenants.admin.RLSTenantModelAdmin ¶

Bases: ModelAdmin

ModelAdmin mixin that scopes every admin DB operation to the RLS context.

Register it like any ModelAdmin::

from django.contrib import admin
from django_rls_tenants import RLSTenantModelAdmin
from myapp.models import Order

@admin.register(Order)
class OrderAdmin(RLSTenantModelAdmin):
    list_display = ("product", "amount")

request.user must satisfy the :class:~django_rls_tenants.tenants.types.TenantUser protocol (is_tenant_admin / rls_tenant_id) -- the same contract the middleware and for_user() rely on.

Attributes:

Name	Type	Description
`rls_allow_tenant_switch`	`bool`	Give cross-tenant admins the tenant switcher. When `False`, such admins always operate in `admin_context()` (all tenants). Defaults to `True`.
`rls_session_key`	`str`	Session key holding the switcher selection. Defaults to `"rls_admin_tenant"`.
`rls_tenant_query_param`	`str`	Query parameter the switcher uses. Defaults to `"rls_tenant"` (matches :class:`TenantSwitchListFilter`).
`rls_deny_without_tenant`	`bool`	When `True` (default), a non-admin user with no tenant raises `PermissionDenied` instead of running unscoped.

changelist_view ¶

changelist_view(
    request: HttpRequest,
    extra_context: dict[str, Any] | None = None,
) -> HttpResponse

Wrap the changelist in the request's RLS context.

Source code in django_rls_tenants/tenants/admin.py

def changelist_view(
    self,
    request: HttpRequest,
    extra_context: dict[str, Any] | None = None,
) -> HttpResponse:
    """Wrap the changelist in the request's RLS context."""
    self._sync_tenant_selection(request)
    with self._rls_context(request):
        return super().changelist_view(request, extra_context)

add_view ¶

add_view(
    request: HttpRequest,
    form_url: str = "",
    extra_context: dict[str, Any] | None = None,
) -> HttpResponse

Wrap the add view in the request's RLS context.

Source code in django_rls_tenants/tenants/admin.py

def add_view(
    self,
    request: HttpRequest,
    form_url: str = "",
    extra_context: dict[str, Any] | None = None,
) -> HttpResponse:
    """Wrap the add view in the request's RLS context."""
    self._sync_tenant_selection(request)
    with self._rls_context(request):
        return super().add_view(request, form_url, extra_context)

change_view ¶

change_view(
    request: HttpRequest,
    object_id: str,
    form_url: str = "",
    extra_context: dict[str, Any] | None = None,
) -> HttpResponse

Wrap the change view in the request's RLS context.

Source code in django_rls_tenants/tenants/admin.py

def change_view(
    self,
    request: HttpRequest,
    object_id: str,
    form_url: str = "",
    extra_context: dict[str, Any] | None = None,
) -> HttpResponse:
    """Wrap the change view in the request's RLS context."""
    self._sync_tenant_selection(request)
    with self._rls_context(request):
        return super().change_view(request, object_id, form_url, extra_context)

delete_view ¶

delete_view(
    request: HttpRequest,
    object_id: str,
    extra_context: dict[str, Any] | None = None,
) -> HttpResponse

Wrap the delete view in the request's RLS context.

Source code in django_rls_tenants/tenants/admin.py

def delete_view(
    self,
    request: HttpRequest,
    object_id: str,
    extra_context: dict[str, Any] | None = None,
) -> HttpResponse:
    """Wrap the delete view in the request's RLS context."""
    self._sync_tenant_selection(request)
    with self._rls_context(request):
        return super().delete_view(request, object_id, extra_context)

history_view ¶

history_view(
    request: HttpRequest,
    object_id: str,
    extra_context: dict[str, Any] | None = None,
) -> HttpResponse

Wrap the history view in the request's RLS context.

Source code in django_rls_tenants/tenants/admin.py

def history_view(
    self,
    request: HttpRequest,
    object_id: str,
    extra_context: dict[str, Any] | None = None,
) -> HttpResponse:
    """Wrap the history view in the request's RLS context."""
    self._sync_tenant_selection(request)
    with self._rls_context(request):
        return super().history_view(request, object_id, extra_context)

get_exclude ¶

get_exclude(
    request: HttpRequest, obj: Model | None = None
) -> _ListOrTuple[str] | None

Hide the tenant FK when the effective tenant is implicit.

For a scoped user, or an admin who has selected a tenant, the tenant is known and set by :meth:save_model, so the field is excluded. A global admin with no selection keeps it visible to choose a tenant explicitly.

Source code in django_rls_tenants/tenants/admin.py

def get_exclude(
    self,
    request: HttpRequest,
    obj: Model | None = None,
) -> _ListOrTuple[str] | None:
    """Hide the tenant FK when the effective tenant is implicit.

    For a scoped user, or an admin who has selected a tenant, the tenant is
    known and set by :meth:`save_model`, so the field is excluded. A global
    admin with no selection keeps it visible to choose a tenant explicitly.
    """
    exclude = super().get_exclude(request, obj)
    if self._effective_tenant_id(request) is None:
        return exclude
    field = rls_tenants_config.TENANT_FK_FIELD
    existing: tuple[str, ...] = tuple(exclude) if exclude else ()
    if field in existing:
        return existing
    return (*existing, field)

get_fieldsets ¶

get_fieldsets(
    request: HttpRequest, obj: Model | None = None
) -> _FieldsetSpec

Drop the tenant FK from explicit fieldsets when the tenant is implicit.

Mirrors :meth:get_exclude: when the effective tenant is known the FK is excluded from the form, so a fieldsets layout that still names it would raise KeyError at render time. A global admin with no selection keeps the field (and its layout slot) so they can assign a tenant explicitly.

Source code in django_rls_tenants/tenants/admin.py

def get_fieldsets(
    self,
    request: HttpRequest,
    obj: Model | None = None,
) -> _FieldsetSpec:
    """Drop the tenant FK from explicit fieldsets when the tenant is implicit.

    Mirrors :meth:`get_exclude`: when the effective tenant is known the FK is
    excluded from the form, so a ``fieldsets`` layout that still names it would
    raise ``KeyError`` at render time. A global admin with no selection keeps
    the field (and its layout slot) so they can assign a tenant explicitly.
    """
    fieldsets = super().get_fieldsets(request, obj)
    if self._effective_tenant_id(request) is None:
        return fieldsets
    return _fieldsets_without(fieldsets, rls_tenants_config.TENANT_FK_FIELD)

save_model ¶

save_model(
    request: HttpRequest, obj: Model, form: Any, change: Any
) -> None

Stamp the effective tenant onto the object before saving.

When the tenant FK is hidden (implicit tenant) the form never sets it, so it is assigned here. When the effective tenant is None (global admin viewing all) the field is visible and the admin's own choice stands.

Source code in django_rls_tenants/tenants/admin.py

def save_model(
    self,
    request: HttpRequest,
    obj: Model,
    form: Any,
    change: Any,
) -> None:
    """Stamp the effective tenant onto the object before saving.

    When the tenant FK is hidden (implicit tenant) the form never sets it, so
    it is assigned here. When the effective tenant is ``None`` (global admin
    viewing all) the field is visible and the admin's own choice stands.
    """
    tenant_id = self._effective_tenant_id(request)
    if tenant_id is not None:
        # The switcher persists its selection as a session string; normalise it
        # to the tenant pk's Python type (int/UUID) via the FK's target field so
        # pre_save signal handlers see the column's real type, not a bare str.
        fk = self.model._meta.get_field(rls_tenants_config.TENANT_FK_FIELD)  # noqa: SLF001
        target = fk.target_field  # the Tenant pk field the FK points at
        setattr(obj, fk.attname, target.to_python(tenant_id))
    super().save_model(request, obj, form, change)

get_list_filter ¶

get_list_filter(request: HttpRequest) -> list[Any]

Prepend the tenant switcher for switch-capable admins.

Source code in django_rls_tenants/tenants/admin.py

def get_list_filter(self, request: HttpRequest) -> list[Any]:
    """Prepend the tenant switcher for switch-capable admins."""
    list_filter = list(super().get_list_filter(request))
    if not self._user_can_switch_tenant(request):
        return list_filter
    return [self._tenant_switch_filter_class, *list_filter]

django_rls_tenants.tenants.admin.TenantSwitchListFilter ¶

TenantSwitchListFilter(
    request: HttpRequest,
    params: dict[str, list[str]],
    model: type[Model],
    model_admin: ModelAdmin[Any],
)

Bases: SimpleListFilter

Changelist filter that renders the cross-tenant admin's tenant switcher.

The dropdown lists every tenant from RLS_TENANTS["TENANT_MODEL"] plus an "All" entry. Selecting an option reloads the changelist with ?<param>=<tenant-pk> (or ?<param>=__all__); the owning :class:RLSTenantModelAdmin persists that choice to the session and applies the actual scoping by activating the matching context. The filter's own :meth:queryset is therefore a no-op -- it exists to render the control and to register the query parameter so the changelist accepts it.

Only attached for users who may switch tenants (see :meth:RLSTenantModelAdmin.get_list_filter).

Source code in django_rls_tenants/tenants/admin.py

def __init__(
    self,
    request: HttpRequest,
    params: dict[str, list[str]],
    model: type[Model],
    model_admin: admin.ModelAdmin[Any],
) -> None:
    # Remember the persisted selection so the dropdown reflects the tenant
    # currently in effect even on a paramless changelist load (the owning
    # admin syncs the session before the filter is built).
    session_key: str = getattr(model_admin, "rls_session_key", _DEFAULT_SESSION_KEY)
    self._session_value: str | None = request.session.get(session_key)
    super().__init__(request, params, model, model_admin)

lookups ¶

lookups(
    request: HttpRequest, model_admin: ModelAdmin[Any]
) -> list[tuple[str, str]]

Return (pk, label) pairs for every tenant.

The tenant model is global (not RLS-protected), so this is unaffected by the active context and always lists all tenants.

Source code in django_rls_tenants/tenants/admin.py

def lookups(
    self,
    request: HttpRequest,  # noqa: ARG002  -- required by the base signature
    model_admin: admin.ModelAdmin[Any],  # noqa: ARG002  -- required by the base signature
) -> list[tuple[str, str]]:
    """Return ``(pk, label)`` pairs for every tenant.

    The tenant model is global (not RLS-protected), so this is unaffected by
    the active context and always lists all tenants.
    """
    tenant_model = apps.get_model(rls_tenants_config.TENANT_MODEL)
    return [
        (str(tenant.pk), str(tenant))
        for tenant in tenant_model._default_manager.all()  # noqa: SLF001
    ]

value ¶

value() -> str | None

Return the selected tenant id.

The query parameter wins; otherwise fall back to the session-persisted selection so the dropdown still shows the active tenant on add/change pages that carry no query string.

Source code in django_rls_tenants/tenants/admin.py

def value(self) -> str | None:
    """Return the selected tenant id.

    The query parameter wins; otherwise fall back to the session-persisted
    selection so the dropdown still shows the active tenant on add/change
    pages that carry no query string.
    """
    return super().value() or self._session_value

queryset ¶

queryset(request: HttpRequest, queryset: Any) -> Any

No-op: scoping is done by the owning admin's context, not here.

Source code in django_rls_tenants/tenants/admin.py

def queryset(self, request: HttpRequest, queryset: Any) -> Any:  # noqa: ARG002
    """No-op: scoping is done by the owning admin's context, not here."""
    return queryset

choices ¶

choices(
    changelist: ChangeList,
) -> Iterator[_ListFilterChoices]

Yield the "All" entry (clearing the selection) followed by each tenant.

Source code in django_rls_tenants/tenants/admin.py

def choices(self, changelist: ChangeList) -> Iterator[_ListFilterChoices]:
    """Yield the "All" entry (clearing the selection) followed by each tenant."""
    current = self.value()
    yield {
        "selected": current is None or current == _ALL_TENANTS,
        "query_string": changelist.get_query_string({self.parameter_name: _ALL_TENANTS}),
        "display": "All",
    }
    for lookup, title in self.lookup_choices:
        yield {
            "selected": current == str(lookup),
            "query_string": changelist.get_query_string({self.parameter_name: lookup}),
            "display": title,
        }

Types¶

django_rls_tenants.tenants.types.TenantUser ¶

Bases: Protocol

Protocol that user objects must satisfy for RLS context resolution.

Implement these two properties on your User model::

class User(AbstractUser, RLSProtectedModel):
    @property
    def is_tenant_admin(self) -> bool:
        return self.role.name == "ADMIN"

    @property
    def rls_tenant_id(self) -> int | str | None:
        return self.tenant_id if self.tenant_id else None

is_tenant_admin `property` ¶

is_tenant_admin: bool

Return True if this user bypasses RLS (super-admin).

rls_tenant_id `property` ¶

rls_tenant_id: int | str | None

Return the tenant ID for RLS filtering, or None for admins.

Bypass Helpers¶

django_rls_tenants.tenants.bypass.set_bypass_flag ¶

set_bypass_flag(
    flag_name: str,
    *,
    is_local: bool = False,
    using: str = "default",
) -> None

Set a bypass flag on the current database connection.

The flag name should match one of the extra_bypass_flags configured on an RLSConstraint::

set_bypass_flag("rls.is_login_request")

Parameters:

Name	Type	Description	Default
`flag_name`	`str`	GUC variable name (e.g., `"rls.is_login_request"`).	required
`is_local`	`bool`	If `True`, flag is transaction-scoped.	`False`
`using`	`str`	Database alias. Default: `"default"`.	`'default'`

Source code in django_rls_tenants/tenants/bypass.py

def set_bypass_flag(
    flag_name: str,
    *,
    is_local: bool = False,
    using: str = "default",
) -> None:
    """Set a bypass flag on the current database connection.

    The flag name should match one of the ``extra_bypass_flags``
    configured on an ``RLSConstraint``::

        set_bypass_flag("rls.is_login_request")

    Args:
        flag_name: GUC variable name (e.g., ``"rls.is_login_request"``).
        is_local: If ``True``, flag is transaction-scoped.
        using: Database alias. Default: ``"default"``.
    """
    set_guc(flag_name, "true", is_local=is_local, using=using)

django_rls_tenants.tenants.bypass.clear_bypass_flag ¶

clear_bypass_flag(
    flag_name: str,
    *,
    is_local: bool = False,
    using: str = "default",
) -> None

Clear a bypass flag on the current database connection.

Pass the same is_local that was used in set_bypass_flag so a transaction-scoped flag is cleared transaction-locally too::

with transaction.atomic():
    set_bypass_flag("rls.is_login_request", is_local=True)
    ...
    clear_bypass_flag("rls.is_login_request", is_local=True)

Parameters:

Name	Type	Description	Default
`flag_name`	`str`	GUC variable name (e.g., `"rls.is_login_request"`).	required
`is_local`	`bool`	If `True`, use `SET LOCAL` (transaction-scoped).	`False`
`using`	`str`	Database alias. Default: `"default"`.	`'default'`

Raises:

Type	Description
`RuntimeError`	If `is_local=True` outside `transaction.atomic()`.

Source code in django_rls_tenants/tenants/bypass.py

def clear_bypass_flag(
    flag_name: str,
    *,
    is_local: bool = False,
    using: str = "default",
) -> None:
    """Clear a bypass flag on the current database connection.

    Pass the same ``is_local`` that was used in ``set_bypass_flag`` so a
    transaction-scoped flag is cleared transaction-locally too::

        with transaction.atomic():
            set_bypass_flag("rls.is_login_request", is_local=True)
            ...
            clear_bypass_flag("rls.is_login_request", is_local=True)

    Args:
        flag_name: GUC variable name (e.g., ``"rls.is_login_request"``).
        is_local: If ``True``, use ``SET LOCAL`` (transaction-scoped).
        using: Database alias. Default: ``"default"``.

    Raises:
        RuntimeError: If ``is_local=True`` outside ``transaction.atomic()``.
    """
    clear_guc(flag_name, is_local=is_local, using=using)

Testing¶

django_rls_tenants.tenants.testing.rls_bypass ¶

rls_bypass(*, using: str = 'default') -> Iterator[None]

Temporarily enable admin bypass for tests.

Parameters:

Name	Type	Description	Default
`using`	`str`	Database alias. Default: `"default"`.	`'default'`

Usage::

with rls_bypass():
    all_orders = Order.objects.all()  # sees all tenants

Source code in django_rls_tenants/tenants/testing.py

@contextmanager
def rls_bypass(*, using: str = "default") -> Iterator[None]:
    """Temporarily enable admin bypass for tests.

    Args:
        using: Database alias. Default: ``"default"``.

    Usage::

        with rls_bypass():
            all_orders = Order.objects.all()  # sees all tenants
    """
    with admin_context(using=using):
        yield

django_rls_tenants.tenants.testing.rls_as_tenant ¶

rls_as_tenant(
    tenant_id: int | str, *, using: str = "default"
) -> Iterator[None]

Scope to a specific tenant for tests.

Parameters:

Name	Type	Description	Default
`tenant_id`	`int \| str`	The tenant PK to scope to.	required
`using`	`str`	Database alias. Default: `"default"`.	`'default'`

Usage::

with rls_as_tenant(tenant_id=42):
    orders = Order.objects.all()  # only tenant 42

Source code in django_rls_tenants/tenants/testing.py

@contextmanager
def rls_as_tenant(tenant_id: int | str, *, using: str = "default") -> Iterator[None]:
    """Scope to a specific tenant for tests.

    Args:
        tenant_id: The tenant PK to scope to.
        using: Database alias. Default: ``"default"``.

    Usage::

        with rls_as_tenant(tenant_id=42):
            orders = Order.objects.all()  # only tenant 42
    """
    with tenant_context(tenant_id, using=using):
        yield

django_rls_tenants.tenants.testing.assert_rls_enabled ¶

assert_rls_enabled(
    table_name: str, *, using: str = "default"
) -> None

Assert that RLS is enabled and forced on the given table.

Parameters:

Name	Type	Description	Default
`table_name`	`str`	The database table name to check.	required
`using`	`str`	Database alias. Default: `"default"`.	`'default'`

Raises:

Type	Description
`AssertionError`	If RLS is not enabled or not forced.

Source code in django_rls_tenants/tenants/testing.py

def assert_rls_enabled(table_name: str, *, using: str = "default") -> None:
    """Assert that RLS is enabled and forced on the given table.

    Args:
        table_name: The database table name to check.
        using: Database alias. Default: ``"default"``.

    Raises:
        AssertionError: If RLS is not enabled or not forced.
    """
    conn = connections[using]
    with conn.cursor() as cursor:
        cursor.execute(
            "SELECT relrowsecurity, relforcerowsecurity FROM pg_class WHERE relname = %s",
            [table_name],
        )
        row = cursor.fetchone()
        assert row is not None, f"Table '{table_name}' does not exist"
        assert row[0] is True, f"RLS is not enabled on table '{table_name}'"
        assert row[1] is True, f"RLS is not forced on table '{table_name}'"

django_rls_tenants.tenants.testing.assert_rls_policy_exists ¶

assert_rls_policy_exists(
    table_name: str,
    policy_name: str | None = None,
    *,
    using: str = "default",
) -> None

Assert that an RLS policy exists on the given table.

Parameters:

Name	Type	Description	Default
`table_name`	`str`	The database table name to check.	required
`policy_name`	`str \| None`	Expected policy name. Defaults to `"{table_name}_tenant_isolation_policy"`.	`None`
`using`	`str`	Database alias. Default: `"default"`.	`'default'`

Source code in django_rls_tenants/tenants/testing.py

def assert_rls_policy_exists(
    table_name: str,
    policy_name: str | None = None,
    *,
    using: str = "default",
) -> None:
    """Assert that an RLS policy exists on the given table.

    Args:
        table_name: The database table name to check.
        policy_name: Expected policy name. Defaults to
            ``"{table_name}_tenant_isolation_policy"``.
        using: Database alias. Default: ``"default"``.
    """
    if policy_name is None:
        policy_name = f"{table_name}_tenant_isolation_policy"

    conn = connections[using]
    with conn.cursor() as cursor:
        cursor.execute(
            "SELECT 1 FROM pg_policies WHERE tablename = %s AND policyname = %s",
            [table_name, policy_name],
        )
        assert cursor.fetchone() is not None, (
            f"RLS policy '{policy_name}' does not exist on table '{table_name}'"
        )

django_rls_tenants.tenants.testing.assert_rls_blocks_without_context ¶

assert_rls_blocks_without_context(
    model_class: type[Model], *, using: str = "default"
) -> None

Assert that the database returns zero rows of model_class with no RLS context.

Verifies the fail-closed behavior of the table's RLS policy. Requires at least one row in the table (the caller must set up test data first).

Inside :func:~django_rls_tenants.tenants.context.admin_context, the table must hold at least one row, so the assertion cannot pass vacuously.
With the admin and tenant GUCs cleared on the using connection, the table must return zero rows. The GUCs are restored before the assertion (with SET LOCAL when USE_LOCAL_SET=True), so the result does not depend on the caller's admin or tenant context, and such a context (for example :func:rls_bypass or :func:rls_as_tenant) is intact when the helper returns or fails. With USE_LOCAL_SET=True outside transaction.atomic(), where such a context lives in the ContextVar values only and the GUC affinity guard applies it to each statement, the count runs without it, and the GUCs are cleared and restored for the session. Other GUCs, such as a flag set with bypass_flag(), are left as they are.

Both counts are raw SQL on model_class._meta.db_table, so the STRICT_MODE guard, the ORM auto-scope filter and the filters of the model's default manager are not involved: the helper works with STRICT_MODE=True, and every row in the table must be hidden.

Parameters:

Name	Type	Description	Default
`model_class`	`type[Model]`	The RLS-protected model class to query.	required
`using`	`str`	Database alias. Default: `"default"`.	`'default'`

Raises:

Type	Description
`AssertionError`	If any rows are returned, or if the table is empty (which would make the assertion pass vacuously).
`RuntimeError`	If `USE_LOCAL_SET=True`, `using` is not in `RLS_TENANTS["DATABASES"]` and its connection is not inside `transaction.atomic()`.

Source code in django_rls_tenants/tenants/testing.py

def assert_rls_blocks_without_context(
    model_class: type[models.Model],
    *,
    using: str = "default",
) -> None:
    """Assert that the database returns zero rows of ``model_class`` with no RLS context.

    Verifies the fail-closed behavior of the table's RLS policy. Requires at
    least one row in the table (the caller must set up test data first).

    1. Inside :func:`~django_rls_tenants.tenants.context.admin_context`, the
       table must hold at least one row, so the assertion cannot pass
       vacuously.
    2. With the admin and tenant GUCs cleared on the ``using`` connection, the
       table must return zero rows. The GUCs are restored before the
       assertion (with ``SET LOCAL`` when ``USE_LOCAL_SET=True``), so the
       result does not depend on the caller's admin or tenant context, and
       such a context (for example :func:`rls_bypass` or
       :func:`rls_as_tenant`) is intact when the helper returns or fails.
       With ``USE_LOCAL_SET=True`` outside ``transaction.atomic()``, where
       such a context lives in the ``ContextVar`` values only and the GUC
       affinity guard applies it to each statement, the count runs without
       it, and the GUCs are cleared and restored for the session.
       Other GUCs, such as a flag set with ``bypass_flag()``, are left as they
       are.

    Both counts are raw SQL on ``model_class._meta.db_table``, so the
    ``STRICT_MODE`` guard, the ORM auto-scope filter and the filters of the
    model's default manager are not involved: the helper works with
    ``STRICT_MODE=True``, and every row in the table must be hidden.

    Args:
        model_class: The RLS-protected model class to query.
        using: Database alias. Default: ``"default"``.

    Raises:
        AssertionError: If any rows are returned, or if the table is empty
            (which would make the assertion pass vacuously).
        RuntimeError: If ``USE_LOCAL_SET=True``, ``using`` is not in
            ``RLS_TENANTS["DATABASES"]`` and its connection is not inside
            ``transaction.atomic()``.
    """
    conf = rls_tenants_config

    # Pre-check: verify the table has data (via admin bypass) so the
    # assertion is not vacuously true on an empty table.
    with admin_context(using=using):
        total = _count_rows(model_class, using=using)
    assert total > 0, (
        f"assert_rls_blocks_without_context requires at least one row in "
        f"{model_class.__name__}, but the table is empty. "
        f"Set up test data before calling this helper."
    )

    # Clear the admin and tenant GUCs for the count only, so a context the
    # caller holds cannot make rows visible, then restore them.
    no_context = {conf.GUC_IS_ADMIN: "", conf.GUC_CURRENT_TENANT: ""}
    if _writes_gucs(using, is_local=conf.USE_LOCAL_SET):
        with rls_context(no_context, is_local=conf.USE_LOCAL_SET, using=using):
            count = _count_rows(model_class, using=using)
    else:
        # USE_LOCAL_SET=True outside a transaction: the guard would run the count in
        # a transaction of its own with the caller's context.
        with _no_context_for_the_guard(), rls_context(no_context, using=using):
            count = _count_rows(model_class, using=using)
    table = model_class._meta.db_table  # noqa: SLF001  -- standard Django _meta access
    assert count == 0, (
        f"Expected 0 rows from {model_class.__name__} (table '{table}') without RLS "
        f"context, got {count}, so RLS may not be enforced. Check that the connection "
        f"role is not a superuser and has no BYPASSRLS (system check W005). Check that "
        f"RLS is enabled and forced on the table and that it has a policy "
        f"(assert_rls_enabled(), assert_rls_policy_exists()). Check that no bypass flag "
        f"is set."
    )

Async Testing Helpers¶

For async def tests. Which database connection they use depends on how the test runs the coroutine; see Async Tests.

django_rls_tenants.tenants.testing.async_rls_bypass `async` ¶

async_rls_bypass(
    *, using: str = "default"
) -> AsyncIterator[None]

Async counterpart of :func:rls_bypass, for async with in async tests.

Wraps :func:~django_rls_tenants.tenants.context.async_admin_context: one thread_sensitive hop on entry and one on exit, with the admin state set in the test coroutine itself, so the async ORM sees it. The hops reach the test's own connection and transaction only when the coroutine is driven by async_to_sync, as Django's TestCase does for async def test methods (see the testing guide, "Async Tests").

Parameters:

Name	Type	Description	Default
`using`	`str`	Database alias. Default: `"default"`.	`'default'`

Raises:

Type	Description
`RLSConfigurationError`	If `USE_LOCAL_SET=True`, `using` is not in `RLS_TENANTS["DATABASES"]` and no transaction is open on the hop thread.

Usage::

async with async_rls_bypass():
    total = await Order.objects.acount()  # sees all tenants

Source code in django_rls_tenants/tenants/testing.py

@asynccontextmanager
async def async_rls_bypass(*, using: str = "default") -> AsyncIterator[None]:
    """Async counterpart of :func:`rls_bypass`, for ``async with`` in async tests.

    Wraps :func:`~django_rls_tenants.tenants.context.async_admin_context`: one
    ``thread_sensitive`` hop on entry and one on exit, with the admin state set
    in the test coroutine itself, so the async ORM sees it. The hops reach the
    test's own connection and transaction only when the coroutine is driven by
    ``async_to_sync``, as Django's ``TestCase`` does for ``async def`` test
    methods (see the testing guide, "Async Tests").

    Args:
        using: Database alias. Default: ``"default"``.

    Raises:
        RLSConfigurationError: If ``USE_LOCAL_SET=True``, ``using`` is not in
            ``RLS_TENANTS["DATABASES"]`` and no transaction is open on the hop
            thread.

    Usage::

        async with async_rls_bypass():
            total = await Order.objects.acount()  # sees all tenants
    """
    async with async_admin_context(using=using):
        yield

django_rls_tenants.tenants.testing.async_rls_as_tenant `async` ¶

async_rls_as_tenant(
    tenant_id: int | str, *, using: str = "default"
) -> AsyncIterator[None]

Async counterpart of :func:rls_as_tenant, for async with in async tests.

Wraps :func:~django_rls_tenants.tenants.context.async_tenant_context, with the same hop model as :func:async_rls_bypass.

Parameters:

Name	Type	Description	Default
`tenant_id`	`int \| str`	The tenant PK to scope to.	required
`using`	`str`	Database alias. Default: `"default"`.	`'default'`

Raises:

Type	Description
`NoTenantContextError`	If `tenant_id` is `None`.
`RLSConfigurationError`	If `USE_LOCAL_SET=True`, `using` is not in `RLS_TENANTS["DATABASES"]` and no transaction is open on the hop thread.

Usage::

async with async_rls_as_tenant(tenant_id=42):
    orders = [order async for order in Order.objects.all()]  # only tenant 42

Source code in django_rls_tenants/tenants/testing.py

@asynccontextmanager
async def async_rls_as_tenant(
    tenant_id: int | str,
    *,
    using: str = "default",
) -> AsyncIterator[None]:
    """Async counterpart of :func:`rls_as_tenant`, for ``async with`` in async tests.

    Wraps :func:`~django_rls_tenants.tenants.context.async_tenant_context`,
    with the same hop model as :func:`async_rls_bypass`.

    Args:
        tenant_id: The tenant PK to scope to.
        using: Database alias. Default: ``"default"``.

    Raises:
        NoTenantContextError: If ``tenant_id`` is ``None``.
        RLSConfigurationError: If ``USE_LOCAL_SET=True``, ``using`` is not in
            ``RLS_TENANTS["DATABASES"]`` and no transaction is open on the hop
            thread.

    Usage::

        async with async_rls_as_tenant(tenant_id=42):
            orders = [order async for order in Order.objects.all()]  # only tenant 42
    """
    async with async_tenant_context(tenant_id, using=using):
        yield

django_rls_tenants.tenants.testing.aassert_rls_enabled `async` ¶

aassert_rls_enabled(
    table_name: str, *, using: str = "default"
) -> None

Async counterpart of :func:assert_rls_enabled.

Runs the sync assertion in one thread_sensitive sync_to_async hop.

Parameters:

Name	Type	Description	Default
`table_name`	`str`	The database table name to check.	required
`using`	`str`	Database alias. Default: `"default"`.	`'default'`

Raises:

Type	Description
`AssertionError`	If the table does not exist, or RLS is not enabled or not forced on it.

Source code in django_rls_tenants/tenants/testing.py

async def aassert_rls_enabled(table_name: str, *, using: str = "default") -> None:
    """Async counterpart of :func:`assert_rls_enabled`.

    Runs the sync assertion in one ``thread_sensitive`` ``sync_to_async`` hop.

    Args:
        table_name: The database table name to check.
        using: Database alias. Default: ``"default"``.

    Raises:
        AssertionError: If the table does not exist, or RLS is not enabled or
            not forced on it.
    """
    await sync_to_async(assert_rls_enabled, thread_sensitive=True)(table_name, using=using)

django_rls_tenants.tenants.testing.aassert_rls_policy_exists `async` ¶

aassert_rls_policy_exists(
    table_name: str,
    policy_name: str | None = None,
    *,
    using: str = "default",
) -> None

Async counterpart of :func:assert_rls_policy_exists.

Runs the sync assertion in one thread_sensitive sync_to_async hop.

Parameters:

Name	Type	Description	Default
`table_name`	`str`	The database table name to check.	required
`policy_name`	`str \| None`	Expected policy name. Defaults to `"{table_name}_tenant_isolation_policy"`.	`None`
`using`	`str`	Database alias. Default: `"default"`.	`'default'`

Raises:

Type	Description
`AssertionError`	If the policy does not exist on the table.

Source code in django_rls_tenants/tenants/testing.py

async def aassert_rls_policy_exists(
    table_name: str,
    policy_name: str | None = None,
    *,
    using: str = "default",
) -> None:
    """Async counterpart of :func:`assert_rls_policy_exists`.

    Runs the sync assertion in one ``thread_sensitive`` ``sync_to_async`` hop.

    Args:
        table_name: The database table name to check.
        policy_name: Expected policy name. Defaults to
            ``"{table_name}_tenant_isolation_policy"``.
        using: Database alias. Default: ``"default"``.

    Raises:
        AssertionError: If the policy does not exist on the table.
    """
    await sync_to_async(assert_rls_policy_exists, thread_sensitive=True)(
        table_name,
        policy_name,
        using=using,
    )

django_rls_tenants.tenants.testing.aassert_rls_blocks_without_context `async` ¶

aassert_rls_blocks_without_context(
    model_class: type[Model], *, using: str = "default"
) -> None

Async counterpart of :func:assert_rls_blocks_without_context.

Runs the sync assertion in one thread_sensitive sync_to_async hop: the admin pre-check, the clearing of the admin and tenant GUCs, the no-context count and the restore all happen inside it, on the connection that holds the caller's GUCs. Like the sync helper it works with STRICT_MODE=True and inside :func:async_rls_bypass or :func:async_rls_as_tenant, and leaves their context intact. Requires at least one row in the table.

Parameters:

Name	Type	Description	Default
`model_class`	`type[Model]`	The RLS-protected model class to query.	required
`using`	`str`	Database alias. Default: `"default"`.	`'default'`

Raises:

Type	Description
`AssertionError`	If any rows are returned, or if the table is empty (which would make the assertion pass vacuously).
`RuntimeError`	If `USE_LOCAL_SET=True`, `using` is not in `RLS_TENANTS["DATABASES"]` and the hop thread's connection is not inside `transaction.atomic()`.

Source code in django_rls_tenants/tenants/testing.py

async def aassert_rls_blocks_without_context(
    model_class: type[models.Model],
    *,
    using: str = "default",
) -> None:
    """Async counterpart of :func:`assert_rls_blocks_without_context`.

    Runs the sync assertion in one ``thread_sensitive`` ``sync_to_async``
    hop: the admin pre-check, the clearing of the admin and tenant GUCs, the
    no-context count and the restore all happen inside it, on the connection
    that holds the caller's GUCs. Like the sync helper it works with
    ``STRICT_MODE=True`` and inside :func:`async_rls_bypass` or
    :func:`async_rls_as_tenant`, and leaves their context intact. Requires at
    least one row in the table.

    Args:
        model_class: The RLS-protected model class to query.
        using: Database alias. Default: ``"default"``.

    Raises:
        AssertionError: If any rows are returned, or if the table is empty
            (which would make the assertion pass vacuously).
        RuntimeError: If ``USE_LOCAL_SET=True``, ``using`` is not in
            ``RLS_TENANTS["DATABASES"]`` and the hop thread's connection is
            not inside ``transaction.atomic()``.
    """
    await sync_to_async(assert_rls_blocks_without_context, thread_sensitive=True)(
        model_class,
        using=using,
    )

Connection Pool¶

Hook for Django's psycopg connection pool (OPTIONS["pool"], Django 5.1+). It is not re-exported from the top-level package; import it from django_rls_tenants.pool, which is safe to import from settings.py. See Django's Connection Pool.

django_rls_tenants.pool.reset_rls_gucs ¶

reset_rls_gucs(conn: _PoolConnection) -> None

Clear the RLS GUCs on a connection returned to Django's psycopg pool.

Pass it as the pool's reset callback; Django forwards every key of OPTIONS["pool"] to psycopg_pool.ConnectionPool::

from django_rls_tenants.pool import reset_rls_gucs

DATABASES = {
    "default": {
        "ENGINE": "django.db.backends.postgresql",
        # ...
        "OPTIONS": {"pool": {"reset": reset_rls_gucs}},
    },
}

The pool calls it on the raw psycopg connection, in one of its worker threads, after rolling back any transaction left open. It clears every GUC the library's RLS policies read (the configured tenant and admin GUCs, and each RLSConstraint's GUCs and extra_bypass_flags) in one statement and leaves the connection idle, as the pool requires. If it raises, the pool discards the connection instead of reusing it.

Unlike DISCARD ALL or RESET ALL, it keeps the session settings Django applies only once per pooled connection, such as the time zone.

Parameters:

Name	Type	Description	Default
`conn`	`_PoolConnection`	The psycopg 3 connection being returned to the pool.	required

Source code in django_rls_tenants/pool.py

def reset_rls_gucs(conn: _PoolConnection) -> None:
    """Clear the RLS GUCs on a connection returned to Django's psycopg pool.

    Pass it as the pool's ``reset`` callback; Django forwards every key of
    ``OPTIONS["pool"]`` to ``psycopg_pool.ConnectionPool``::

        from django_rls_tenants.pool import reset_rls_gucs

        DATABASES = {
            "default": {
                "ENGINE": "django.db.backends.postgresql",
                # ...
                "OPTIONS": {"pool": {"reset": reset_rls_gucs}},
            },
        }

    The pool calls it on the raw psycopg connection, in one of its worker
    threads, after rolling back any transaction left open. It clears every GUC
    the library's RLS policies read (the configured tenant and admin GUCs, and
    each ``RLSConstraint``'s GUCs and ``extra_bypass_flags``) in one statement
    and leaves the connection idle, as the pool requires. If it raises, the pool
    discards the connection instead of reusing it.

    Unlike ``DISCARD ALL`` or ``RESET ALL``, it keeps the session settings
    Django applies only once per pooled connection, such as the time zone.

    Args:
        conn: The psycopg 3 connection being returned to the pool.
    """
    names = _policy_guc_names()
    if not names:
        return
    conn.execute(_build_clear_gucs_sql(len(names)), names)
    if not conn.autocommit:
        # Outside autocommit the statement opened a transaction; the pool
        # discards a connection that is not idle when this returns.
        conn.commit()

Contrib¶

Optional integrations with third-party libraries. These are not re-exported from the top-level package; import them from their module.

Celery¶

Native Celery integration. Requires the celery extra (pip install django-rls-tenants[celery]); import from django_rls_tenants.contrib.celery. See the Celery Tasks guide for usage, chains/groups, and the install() escape hatch.

django_rls_tenants.contrib.celery.rls_task ¶

rls_task(*args: Any, **options: Any) -> Any

Define a Celery task that propagates the RLS context. Like shared_task.

A thin wrapper over :func:celery.shared_task that defaults base to :class:RLSTask. Use it exactly as you would shared_task -- bare or with options::

from django_rls_tenants.contrib.celery import rls_task

@rls_task
def reindex(): ...

@rls_task(bind=True, max_retries=3)
def sync(self): ...

Enqueue inside an RLS context and the worker runs the body scoped to the same tenant::

with tenant_context(tenant.pk):
    reindex.delay()          # runs on the worker under tenant_context(tenant.pk)

Parameters:

Name	Type	Description	Default
`*args`	`Any`	Forwarded to `shared_task` (e.g. the decorated function).	`()`
`**options`	`Any`	Forwarded to `shared_task`. `base` defaults to :class:`RLSTask`; pass your own `base` (a `RLSTask` subclass) to override -- for example to set `rls_require_context = True`.	`{}`

Returns:

Type	Description
`Any`	The shared task, or a decorator when called with options.

Source code in django_rls_tenants/contrib/celery.py

def rls_task(*args: Any, **options: Any) -> Any:
    """Define a Celery task that propagates the RLS context. Like ``shared_task``.

    A thin wrapper over :func:`celery.shared_task` that defaults ``base`` to
    :class:`RLSTask`. Use it exactly as you would ``shared_task`` -- bare or with
    options::

        from django_rls_tenants.contrib.celery import rls_task

        @rls_task
        def reindex(): ...

        @rls_task(bind=True, max_retries=3)
        def sync(self): ...

    Enqueue inside an RLS context and the worker runs the body scoped to the same
    tenant::

        with tenant_context(tenant.pk):
            reindex.delay()          # runs on the worker under tenant_context(tenant.pk)

    Args:
        *args: Forwarded to ``shared_task`` (e.g. the decorated function).
        **options: Forwarded to ``shared_task``. ``base`` defaults to
            :class:`RLSTask`; pass your own ``base`` (a ``RLSTask`` subclass) to
            override -- for example to set ``rls_require_context = True``.

    Returns:
        The shared task, or a decorator when called with options.
    """
    options.setdefault("base", RLSTask)
    return shared_task(*args, **options)

django_rls_tenants.contrib.celery.RLSTask ¶

Bases: Task

Celery task base that carries the RLS context from caller to worker.

On enqueue (apply_async on a real broker, or apply in eager mode and for canvas steps) the active tenant/admin context is captured into the task headers. On the worker, :meth:__call__ reads those headers and runs the body inside the matching tenant_context() / admin_context(), which restores cleanly on both success and exception.

Prefer the :func:rls_task decorator; subclass or pass shared_task(base=RLSTask) only when you need a custom base.

Attributes:

Name	Type	Description
`rls_require_context`	`bool`	When `True`, a task that arrives without any propagated context raises :class:`NoTenantContextError` instead of running unscoped. Defaults to `False` (fail-closed: run with no context, so RLS returns zero rows). Set it per task for jobs that must never run tenant-blind.

apply_async ¶

apply_async(
    args: Any = None, kwargs: Any = None, **options: Any
) -> Any

Capture the active context into headers, then enqueue normally.

Source code in django_rls_tenants/contrib/celery.py

def apply_async(self, args: Any = None, kwargs: Any = None, **options: Any) -> Any:
    """Capture the active context into headers, then enqueue normally."""
    merged_headers = _merge_headers(options)  # also pops "headers" from options
    return super().apply_async(args, kwargs, headers=merged_headers, **options)

apply ¶

apply(
    args: Any = None, kwargs: Any = None, **options: Any
) -> Any

Capture context for eager execution and for canvas steps.

Eager mode and canvas (chain/group) dispatch call apply directly rather than going through apply_async, so the capture is wired in here as well to cover those paths.

Source code in django_rls_tenants/contrib/celery.py

def apply(self, args: Any = None, kwargs: Any = None, **options: Any) -> Any:
    """Capture context for eager execution and for canvas steps.

    Eager mode and canvas (chain/group) dispatch call ``apply`` directly
    rather than going through ``apply_async``, so the capture is wired in
    here as well to cover those paths.
    """
    merged_headers = _merge_headers(options)  # also pops "headers" from options
    return super().apply(args, kwargs, headers=merged_headers, **options)

call ¶

__call__(*args: Any, **kwargs: Any) -> Any

Restore the propagated RLS context around the task body.

Source code in django_rls_tenants/contrib/celery.py

def __call__(self, *args: Any, **kwargs: Any) -> Any:
    """Restore the propagated RLS context around the task body."""
    tenant_id, is_admin = _request_context(self.request)
    if tenant_id is not None:
        with tenant_context(tenant_id):
            return super().__call__(*args, **kwargs)
    if is_admin:
        with admin_context():
            return super().__call__(*args, **kwargs)
    if self.rls_require_context:
        msg = f"Task {self.name!r} ran without an RLS context."
        raise NoTenantContextError(msg, hint=HINT_NO_CONTEXT)
    return super().__call__(*args, **kwargs)

django_rls_tenants.contrib.celery.install ¶

install() -> None

Globally propagate RLS context for all Celery tasks, via signals.

Connects before_task_publish (capture into headers) and task_prerun / task_postrun (restore around the body) so context flows even for tasks that are not based on :class:RLSTask. Use it as an escape hatch for third-party or legacy tasks you cannot re-base.

Prefer :func:rls_task / :class:RLSTask where you can: they restore the context for the whole body including its own apply_async calls, are scoped per task, and need no global wiring. install() and the base class compose safely -- RLSTask instances are skipped by the signal handlers.

Call it once during startup (for example in your Celery app module). It is idempotent: a repeated call does not double-connect. Reverse it with :func:uninstall.

Source code in django_rls_tenants/contrib/celery.py

def install() -> None:
    """Globally propagate RLS context for *all* Celery tasks, via signals.

    Connects ``before_task_publish`` (capture into headers) and
    ``task_prerun`` / ``task_postrun`` (restore around the body) so context flows
    even for tasks that are not based on :class:`RLSTask`. Use it as an escape
    hatch for third-party or legacy tasks you cannot re-base.

    Prefer :func:`rls_task` / :class:`RLSTask` where you can: they restore the
    context for the whole body including its own ``apply_async`` calls, are
    scoped per task, and need no global wiring. ``install()`` and the base class
    compose safely -- ``RLSTask`` instances are skipped by the signal handlers.

    Call it once during startup (for example in your Celery app module). It is
    idempotent: a repeated call does not double-connect. Reverse it with
    :func:`uninstall`.
    """
    before_task_publish.connect(_before_task_publish, dispatch_uid=_DISPATCH_UID, weak=False)
    task_prerun.connect(_task_prerun, dispatch_uid=_DISPATCH_UID, weak=False)
    task_postrun.connect(_task_postrun, dispatch_uid=_DISPATCH_UID, weak=False)

django_rls_tenants.contrib.celery.uninstall ¶

uninstall() -> None

Disconnect the signal handlers connected by :func:install.

Idempotent: safe to call when :func:install was never called. Does not affect tasks based on :class:RLSTask, which never relied on the signals. Any contexts still open from :func:_task_prerun are exited here as a best-effort cleanup so a stale context cannot leak into the next task.

Warning

Do not call this while tasks are still executing on other threads. A ContextVar token can only be reset on the thread that created it, and Django database connections are thread-local, so this cleanup only unwinds contexts entered on the calling thread -- an in-flight task on a worker thread keeps its context until it finishes. Call uninstall() at shutdown, or from the worker thread between tasks.

Source code in django_rls_tenants/contrib/celery.py

def uninstall() -> None:
    """Disconnect the signal handlers connected by :func:`install`.

    Idempotent: safe to call when :func:`install` was never called. Does not
    affect tasks based on :class:`RLSTask`, which never relied on the signals.
    Any contexts still open from :func:`_task_prerun` are exited here as a
    best-effort cleanup so a stale context cannot leak into the next task.

    Warning:
        Do not call this while tasks are still executing on *other* threads. A
        ``ContextVar`` token can only be reset on the thread that created it, and
        Django database connections are thread-local, so this cleanup only
        unwinds contexts entered on the calling thread -- an in-flight task on a
        worker thread keeps its context until it finishes. Call ``uninstall()``
        at shutdown, or from the worker thread between tasks.
    """
    before_task_publish.disconnect(_before_task_publish, dispatch_uid=_DISPATCH_UID)
    task_prerun.disconnect(_task_prerun, dispatch_uid=_DISPATCH_UID)
    task_postrun.disconnect(_task_postrun, dispatch_uid=_DISPATCH_UID)
    for stack in list(_signal_contexts.values()):
        for context in reversed(stack):
            _safe_exit(context)
    _signal_contexts.clear()

API Reference¶

Top-Level Exports¶

Exceptions¶

django_rls_tenants.exceptions.RLSTenantError ¶

django_rls_tenants.exceptions.NoTenantContextError ¶

django_rls_tenants.exceptions.RLSConfigurationError ¶

RLS Layer¶

GUC Helpers¶

django_rls_tenants.rls.guc.set_guc ¶

django_rls_tenants.rls.guc.get_guc ¶

django_rls_tenants.rls.guc.clear_guc ¶

django_rls_tenants.rls.guc.restore_guc ¶

Async GUC Helpers¶

django_rls_tenants.rls.guc.aset_guc async ¶

django_rls_tenants.rls.guc.aget_guc async ¶

django_rls_tenants.rls.guc.aclear_guc async ¶

django_rls_tenants.rls.guc.arestore_guc async ¶

Constraints¶

django_rls_tenants.rls.constraints.RLSConstraint ¶

constraint_sql ¶

create_sql ¶

remove_sql ¶

validate ¶

deconstruct ¶

django_rls_tenants.rls.constraints.RLSM2MConstraint ¶

constraint_sql ¶

create_sql ¶

remove_sql ¶

validate ¶

deconstruct ¶

Migration Operations¶

django_rls_tenants.operations.AddM2MRLSPolicy ¶

state_forwards ¶

database_forwards ¶

database_backwards ¶

describe ¶

deconstruct ¶

Context Managers¶

django_rls_tenants.rls.context.rls_context ¶

django_rls_tenants.rls.context.bypass_flag ¶

django_rls_tenants.rls.context.async_rls_context async ¶

django_rls_tenants.rls.context.async_bypass_flag async ¶

Tenants Layer¶

Configuration¶

django_rls_tenants.tenants.conf.RLSTenantsConfig ¶

TENANT_MODEL property ¶

GUC_PREFIX property ¶

GUC_CURRENT_TENANT property ¶

GUC_IS_ADMIN property ¶

TENANT_FK_FIELD property ¶

USER_PARAM_NAME property ¶

TENANT_PK_TYPE property ¶

USE_LOCAL_SET property ¶

STRICT_MODE property ¶

DATABASES property ¶

reload ¶

Models¶

django_rls_tenants.tenants.models.RLSProtectedModel ¶

Managers¶

django_rls_tenants.tenants.managers.TenantQuerySet ¶

for_user ¶

select_related ¶

count ¶

exists ¶

aggregate ¶

update ¶

delete ¶

iterator ¶

aiterator async ¶

create ¶

bulk_create ¶

bulk_update ¶

update_or_create ¶

explain ¶

get ¶

first ¶

last ¶

django_rls_tenants.tenants.managers.RLSManager ¶

get_queryset ¶

for_user ¶

django_rls_tenants.rls.guc.aset_guc `async` ¶

django_rls_tenants.rls.guc.aget_guc `async` ¶

django_rls_tenants.rls.guc.aclear_guc `async` ¶

django_rls_tenants.rls.guc.arestore_guc `async` ¶

django_rls_tenants.rls.context.async_rls_context `async` ¶

django_rls_tenants.rls.context.async_bypass_flag `async` ¶

TENANT_MODEL `property` ¶

GUC_PREFIX `property` ¶

GUC_CURRENT_TENANT `property` ¶

GUC_IS_ADMIN `property` ¶

TENANT_FK_FIELD `property` ¶

USER_PARAM_NAME `property` ¶

TENANT_PK_TYPE `property` ¶

USE_LOCAL_SET `property` ¶

STRICT_MODE `property` ¶

DATABASES `property` ¶

aiterator `async` ¶

django_rls_tenants.tenants.context.async_tenant_context `async` ¶

django_rls_tenants.tenants.context.async_admin_context `async` ¶

call ¶

acall `async` ¶

is_tenant_admin `property` ¶

rls_tenant_id `property` ¶

django_rls_tenants.tenants.testing.async_rls_bypass `async` ¶

django_rls_tenants.tenants.testing.async_rls_as_tenant `async` ¶

django_rls_tenants.tenants.testing.aassert_rls_enabled `async` ¶

django_rls_tenants.tenants.testing.aassert_rls_policy_exists `async` ¶

django_rls_tenants.tenants.testing.aassert_rls_blocks_without_context `async` ¶

call ¶