Metadata-Version: 2.4
Name: softtenant
Version: 0.1.0
Summary: Retrofit multi-tenancy into existing FastAPI + SQLAlchemy applications
Project-URL: Repository, https://github.com/suhanapthn24/softtenant
Project-URL: Issues, https://github.com/suhanapthn24/softtenant/issues
Project-URL: Changelog, https://github.com/suhanapthn24/softtenant/blob/main/CHANGELOG.md
Author-email: Soft Tech Talks <pthnsuhana12@gmail.com>
License: MIT License
        
        Copyright (c) 2026 Soft Tech Talks
        
        Permission is hereby granted, free of charge, to any person obtaining a copy
        of this software and associated documentation files (the "Software"), to deal
        in the Software without restriction, including without limitation the rights
        to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
        copies of the Software, and to permit persons to whom the Software is
        furnished to do so, subject to the following conditions:
        
        The above copyright notice and this permission notice shall be included in all
        copies or substantial portions of the Software.
        
        THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
        IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
        FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
        AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
        LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
        OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
        SOFTWARE.
License-File: LICENSE
Keywords: fastapi,multi-tenancy,saas,sqlalchemy,tenancy
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Database
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Python: >=3.11
Requires-Dist: alembic>=1.13
Requires-Dist: sqlalchemy>=2.0
Provides-Extra: dev
Requires-Dist: aiosqlite>=0.20; extra == 'dev'
Requires-Dist: fastapi>=0.115; extra == 'dev'
Requires-Dist: httpx>=0.27; extra == 'dev'
Requires-Dist: mypy>=1.10; extra == 'dev'
Requires-Dist: pytest-asyncio>=0.23; extra == 'dev'
Requires-Dist: pytest>=8.0; extra == 'dev'
Requires-Dist: ruff>=0.4; extra == 'dev'
Requires-Dist: starlette>=0.40; extra == 'dev'
Requires-Dist: testcontainers[mysql,postgres]>=4.0; extra == 'dev'
Provides-Extra: fastapi
Requires-Dist: fastapi>=0.115; extra == 'fastapi'
Requires-Dist: starlette>=0.40; extra == 'fastapi'
Description-Content-Type: text/markdown

# softtenant

**Retrofit multi-tenancy into existing FastAPI + SQLAlchemy applications — without rewriting your queries.**

