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,
    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)

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:

1. Enables RLS on the table (ALTER TABLE ... ENABLE ROW LEVEL SECURITY).
2. Forces RLS for the table owner (ALTER TABLE ... FORCE ROW LEVEL SECURITY).
3. 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 IN (SELECT ...) subqueries.

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).

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).

    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] = {}
    if not is_local:
        for name in variables:
            previous_values[name] = get_guc(name, using=using)

    for name, value in variables.items():
        set_guc(name, value, is_local=is_local, using=using)
    try:
        yield
    finally:
        if not is_local:  # SET LOCAL auto-clears; session-level needs manual restore
            for name, prev in previous_values.items():
                if prev is not None:
                    set_guc(name, prev, using=using)
                else:
                    clear_guc(name, using=using)

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 previous value on exit (supports nesting).

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 previous value on exit (supports nesting).

    Usage::

        with bypass_flag("rls.is_login_request"):
            user = User.objects.get(email=email)
    """
    previous: str | None = None
    if not is_local:
        previous = get_guc(flag_name, using=using)

    set_guc(flag_name, "true", is_local=is_local, using=using)
    try:
        yield
    finally:
        if not is_local:
            if previous is not None:
                set_guc(flag_name, previous, using=using)
            else:
                clear_guc(flag_name, using=using)

---

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

Use SET LOCAL instead of set_config. Default: False.

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 to set GUCs on. Default: ["default"].

In multi-database setups (e.g., read replicas), add all aliases that serve RLS-protected queries::

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

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 sets RLS GUC variables at evaluation time.

Stores the user reference from for_user() and defers GUC setup to _fetch_all(), ensuring lazy querysets work correctly with RLS.

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.

.. warning:: Limitation of for_user() GUC management

GUC variables are only set during ``_fetch_all()`` (iteration).
QuerySet methods that bypass ``_fetch_all()`` — such as
``count()``, ``exists()``, ``aggregate()``, ``update()``,
``delete()``, and ``iterator()`` — will **not** have GUC
variables set by ``for_user()``.

For non-admin users this is safe because ``for_user()`` also
adds a Django ORM ``WHERE tenant_id = X`` filter. For **admin
users** (``is_tenant_admin=True``), no ORM filter is applied,
so these methods run against whatever GUC state the connection
already has.

For non-middleware contexts (Celery tasks, management commands),
use ``tenant_context()`` or ``admin_context()`` instead, which
set GUCs at the connection level for the entire block.

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. GUC variables are set when the queryset is evaluated, not when for_user() is called.

.. warning::

GUC variables are only set during iteration (``_fetch_all``).
Methods like ``count()``, ``exists()``, ``aggregate()``,
``update()``, ``delete()``, and ``iterator()`` bypass
``_fetch_all`` and will **not** have GUC variables set.
For admin users this means those methods run without GUC
protection. Use ``tenant_context()`` or ``admin_context()``
for full coverage in non-middleware contexts.

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. GUC variables are set
    when the queryset is evaluated, not when ``for_user()`` is called.

    .. warning::

        GUC variables are only set during iteration (``_fetch_all``).
        Methods like ``count()``, ``exists()``, ``aggregate()``,
        ``update()``, ``delete()``, and ``iterator()`` bypass
        ``_fetch_all`` and will **not** have GUC variables set.
        For admin users this means those methods run without GUC
        protection. Use ``tenant_context()`` or ``admin_context()``
        for full coverage in non-middleware contexts.

    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 or when called with no arguments (select-all mode).

Handles select_related(False) (Django 5.x) and select_related(None) (Django 6.0+) for clearing without adding tenant filters.

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 or when called with no arguments (select-all mode).

    Handles ``select_related(False)`` (Django 5.x) and
    ``select_related(None)`` (Django 6.0+) for clearing without
    adding tenant filters.
    """
    # Django 5.x uses False, Django 6.0+ uses None to clear
    # select_related. Normalize to None before calling super() to
    # avoid AttributeError: 'bool' object has no attribute 'split'.
    if fields in ((False,), (None,)):
        return super().select_related(None)
    if not fields:
        return super().select_related()
    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 = _resolve_related_model(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})
            # For nullable FKs, preserve LEFT OUTER JOIN semantics:
            # include rows where the FK is NULL (no related object).
            # Without this, .filter() forces an INNER JOIN which
            # silently drops rows with NULL FKs.
            try:
                fk_field = self.model._meta.get_field(field_path)  # noqa: SLF001
                if getattr(fk_field, "null", False):
                    tenant_filter = tenant_filter | Q(**{f"{field_path}__isnull": True})
            except FieldDoesNotExist:
                pass
            qs = qs.filter(tenant_filter)
    return qs

