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,
    NoTenantContextError,
    RLSConfigurationError,
    RLSConstraint,
    RLSM2MConstraint,
    RLSManager,
    RLSProtectedModel,
    RLSTenantError,
    RLSTenantMiddleware,
    TenantQuerySet,
    TenantUser,
    admin_context,
    get_current_tenant_id,
    get_rls_context_active,
    reset_current_tenant_id,
    reset_rls_context_active,
    set_current_tenant_id,
    set_rls_context_active,
    tenant_context,
    with_rls_context,
)

---

Exceptions

Custom exception hierarchy for precise error handling. All exceptions live in django_rls_tenants.exceptions and are re-exported from the top-level package.

django_rls_tenants.exceptions.RLSTenantError

Bases: Exception

Base exception for all django-rls-tenants errors.

Catch this to handle any error raised by the library.

django_rls_tenants.exceptions.NoTenantContextError

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.

django_rls_tenants.exceptions.RLSConfigurationError

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.

---

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 = (
        f"{self.field}_id = nullif("
        f"current_setting('{self.guc_tenant_var}', true), '')"
        f"::{self.tenant_pk_type}"
    )
    admin_check = f"current_setting('{self.guc_admin_var}', true) = 'true'"

    # 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(
        f"current_setting('{flag}', true) = 'true'" 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 = f"current_setting('{self.guc_admin_var}', true) = 'true'"
    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 = f"current_setting('{self.guc_admin_var}', true) = 'true'"
    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],
    as_user: TenantUser,  # noqa: ARG002  -- part of public API
) -> 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).
    """
    from django.apps import apps  # noqa: PLC0415  -- lazy import avoids circular

    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. For admin access, use admin_context() instead."
        raise NoTenantContextError(msg)

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

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

django_rls_tenants.tenants.context.with_rls_context

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

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[..., Any] | 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

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[..., Any] | None = None,
    *,
    user_param: str | None = None,
) -> Callable[..., Any]:
    """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"``).

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

    def decorator(fn: Callable[..., Any]) -> Callable[..., Any]:
        # 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: Any, **kwargs: Any) -> Any:
            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 "
                        f"rls_tenant_id=None. Assign the user to a tenant "
                        f"or set is_tenant_admin=True."
                    )
                    raise NoTenantContextError(msg)
                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

State

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)

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

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