[![PyPI](https://img.shields.io/pypi/v/softtenant)](https://pypi.org/project/softtenant/)
[![Python](https://img.shields.io/pypi/pyversions/softtenant)](https://pypi.org/project/softtenant/)
[![License: MIT](https://img.shields.io/badge/license-MIT-blue.svg)](LICENSE)
[![Tests](https://github.com/suhanapthn24/softtenant/actions/workflows/ci.yml/badge.svg)](https://github.com/suhanapthn24/softtenant/actions)

---

## The problem

Adding multi-tenancy to an existing single-tenant app is tedious and risky.
The naive approach — add `company_id` to every table, then manually filter every query — takes weeks and leaks data the moment a developer forgets a `WHERE` clause:

```python
# Before: developer must remember to filter EVERY query
orders = session.execute(
    select(Order).where(Order.company_id == current_tenant_id)  # forget this once → data leak
).scalars().all()
```

With dozens of models and hundreds of query sites, missing even one is guaranteed.

## The solution

softtenant hooks into SQLAlchemy's event system and **injects the `WHERE company_id = <tid>` clause automatically** on every SELECT, UPDATE, and DELETE against a marked model. Developers write normal queries; the library enforces isolation.

```python
# After: no filter needed — softtenant adds it automatically
orders = session.execute(select(Order)).scalars().all()

# If no tenant context is active, the query raises instead of returning wrong data.
# There is no way to accidentally leak cross-tenant data.
```

---

## Installation

```bash
pip install softtenant              # core — SQLAlchemy + Alembic only
pip install softtenant[fastapi]     # adds FastAPI dependency + middleware helpers
```

**Requirements:** Python 3.11+, SQLAlchemy 2.x, Alembic 1.13+

---

## Quick start (5 minutes)

### 1. Mark your models

```python
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from softtenant import TenantScopedMixin

class Base(DeclarativeBase):
    pass

class Order(TenantScopedMixin, Base):   # <-- only change from a vanilla model
    __tablename__ = "orders"
    id: Mapped[int] = mapped_column(primary_key=True)
    item: Mapped[str]
    amount: Mapped[float]
    # company_id is added automatically by TenantScopedMixin
```

Inheriting `TenantScopedMixin` adds a nullable `company_id: int` column (the migration makes it `NOT NULL`) and registers the model with the query-scoping hook.

### 2. Install the session hook (once at startup)

```python
from softtenant import install_scoping_hooks
from sqlalchemy.ext.asyncio import AsyncSession

# Sync apps
from sqlalchemy.orm import Session
install_scoping_hooks(Session)

# Async apps
install_scoping_hooks(AsyncSession)
```

One call, any time before the first query. Safe to call multiple times — idempotent.

### 3. Wrap requests in a tenant context

```python
from softtenant import tenant_context

with tenant_context(tenant_id=42):
    orders = session.execute(select(Order)).scalars().all()
    # SQL: SELECT * FROM orders WHERE company_id = 42
```

### 4. Generate the Alembic migration

```bash
softtenant generate-migration \
    --base myapp.models:Base \
    --out migrations/versions/add_multi_tenancy.py
```

Then run it normally:

```bash
alembic upgrade head
```

---

## Core concepts

### `TenantScopedMixin`

```python
from softtenant import TenantScopedMixin

class Invoice(TenantScopedMixin, Base):
    __tablename__ = "invoices"
    id: Mapped[int] = mapped_column(primary_key=True)
    amount: Mapped[float]
```

The mixin adds:
- `company_id: Mapped[int | None]` — nullable in the model definition (migration enforces NOT NULL)
- Auto-registration with the query-scoping system

**Opting a model out** — set `__multi_tenant__ = False` to exclude a model from filtering:

```python
class AuditLog(Base):
    __tablename__ = "audit_logs"
    __multi_tenant__ = False        # visible to all tenants regardless of context
    id: Mapped[int] = mapped_column(primary_key=True)
    message: Mapped[str]
```

### `install_scoping_hooks(session_class)`

Registers two SQLAlchemy event listeners on the given Session class:

| Event | What it does |
|---|---|
| `do_orm_execute` | Injects `WHERE company_id = <tid>` into every SELECT, UPDATE, and DELETE against a scoped model |
| `before_flush` | Stamps `company_id` on new objects that don't have it set yet |

**Idempotent** — registering twice has no effect (safe to call from multiple test fixtures or multiple `startup` handlers).

### Tenant context

The active tenant ID is stored in a `ContextVar`, which is:
- **Async-safe**: each asyncio Task has its own copy — no cross-request leakage
- **Thread-safe**: same story for threaded WSGI apps

```python
from softtenant import tenant_context, set_current_tenant, get_current_tenant

# Option A — context manager (recommended)
with tenant_context(42):
    ...   # company_id = 42 here

# Option B — imperative (for middleware)
token = set_current_tenant(42)
try:
    ...
finally:
    token.var.reset(token)     # restores previous value

# Read the current tenant
tid = get_current_tenant()     # raises TenantContextNotSetError if unset
```

### Auto-stamp on INSERT

When you add a new object inside a `tenant_context`, you never need to set `company_id`:

```python
with tenant_context(42):
    order = Order(item="Widget", amount=99.0)   # no company_id argument
    session.add(order)
    session.flush()
    assert order.company_id == 42               # stamped automatically
```

The `before_flush` hook sets it from the ContextVar before the INSERT fires.

### Safety guarantee

If code tries to query a scoped model **without** an active tenant context, softtenant **raises `TenantContextNotSetError`** rather than silently returning no rows or all rows:

```python
# No tenant_context() active — this raises, not leaks
session.execute(select(Order))
# → TenantContextNotSetError: No tenant context is set. ...
```

This makes the failure mode loud and catches missing context during development rather than in production.

---

## FastAPI integration

### Option A — Dependency injection (recommended)

The dependency approach is opt-in per route, composable with auth, and overridable in tests.

```python
from typing import Annotated
from fastapi import Depends, FastAPI
from starlette.requests import Request
from softtenant.fastapi import make_tenant_dependency

app = FastAPI()

# Define how to resolve a tenant id from a request
async def resolve_tenant(request: Request) -> int:
    raw = request.headers.get("X-Tenant-ID")
    if not raw:
        from fastapi import HTTPException
        raise HTTPException(status_code=401, detail="X-Tenant-ID required")
    return int(raw)

# Create the dependency
require_tenant = make_tenant_dependency(resolver=resolve_tenant)
TenantDep = Annotated[int, Depends(require_tenant)]

@app.get("/orders")
async def list_orders(_: TenantDep, db: DBDep) -> list[OrderRead]:
    result = await db.execute(select(Order))   # WHERE company_id = <tid> injected
    return result.scalars().all()

@app.post("/orders", status_code=201)
async def create_order(body: OrderCreate, _: TenantDep, db: DBDep) -> OrderRead:
    order = Order(item=body.item, amount=body.amount)  # company_id stamped on flush
    db.add(order)
    await db.commit()
    return order
```

**Resolver signature:** `(Request) -> int | Awaitable[int]` — sync or async, both work.

**Testing with overrides:**

```python
app.dependency_overrides[require_tenant] = lambda: 99  # pin to tenant 99 in tests
```

### Option B — Middleware

Use middleware when you want every request (including 404s, `/docs`, health checks) to resolve a tenant automatically.

```python
from softtenant.fastapi import TenantMiddleware

app.add_middleware(TenantMiddleware, resolver=resolve_tenant)
```

The middleware resolves the tenant, calls `set_current_tenant`, and resets the context after the response. If the resolver raises `HTTPException`, it is converted to a JSON error response.

### Admin routes — cross-tenant access

Use `bypass_tenant_scope()` for routes that legitimately need all-tenant data:

```python
from softtenant import bypass_tenant_scope

@app.get("/admin/orders")
async def admin_list_orders(db: DBDep) -> list[OrderRead]:
    with bypass_tenant_scope():
        result = await db.execute(select(Order))   # no WHERE company_id filter
        return result.scalars().all()
```

`bypass_tenant_scope()` is an **explicit escape hatch** — it cannot be entered by accident. It restores the previous scoping state when it exits.

---

## Migration generator

### CLI

```bash
softtenant generate-migration \
    --base myapp.models:Base \
    --out migrations/versions/add_multi_tenancy.py
```

**Options:**

| Flag | Default | Description |
|---|---|---|
| `--base MODULE:ATTR` | *(required)* | `import_path:attribute` of your declarative Base |
| `--out PATH` | *(required)* | Output path for the generated migration file |
| `--companies-table` | `companies` | Name for the tenant lookup table |
| `--tenant-id-column` | `company_id` | Name for the FK column added to scoped tables |
| `--default-tenant-name` | `Default` | Name of the seed tenant inserted during upgrade |
| `--batch-size` | `500` | Rows per UPDATE batch during the backfill step |

### What the generated migration does

1. **Creates** `companies (id, name, created_at)` — the tenant lookup table
2. **Inserts** a default tenant row (`id=1, name="Default"`)
3. For each tenant-scoped table:
   - **Adds** `company_id INTEGER NULL` — fast, no table lock
   - **Backfills** `company_id = 1` in batches using keyset pagination (no OFFSET scan)
   - **Alters** `company_id` to `NOT NULL`
   - **Creates** an index on `company_id`
   - **Adds** a FK constraint to `companies.id`
4. **Downgrade** reverses all of the above

All schema changes use `op.batch_alter_table()` so the migration works on **SQLite**, **PostgreSQL**, and **MySQL** without modification.

### Programmatic API

```python
from softtenant.migration import generate_migration

generate_migration(
    base=Base,
    output_path="migrations/versions/add_multi_tenancy.py",
    companies_table="companies",
    tenant_id_column="company_id",
    default_tenant_id=1,
    default_tenant_name="Acme Corp",
    batch_size=1000,
)
```

---

## Seeding data in tests

Use `bypass_tenant_scope()` when writing seed data that needs explicit `company_id` values (e.g., in fixtures):

```python
from softtenant import bypass_tenant_scope

with bypass_tenant_scope():
    session.add(Order(id=1, item="Widget", amount=100.0, company_id=1))
    session.add(Order(id=2, item="Gadget", amount=200.0, company_id=2))
    session.commit()
```

---

## Nested contexts

`tenant_context()` nests correctly — the outer context is restored when the inner one exits:

```python
with tenant_context(1):
    orders_t1 = session.execute(select(Order)).scalars().all()  # tenant 1
    with tenant_context(2):
        orders_t2 = session.execute(select(Order)).scalars().all()  # tenant 2
    orders_t1_again = session.execute(select(Order)).scalars().all()  # tenant 1 again
```

---

## Configuration

Override library defaults once at application startup:

```python
from softtenant import configure

configure(
    companies_table="tenants",   # default: "companies"
    tenant_id_column="tenant_id",  # default: "company_id"
    batch_size=1000,             # default: 500 (migration backfill batch size)
)
```

`configure()` raises `SoftTenantConfigError` on unknown keys, catching typos at startup.

---

## Error reference

| Exception | When raised |
|---|---|
| `TenantContextNotSetError` | A SELECT/UPDATE/DELETE is issued against a scoped model with no active tenant context |
| `TenantResolutionError` | The tenant resolver returned a non-integer value |
| `SoftTenantConfigError` | `configure()` was called with an unknown configuration key |

All exceptions inherit from `SoftTenantError`.

---

## Examples

The [`examples/`](examples/) directory contains two runnable demos:

**`examples/demo.py`** — pure SQLAlchemy, no server required:
```bash
python examples/demo.py
```
Demonstrates every core feature (SELECT isolation, auto-stamp, bypass, bulk UPDATE/DELETE, nested contexts) against an in-memory SQLite database.

**`examples/fastapi_app.py`** — complete async FastAPI application:
```bash
pip install uvicorn aiosqlite
python examples/fastapi_app.py
# Docs at http://localhost:8000/docs
```
Then:
```bash
curl -H "X-Tenant-ID: 1" http://localhost:8000/orders
curl -H "X-Tenant-ID: 2" http://localhost:8000/orders
curl http://localhost:8000/admin/orders   # cross-tenant, no header required
```

---

## How it works

softtenant uses two SQLAlchemy session events:

**`do_orm_execute`** — fires before every ORM statement executes. For SELECT, softtenant adds a `with_loader_criteria` option that SQLAlchemy propagates through joins, subqueries, and eager loads automatically. For UPDATE/DELETE, it appends a `WHERE company_id = <tid>` clause directly to the statement.

**`before_flush`** — fires before new objects are written. softtenant iterates `session.new`, finds objects that are `TenantScopedMixin` instances without `company_id` set, and stamps them from the ContextVar.

The active tenant is stored in a `ContextVar[int | None]`, which is **per-asyncio-Task** and **per-thread**, making it safe for both async and threaded environments without any synchronization.

---

## Supported databases

| Database | Status |
|---|---|
| SQLite | Full support (tested in every CI run) |
| PostgreSQL | Full support (CI with service container) |
| MySQL / MariaDB | Full support (CI with service container) |

---

## Contributing

```bash
git clone https://github.com/suhanapthn24/softtenant
cd softtenant
pip install -e ".[dev]"
pytest
```

To run against all three database backends locally, set the env vars before running pytest:

```bash
export POSTGRES_URL="postgresql+psycopg2://user:pass@localhost/softtenant_test"
export MYSQL_URL="mysql+pymysql://user:pass@localhost/softtenant_test"
pytest
```

---

## License

MIT — see [LICENSE](LICENSE).