count

count() -> int

Guard count() with strict mode check.

Source code in django_rls_tenants/tenants/managers.py

def count(self) -> int:
    """Guard ``count()`` with strict mode check."""
    self._check_strict_mode()
    return super().count()

exists

exists() -> bool

Guard exists() with strict mode check.

Source code in django_rls_tenants/tenants/managers.py

def exists(self) -> bool:
    """Guard ``exists()`` with strict mode check."""
    self._check_strict_mode()
    return super().exists()

aggregate

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

Guard aggregate() with strict mode check.

Source code in django_rls_tenants/tenants/managers.py

def aggregate(self, *args: Any, **kwargs: Any) -> dict[str, Any]:
    """Guard ``aggregate()`` with strict mode check."""
    self._check_strict_mode()
    return super().aggregate(*args, **kwargs)

update

update(**kwargs: Any) -> int

Guard update() with strict mode check.

Source code in django_rls_tenants/tenants/managers.py

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

delete

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

Guard delete() with strict mode check.

Source code in django_rls_tenants/tenants/managers.py

def delete(self) -> tuple[int, dict[str, int]]:
    """Guard ``delete()`` with strict mode check."""
    self._check_strict_mode()
    return super().delete()

iterator

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

Guard iterator() with strict mode check.

Source code in django_rls_tenants/tenants/managers.py

def iterator(self, chunk_size: int | None = None) -> Any:
    """Guard ``iterator()`` with strict mode check."""
    self._check_strict_mode()
    return super().iterator(chunk_size=chunk_size)

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]

Guard bulk_create() with strict mode check.

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]:
    """Guard ``bulk_create()`` with strict mode check."""
    self._check_strict_mode()
    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

Guard bulk_update() with strict mode check.

Source code in django_rls_tenants/tenants/managers.py

def bulk_update(
    self,
    objs: Any,
    fields: Any,
    batch_size: int | None = None,
) -> int:
    """Guard ``bulk_update()`` with strict mode check."""
    self._check_strict_mode()
    return super().bulk_update(objs, fields, batch_size=batch_size)

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.

Source code in django_rls_tenants/tenants/managers.py

def first(self) -> Any:
    """Guard ``first()`` with strict mode check."""
    self._check_strict_mode()
    return super().first()

last

last() -> Any

Guard last() with strict mode check.

Source code in django_rls_tenants/tenants/managers.py

def last(self) -> Any:
    """Guard ``last()`` with strict mode check."""
    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.
    """
    qs = TenantQuerySet(self.model, using=self._db)
    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.

Parameters:

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

Raises:

Type	Description
NoTenantContextError	If tenant_id is None.

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.

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

    Raises:
        NoTenantContextError: If ``tenant_id`` is ``None``.
    """
    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

    # Save previous state for nesting support
    prev_admin: str | None = None
    prev_tenant: str | None = None
    if not is_local:
        prev_admin = get_guc(conf.GUC_IS_ADMIN, using=using)
        prev_tenant = get_guc(conf.GUC_CURRENT_TENANT, using=using)

    active_token = set_rls_context_active(True)
    token = set_current_tenant_id(tenant_id)
    try:
        set_guc(conf.GUC_IS_ADMIN, "false", is_local=is_local, using=using)
        set_guc(
            conf.GUC_CURRENT_TENANT,
            str(tenant_id),
            is_local=is_local,
            using=using,
        )
        logger.debug("tenant_context entered: tenant_id=%s, using=%s", tenant_id, using)
        yield
    finally:
        reset_current_tenant_id(token)
        reset_rls_context_active(active_token)
        if not is_local:
            _restore_guc(conf.GUC_IS_ADMIN, prev_admin, using=using)
            _restore_guc(conf.GUC_CURRENT_TENANT, prev_tenant, using=using)
        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.

