How to enforce multi-tenancy & soft deletion

Automatically isolate tenant data and filter soft-deleted records across all repository queries using SoftDeleteMixin and scope_query.

Writing manual WHERE tenant_id == ... or WHERE deleted_at IS NULL clauses on every single database query is error-prone and risks catastrophic data leaks.

ZCore solves this at the architectural root:

SoftDeleteMixin: Provides out-of-the-box timestamp-based soft deletion with automatic query filtering, soft_delete(), restore(), and atomic batch recovery helpers.
scope_query: BaseRepository automatically inspects models for a @classmethod named scope_query before executing any read, listing, count, or dynamic search query.

1. Built-in Soft Deletion (`SoftDeleteMixin`)

To add soft-delete capabilities to any model, simply inherit from SoftDeleteMixin:

# tasks/models.py
import uuid
from sqlalchemy.orm import Mapped, mapped_column
from sqlalchemy import String
from zcore import Base, SoftDeleteMixin

class Task(Base, SoftDeleteMixin):
    __tablename__ = "tasks"

    id: Mapped[uuid.UUID] = mapped_column(primary_key=True, default=uuid.uuid4)
    title: Mapped[str] = mapped_column(String(255))

SoftDeleteMixin automatically injects:

deleted_at: Mapped[datetime | None] (indexed timezone-aware timestamp, nullable).
task.is_deleted property (evaluates whether deleted_at is set).
task.soft_delete() (sets deleted_at = now()).
task.restore() (resets deleted_at = None).
Automatic scope_query filtering WHERE deleted_at IS NULL for all repository queries.

2. Combining Soft Deletion with Multi-Tenancy

If your model requires both soft-delete filtering and tenant isolation, override scope_query and chain it with super().scope_query(query):

# tasks/models.py
import uuid
from sqlalchemy import String, Select
from sqlalchemy.orm import Mapped, mapped_column
from zcore import Base, SoftDeleteMixin, ctx

class Task(Base, SoftDeleteMixin):
    __tablename__ = "tasks"

    id: Mapped[uuid.UUID] = mapped_column(primary_key=True, default=uuid.uuid4)
    title: Mapped[str] = mapped_column(String(255))
    tenant_id: Mapped[str] = mapped_column(String(64), index=True)

    @classmethod
    def scope_query(cls, query: Select) -> Select:
        # 1. Apply base soft-delete filter (deleted_at IS NULL)
        query = super().scope_query(query)

        # 2. Enforce Multi-Tenant isolation from active request context
        active_tenant = ctx.get("tenant_id")
        if active_tenant:
            query = query.where(cls.tenant_id == active_tenant)

        return query

3. Setting Context in Auth / Middleware

Set the tenant identifier in ZContext (ctx) during authentication or inside your authentication dependency:

# Inside your BaseAuth or dependency provider
from zcore import ctx

# Binds the active organization/tenant to the current coroutine context
ctx.set("tenant_id", "org_enterprise_99")

4. Automatic Repository Protection

Once defined, every operation executed by BaseRepository and BaseRouter respects your scoping rules automatically with zero extra code:

# Standard repository calls:
task = await task_repo.get(id=task_id)          # Automatically filtered: non-deleted & matching tenant
all_tasks = await task_repo.get_list()          # Returns only active tenant records
count = await task_repo.count()                 # Counts only active tenant records
search = await task_repo.search(search_request) # Dynamic search applies tenant & soft-delete filter

5. Soft-Deleting, Restoring, and Forced Deletions

In addition to instance-level methods, BaseRepository and BaseService provide high-level, atomic operations:

# 1. Soft-delete a record (or list of records) via repository
await task_repo.delete(task_id)                        # Automatic soft-delete
await task_repo.delete_multi([task_id_1, task_id_2])   # Batch atomic soft-delete

# 2. Force permanent physical deletion (Hard Delete)
await task_repo.delete(task_id, force=True)
await task_repo.delete_multi([task_id_1, task_id_2], force=True)

# 3. Restore soft-deleted records
restored_task = await task_repo.restore(task_id)       # Single restoration
restored_list = await task_repo.restore_multi(         # Atomic batch restoration
    [task_id_1, task_id_2]
)

# 4. Instance-level mutation fallback
task = await task_repo.get(id=task_id)
task.soft_delete()
await db.flush()

Comprehensive Operation Coverage: scope_query is executed inside _get_base_query() and automatically wraps all repository operations: get(), get_by_ids(), get_list(), exist(), count(), and search().

Bypassing Scoping for Superadmins & Audits: If a background worker or superadmin needs to query across all tenants or include soft-deleted records:

@classmethod
def scope_query(cls, query: Select) -> Select:
    if ctx.get("is_superuser"):
        return query # Unfiltered access for audits/superadmins
        
    query = super().scope_query(query)
    tenant = ctx.get("tenant_id")
    return query.where(cls.tenant_id == tenant) if tenant else query