Skip to content

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 (
    RLSConstraint,
    RLSManager,
    RLSProtectedModel,
    TenantQuerySet,
    TenantUser,
    admin_context,
    tenant_context,
    with_rls_context,
)

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 = coalesce("
        f"nullif(current_setting('{self.guc_tenant_var}', true), '')"
        f"::{self.tenant_pk_type}, NULL)"
    )
    admin_bypass = f"coalesce(current_setting('{self.guc_admin_var}', true) = 'true', false)"

    # Extra bypass flags apply to USING (SELECT) only, NOT WITH CHECK (INSERT/UPDATE).
    extra_using_clauses = ""
    for flag in self.extra_bypass_flags:
        extra_using_clauses += (
            "\n                                OR coalesce("
            f"current_setting('{flag}', true) = 'true', false)"
        )

    # 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 (
                            {tenant_match}
                            OR {admin_bypass}{extra_using_clauses}
                        )
                        WITH CHECK (
                            {tenant_match}
                            OR {admin_bypass}
                        );
                    $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)

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
    "GUC_PREFIX": "rls",             # Default: "rls"
    "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.

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.

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.

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.

    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

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.

Source code in django_rls_tenants/tenants/managers.py
def get_queryset(self) -> TenantQuerySet:
    """Return a ``TenantQuerySet`` instance."""
    return TenantQuerySet(self.model, using=self._db)

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
ValueError

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:
        ValueError: If ``tenant_id`` is ``None``.
    """
    if tenant_id is None:
        msg = "tenant_id cannot be None. For admin access, use admin_context() instead."
        raise ValueError(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)

    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,
    )
    try:
        yield
    finally:
        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)

    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)
    try:
        yield
    finally:
        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
                # tenant_context validates None at runtime; narrow type for mypy
                ctx = tenant_context(tenant_id)  # type: ignore[arg-type]
            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

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 based on the authenticated user.

If the first set_guc succeeds but the second fails (e.g., broken connection), both GUCs are cleared to prevent partial state from leaking tenant context.

Source code in django_rls_tenants/tenants/middleware.py
def process_request(self, request: HttpRequest) -> None:
    """Set GUC variables based on the authenticated user.

    If the first ``set_guc`` succeeds but the second fails (e.g.,
    broken connection), both GUCs are cleared to prevent partial
    state from leaking tenant context.
    """
    if not hasattr(request, "user") or not request.user.is_authenticated:
        return

    conf = rls_tenants_config
    user: Any = request.user

    try:
        guc_vars = _resolve_user_guc_vars(user, conf)
        for guc_name, guc_value in guc_vars.items():
            if guc_value:
                set_guc(guc_name, guc_value, is_local=conf.USE_LOCAL_SET)
            else:
                clear_guc(guc_name, is_local=conf.USE_LOCAL_SET)
        _mark_gucs_set()
    except Exception:
        logger.exception("Failed to set RLS GUC variables, clearing both to prevent leak")
        try:
            clear_guc(conf.GUC_IS_ADMIN)
            clear_guc(conf.GUC_CURRENT_TENANT)
        except Exception:  # noqa: S110  -- best-effort cleanup, connection may be dead
            pass
        raise

process_response

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

Clear GUC variables to prevent cross-request leaks.

Source code in django_rls_tenants/tenants/middleware.py
def process_response(
    self,
    request: HttpRequest,  # noqa: ARG002  -- required by MiddlewareMixin
    response: HttpResponse,
) -> HttpResponse:
    """Clear GUC variables to prevent cross-request leaks."""
    conf = rls_tenants_config
    if not conf.USE_LOCAL_SET:
        clear_guc(conf.GUC_IS_ADMIN)
        clear_guc(conf.GUC_CURRENT_TENANT)
    _clear_gucs_set_flag()
    return response

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