Parameters:

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

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.

    Args:
        using: Database alias. Default: ``"default"``.
    """
    conf = rls_tenants_config
    is_local = conf.USE_LOCAL_SET

    # Save previous state for nesting support
    prev_admin: str | None = None
    prev_tenant: str | None = None
    if not is_local:
        prev_admin = get_guc(conf.GUC_IS_ADMIN, using=using)
        prev_tenant = get_guc(conf.GUC_CURRENT_TENANT, using=using)

    active_token = set_rls_context_active(True)
    token = set_current_tenant_id(None)
    try:
        set_guc(conf.GUC_IS_ADMIN, "true", 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
    finally:
        reset_current_tenant_id(token)
        reset_rls_context_active(active_token)
        if not is_local:
            _restore_guc(conf.GUC_IS_ADMIN, prev_admin, using=using)
            _restore_guc(conf.GUC_CURRENT_TENANT, prev_tenant, using=using)
        logger.debug("admin_context exited: using=%s", using)

django_rls_tenants.tenants.context.with_rls_context

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): ...

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.

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(
    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): ...

    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.

    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]:
        # 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,
                fn.__qualname__,
            )

        @functools.wraps(fn)
        def wrapper(*args: _P.args, **kwargs: _P.kwargs) -> _R:
            as_user: TenantUser | None = _get_arg_from_signature(sig, param_name, *args, **kwargs)

            if as_user is not None and as_user.is_tenant_admin:
                ctx = admin_context()
            elif as_user is not None:
                tenant_id = as_user.rls_tenant_id
                if tenant_id is None:
                    msg = f"Non-admin user passed to {fn.__qualname__} has rls_tenant_id=None."
                    raise NoTenantContextError(msg, hint=HINT_USER_NO_TENANT)
                ctx = tenant_context(tenant_id)
            else:
                logger.warning(
                    "with_rls_context: %s is None in call to %s, no RLS context set (fail-closed)",
                    param_name,
                    fn.__qualname__,
                )
                return fn(*args, **kwargs)

            with ctx:
                return fn(*args, **kwargs)

        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

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",
]

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.

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.
    """
    if not hasattr(request, "user") or not request.user.is_authenticated:
        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
    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
        _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; fall back to set() when the token was
        # never created (error occurred before set call).
        if isinstance(tenant_token, Token):
            reset_current_tenant_id(tenant_token)
        else:
            set_current_tenant_id(None)
        if isinstance(active_token, Token):
            reset_rls_context_active(active_token)
        else:
            set_rls_context_active(False)
        _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

Clear RLS state on unhandled exceptions to prevent ContextVar leaks.

Without this, a view exception that prevents process_response from running would leave the ContextVar set for the remainder of the thread (WSGI) or async task (ASGI).

Source code in django_rls_tenants/tenants/middleware.py

def process_exception(
    self,
    request: HttpRequest,
    exception: Exception,  # noqa: ARG002  -- required by MiddlewareMixin
) -> None:
    """Clear RLS state on unhandled exceptions to prevent ContextVar leaks.

    Without this, a view exception that prevents ``process_response``
    from running would leave the ContextVar set for the remainder of
    the thread (WSGI) or async task (ASGI).
    """
    self._cleanup_rls_state(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, *, using: str = "default"
) -> None

Clear a bypass flag on the current database connection.

Source code in django_rls_tenants/tenants/bypass.py

def clear_bypass_flag(
    flag_name: str,
    *,
    using: str = "default",
) -> None:
    """Clear a bypass flag on the current database connection."""
    clear_guc(flag_name, 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 querying with no GUC context returns zero rows.

Verifies the fail-closed behavior. Requires at least one row to exist in the table (caller must set up test data first).

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).

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 querying with no GUC context returns zero rows.

    Verifies the fail-closed behavior. Requires at least one row
    to exist in the table (caller must set up test data first).

    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).
    """
    # 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 = model_class.objects.using(using).count()  # type: ignore[attr-defined]
    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."
    )

    qs = model_class.objects.using(using).all()  # type: ignore[attr-defined]
    count = qs.count()
    assert count == 0, (
        f"Expected 0 rows from {model_class.__name__} without "
        f"RLS context, got {count}. RLS may not be enforced."
    )

---

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()