DEV Community: Ugur Aslim

Multi-Tenant FastAPI + React: Architecture Decisions Explained

Ugur Aslim — Thu, 21 May 2026 18:34:11 +0000

Building multi-tenant SaaS is a sequence of architecture decisions where the wrong choices compound. A tenant isolation model that seemed reasonable at 10 tenants becomes a migration nightmare at 1000. An auth pattern that "works" creates a security gap under load.

This post documents every significant architecture decision in CitizenApp — a GDPR-compliant citizen management SaaS — and the reasoning behind each one. I'll be direct about the tradeoffs and what I'd do differently.

Tenant Isolation: Row-Level vs Schema-Per-Tenant vs Database-Per-Tenant

This is the first decision that constrains everything downstream.

Three options:

Strategy	Isolation	Complexity	Migration cost	Cross-tenant queries
`tenant_id` column	Logical	Low	Low	Easy
Schema per tenant	Physical	Medium	High	Hard
Database per tenant	Full	High	Very high	Impossible

For CitizenApp, I chose tenant_id column on every table. Here's the reasoning:

Scale: Under 500 tenants at launch. Schema-per-tenant is worth the complexity at 10,000+ tenants when you need schema-level customisation.
GDPR compliance: Logical isolation is sufficient for GDPR if implemented correctly. Physical isolation adds ops complexity without a compliance benefit at this scale.
Query patterns: I need cross-tenant analytics for the admin dashboard (aggregate counts, not row data). Schema-per-tenant makes this significantly harder.
Migration tooling: Alembic works cleanly with a single schema. Per-schema migrations require custom tooling.

The implementation:

Every data model inherits from a base that enforces tenant_id:

class TenantMixin:
    tenant_id: Mapped[int] = mapped_column(
        ForeignKey("tenants.id"), nullable=False, index=True
    )

class VatandasModel(TenantMixin, Base):
    __tablename__ = "vatandas"
    id: Mapped[int] = mapped_column(primary_key=True)
    ad_soyad: Mapped[str] = mapped_column(String(200))
    tc_no: Mapped[str] = mapped_column(String(11))  # encrypted at rest
    # ...

The enforcement layer — Repository pattern:

class CitizenRepository:
    def __init__(self, db: AsyncSession, tenant_id: int):
        self.db = db
        self.tenant_id = tenant_id

    async def get_all(self, skip: int = 0, limit: int = 100) -> list[VatandasModel]:
        result = await self.db.execute(
            select(VatandasModel)
            .where(VatandasModel.tenant_id == self.tenant_id)
            .offset(skip)
            .limit(limit)
        )
        return result.scalars().all()

    async def get_by_id(self, citizen_id: int) -> VatandasModel | None:
        result = await self.db.execute(
            select(VatandasModel)
            .where(
                VatandasModel.id == citizen_id,
                VatandasModel.tenant_id == self.tenant_id  # always scoped
            )
        )
        return result.scalar_one_or_none()

The tenant_id filter is in the repository, not the endpoint. Endpoints never construct raw queries — they call repository methods. This makes it structurally difficult to accidentally return cross-tenant data.

What I'd add now: PostgreSQL Row-Level Security policies as a second enforcement layer. Even if application code has a bug, RLS at the database level ensures no cross-tenant data leaks. It's belt-and-suspenders isolation.

JWT Design: What's in the Token and Why

Token design has permanent consequences. Tokens issued today will be used until they expire — you can't retroactively change their structure without forcing re-login.

CitizenApp's access token payload:

{
    "sub": "42",              # user_id
    "tenant_id": 7,           # tenant scope — embedded, not looked up per request
    "role": "operator",       # rbac role — same reasoning
    "email": "user@co.com",   # for audit logging without a DB lookup
    "exp": 1716307200,        # 15 minutes
    "type": "access"
}

Why tenant_id and role are in the token:

Every request needs these. If they're not in the token, every API call requires a database lookup to find the user's tenant and role. At 100 requests/second, that's 100 unnecessary DB queries/second. With token-embedded claims, the FastAPI middleware extracts everything it needs from the token signature verification — zero DB round-trips for auth.

The tradeoff: If you change a user's role, the old token is still valid for up to 15 minutes. This is acceptable for CitizenApp. If immediate revocation were required, I'd use short-lived tokens (5 minutes) plus a Redis token blocklist.

Refresh token rotation:

async def rotate_refresh_token(
    db: AsyncSession, 
    old_token: str
) -> tuple[str, str]:
    stored = await get_refresh_token(db, old_token)

    if not stored or stored.is_revoked:
        # Possible token theft — revoke entire family
        await revoke_token_family(db, stored.family_id)
        raise HTTPException(status_code=401, detail="Token reuse detected")

    await revoke_token(db, old_token)

    new_access  = create_access_token(stored.user_id, stored.tenant_id)
    new_refresh = await create_refresh_token(db, stored.user_id, stored.family_id)

    return new_access, new_refresh

Refresh tokens use family-based rotation. If a refresh token is used twice (token theft scenario), the entire family is revoked. This forces re-login and stops the attacker even if they got the token first.

RBAC: Three Roles, One Decision Point

CitizenApp has three roles: admin, operator, viewer.

What I didn't do: Fine-grained permissions per action. CASL, custom permission tables, or middleware that checks user.permissions.includes("citizen:delete").

Why not: Premature flexibility. At three roles with clear semantics, a permission matrix adds indirection without value. The role hierarchy is:

viewer → read-only access to citizen data
operator → read + write citizen data, run imports
admin → everything + user management + billing + tenant settings

The FastAPI implementation:

def require_role(*roles: str):
    def dependency(current_user: UserModel = Depends(get_current_user)):
        if current_user.role not in roles:
            raise HTTPException(status_code=403, detail="Insufficient permissions")
        return current_user
    return dependency

# Usage
@router.delete("/citizens/{id}")
async def delete_citizen(
    id: int,
    user: UserModel = Depends(require_role("admin", "operator")),
    repo: CitizenRepository = Depends(get_citizen_repo),
):
    ...

@router.get("/users")
async def list_users(
    user: UserModel = Depends(require_role("admin")),
    ...
):
    ...

Clean, readable, testable. When the permission model needs to evolve, it's one function.

The audit trail connects here: Every destructive action logs user_id, tenant_id, action, target_id, and before/after states. Compliance requires knowing who did what to which record at what time. The role doesn't change the requirement — it just determines whether the action was permitted.

Encryption at Rest: Fernet Over Column-Level Encryption

PII like the TC identity number (Turkish national ID) must be encrypted at rest. Three options:

Application-level encryption (Fernet/AES) — encrypt in Python before INSERT
PostgreSQL column encryption (pgcrypto) — encrypt at the database level
Transparent Data Encryption — encrypt at the filesystem/disk level

I chose application-level Fernet encryption:

from cryptography.fernet import Fernet

class EncryptionService:
    def __init__(self, key: bytes):
        self.fernet = Fernet(key)

    def encrypt(self, value: str) -> str:
        return self.fernet.encrypt(value.encode()).decode()

    def decrypt(self, value: str) -> str:
        return self.fernet.decrypt(value.encode()).decode()

# SQLAlchemy event listener — transparent to the rest of the code
@event.listens_for(VatandasModel.tc_no, "set")
def encrypt_tc_no(target, value, oldvalue, initiator):
    if value and not value.startswith("gAAAAA"):  # not already encrypted
        target.tc_no = encryption_service.encrypt(value)

Why not pgcrypto: pgcrypto requires the encryption key to be available to PostgreSQL. That means it's accessible to anyone with database access. Application-level encryption keeps the key in the application environment (Render secrets), separate from the database.

Why not TDE: TDE protects against physical disk theft. It doesn't protect against a compromised database user or a SQL injection that returns raw table data. Application-level encryption does.

Key rotation: Fernet supports key rotation via MultiFernet. Old ciphertext decrypts with old key; new data encrypts with new key. Zero-downtime rotation:

# During rotation window
fernet = MultiFernet([new_key_fernet, old_key_fernet])

Database: Async SQLAlchemy Patterns

CitizenApp uses FastAPI with asyncpg and async SQLAlchemy. This is the right choice but it has sharp edges.

Session scoping:

async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

Each request gets its own session. Auto-commit on success, auto-rollback on exception. No manual transaction management in endpoint code.

The N+1 trap:

# BAD: N+1 queries
citizens = await repo.get_all()
for c in citizens:
    c.audit_count = await audit_repo.count_for(c.id)  # N extra queries

# GOOD: single query with subquery
stmt = (
    select(VatandasModel, func.count(AuditModel.id).label("audit_count"))
    .outerjoin(AuditModel, AuditModel.citizen_id == VatandasModel.id)
    .where(VatandasModel.tenant_id == tenant_id)
    .group_by(VatandasModel.id)
)

Async SQLAlchemy doesn't protect you from N+1. You have to think about it explicitly. I caught several in code review — a query in a loop is always a red flag.

The lazy loading trap:

Async SQLAlchemy doesn't support lazy loading. Any relationship access outside a transaction raises MissingGreenlet. You must declare what you need upfront:

stmt = (
    select(UserModel)
    .options(selectinload(UserModel.tenant))  # explicit eager load
    .where(UserModel.id == user_id)
)

Migration Strategy: Alembic in Production

Every schema change in CitizenApp is an Alembic migration. No exceptions. No CREATE TABLE in the application startup code.

The pattern that works:

# Never autogenerate directly to production
alembic revision --autogenerate -m "add_ai_credits_to_users"
# Review the generated file
# Add explicit data migrations if needed
alembic upgrade head

Zero-downtime migrations — the rules:

Never rename a column directly. Add the new column, backfill, update application code, then drop the old column in a separate migration run.
New NOT NULL columns need a DEFAULT or a two-step migration. Adding NOT NULL without a default locks the table.
Index creation should be CONCURRENT. CREATE INDEX CONCURRENTLY doesn't lock reads/writes. Standard CREATE INDEX does.

# In Alembic migration — concurrent index
op.execute("CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_vatandas_tenant_created ON vatandas(tenant_id, created_at DESC)")

Alembic doesn't generate CONCURRENTLY by default. You have to write it manually.

Frontend Architecture: React 19 + TanStack Query

State management decision: No Redux, no Zustand. TanStack Query for server state, useState for local UI state.

The vast majority of React state is server data — citizen lists, user lists, audit logs. TanStack Query handles fetching, caching, background refresh, and optimistic updates. There's no global client-side state that needs a store.

React 19 useOptimistic for instant UX:

const [optimisticCitizens, addOptimistic] = useOptimistic(citizens);

async function handleDelete(id: number) {
    addOptimistic(prev => prev.filter(c => c.id !== id));  // instant

    try {
        await deleteCitizen(id);
        await queryClient.invalidateQueries(["citizens"]);
    } catch {
        // useOptimistic rolls back automatically on error
        toast.error("Delete failed");
    }
}

The UI updates immediately. If the API call fails, the optimistic update rolls back. Users get instant feedback without fake loading states.

Tenant context:

The tenant is embedded in the JWT. The frontend decodes it (without verification — signature verification is the server's job) to personalise the UI:

function useTenant() {
    const token = localStorage.getItem("access_token");
    if (!token) return null;
    const payload = JSON.parse(atob(token.split(".")[1]));
    return { id: payload.tenant_id, plan: payload.plan };
}

This is fine. The frontend showing wrong data is a UX problem. The backend returning wrong data would be a security problem. The backend always validates the token independently.

What I'd Change

1. PostgreSQL RLS from day one.

Application-level tenant_id filtering is correct but not sufficient as a guarantee. RLS at the database layer would make it impossible for a bug in the repository layer to expose cross-tenant data. Retrofitting RLS on an existing schema is painful — it should be designed in from the start.

2. Separate read and write models (CQRS-lite).

The list endpoints in CitizenApp do a lot of work: filter, sort, paginate, aggregate audit counts. This would be cleaner with dedicated read models or database views optimised for list queries, separate from the write models.

3. Stricter API versioning.

CitizenApp is currently on v1 with no formal versioning. For a multi-tenant SaaS where different tenants may be on different frontend versions, /api/v1/ from the start with a migration plan for v2 would have been smarter.

4. Event sourcing for audit.

The audit log records events but not the full before/after state for every field. Reconstructing "what did this record look like at this point in time" requires replaying events. A proper event-sourced audit table with full snapshots would make compliance reporting significantly easier.

The Bottom Line

Multi-tenant SaaS architecture is about defaults that are safe by construction. Tenant isolation in the repository layer, role enforcement in dependencies, encryption at the ORM level — the goal is to make the dangerous thing hard to do accidentally.

Every architectural decision in CitizenApp was made with the assumption that I would make mistakes later. The architecture's job is to make those mistakes non-catastrophic.

Building a multi-tenant SaaS and want a second opinion on the architecture? I'm available for consulting.

How I Built 9 Claude AI Features into a Production SaaS

Ugur Aslim — Thu, 21 May 2026 18:32:21 +0000

Nine AI features. One production SaaS. Zero hallucinated data in the user's face.

That last part is harder than it sounds.

This is a full technical breakdown of how I designed, built, and shipped nine Claude Haiku features into CitizenApp — a GDPR-compliant citizen management platform. I'll cover the architecture, the security decisions, the failure modes I had to defend against, and what I'd do differently.

Why Haiku, Not Sonnet or Opus

Every AI feature in CitizenApp uses claude-haiku-4-5. The reasoning was deliberate:

Cost. This is a multi-tenant SaaS. Each AI call deducts from the tenant's credit balance. Haiku is significantly cheaper per token than Sonnet — which means I can offer meaningful AI functionality at the Free tier without destroying margins.

Speed. Most features are interactive — the user is waiting. Haiku's latency is typically under 2 seconds for the prompt sizes I'm sending. Sonnet would double or triple that.

Sufficient capability. For structured tasks — SQL generation, data summarisation, duplicate record comparison — Haiku is good enough. I don't need Opus-level reasoning. I need fast, cheap, reliable JSON out.

Consistent timeout budget. Every AI call has a 20-second timeout. Haiku almost always completes in under 5. That matters when you're running on Render's free tier.

The Credit System

Before the features themselves: every AI call in CitizenApp costs credits. This is enforced at the API layer, not the frontend.

async def deduct_credits(db: AsyncSession, tenant_id: int, amount: int = 1) -> None:
    tenant = await db.get(TenantModel, tenant_id)
    if tenant.ai_credits < amount:
        raise HTTPException(status_code=402, detail="Insufficient AI credits")
    tenant.ai_credits -= amount
    await db.commit()

Simple. No credits, no AI call. The 402 Payment Required response tells the frontend to show the upgrade prompt. Credits are deducted before the Claude API call — if the call fails, I refund them. This prevents users from draining credits on broken requests.

Stats queries cost 1 credit. Live DB queries cost 2. The pricing reflects actual Claude token consumption.

Feature 1: CSV Column Mapping

The problem: Users import citizen data from spreadsheets. The column headers are inconsistent — TC No, tc_no, Kimlik No, ID Number — all mean the same thing. Asking users to manually map every import is a UX failure.

The implementation:

async def ai_map_columns(headers: list[str]) -> dict:
    prompt = f"""You are mapping CSV column headers to a fixed schema.

Schema fields: tc_no, ad_soyad, dogum_tarihi, cinsiyet
Headers: {headers}

Return JSON only:
{{"tc_no": "header or null", "ad_soyad": "header or null", 
  "dogum_tarihi": "header or null", "cinsiyet": "header or null"}}"""

    response = await anthropic_client.messages.create(
        model="claude-haiku-4-5",
        max_tokens=200,
        messages=[{"role": "user", "content": prompt}]
    )
    return json.loads(response.content[0].text)

What I learned: Keep the schema description short and the output format rigid. Early versions asked Claude to "suggest the best mapping" — vague enough that it would sometimes return explanatory text instead of JSON. Specifying Return JSON only: with the exact structure eliminated this.

Failure mode handled: If json.loads throws, fall back to fuzzy string matching. AI is the first attempt, not the only one.

Feature 2: Natural Language Search

The problem: Users need to filter 50,000+ citizen records. SQL is not the interface.

The implementation:

ALLOWED_COLUMNS = {"ad_soyad", "dogum_tarihi", "cinsiyet", "created_at", "il", "ilce"}

async def nl_to_sql_filter(query: str, tenant_id: int) -> str:
    prompt = f"""Convert this natural language query to a PostgreSQL WHERE clause.

Table: vatandas
Available columns: {', '.join(ALLOWABLE_COLUMNS)}
tenant_id is always {tenant_id} — always include it.

Query: "{query}"

Return only the WHERE clause starting with 'WHERE'. No explanation."""

    response = await anthropic_client.messages.create(
        model="claude-haiku-4-5",
        max_tokens=300,
        messages=[{"role": "user", "content": prompt}]
    )

    clause = response.content[0].text.strip()
    validate_sql_clause(clause)  # raises on dangerous patterns
    return clause

The critical part — validate_sql_clause:

BLOCKED = ["drop", "delete", "update", "insert", "alter", "exec", 
           "--", ";", "union", "sleep", "pg_"]

def validate_sql_clause(clause: str) -> None:
    lower = clause.lower()
    if not lower.startswith("where"):
        raise ValueError("Invalid clause")
    for pattern in BLOCKED:
        if pattern in lower:
            raise ValueError(f"Blocked pattern: {pattern}")

Never trust LLM output with SQL. The allowed columns list is in the prompt — if Claude generates a column not in that list, SQLAlchemy will raise a column-not-found error before anything executes. Defense in depth.

Feature 3: Anomaly Detection

The problem: Bad data accumulates. Future birthdates, 1890-born citizens, missing gender on 40% of records. Users need to find it without writing reports.

The implementation: I run five deterministic checks in Python — no AI involved in detecting the anomalies:

def detect_anomalies(citizens: list[dict]) -> list[dict]:
    issues = []
    today = date.today()

    for c in citizens:
        if c["dogum_tarihi"] and c["dogum_tarihi"] > today:
            issues.append({"id": c["id"], "type": "future_dob", "severity": "high"})
        if c["dogum_tarihi"] and c["dogum_tarihi"].year < 1900:
            issues.append({"id": c["id"], "type": "implausible_dob", "severity": "medium"})
        if not c["cinsiyet"]:
            issues.append({"id": c["id"], "type": "missing_gender", "severity": "low"})

    return issues

Claude's role is to summarise the results:

summary_prompt = f"""Summarise these data quality issues in 2-3 sentences for a non-technical admin:
{json.dumps(issue_counts)}
Be specific about numbers. Suggest the highest-priority fix first."""

This is the right division of labour. Python finds the issues deterministically. Claude explains them in plain language. AI never makes the binary "is this an anomaly" judgment — it just communicates results.

Feature 4: AI Daily Briefing

The simplest feature architecturally. Every morning, the admin dashboard can show a 3-4 sentence Claude-generated briefing:

stats = {
    "total_records": 12847,
    "missing_gender": 342,
    "missing_dob": 89,
    "potential_duplicates": 23,
    "created_last_7_days": 156
}

prompt = f"""You are a data assistant. Write a 3-4 sentence briefing for an admin 
about the current state of their citizen database. Be concise and actionable.

Stats: {json.dumps(stats)}"""

Lesson: Don't send raw PII to the API. Stats are safe. Individual citizen records are not. I aggregate before sending.

Feature 5: AI Duplicate Merge

This was the most complex feature to get right.

The problem: The duplicate detection identifies potential duplicates by similarity score. But merging them requires judgment — which record has the correct spelling? Which birthdate is more likely accurate?

The implementation:

async def ai_merge_suggest(record_a: dict, record_b: dict) -> dict:
    # Mask TC no before sending to Claude
    safe_a = {k: v for k, v in record_a.items() if k != "tc_no"}
    safe_b = {k: v for k, v in record_b.items() if k != "tc_no"}

    prompt = f"""Compare these two citizen records and suggest the best merged version.

Record A (id={record_a['id']}): {json.dumps(safe_a)}
Record B (id={record_b['id']}): {json.dumps(safe_b)}

Return JSON:
{{
  "keep_id": <id of the record to keep as primary>,
  "merged": {{"ad_soyad": "...", "dogum_tarihi": "YYYY-MM-DD", "cinsiyet": "E/K"}},
  "confidence": "high|medium|low",
  "reasoning": "one sentence explanation"
}}"""

TC identity number (TC kimlik no) is always masked before sending to Claude. This is non-negotiable. It's PII, it's regulated, and it's not needed for the merge decision. The keep_id in the response tells the backend which record to promote — Claude never sees or decides on the TC number.

Claude's output is never applied automatically. It's a suggestion displayed to the admin, who clicks "Apply" to confirm. The AI is an advisor, not an actor.

Feature 6: Natural Language Reports

The problem: Admins need custom data exports. "Give me all female citizens born between 1960 and 1970 in Ankara" is not a filter the UI can anticipate.

The implementation: A dedicated /reports page accepts a natural language query, sends it to Claude for SQL filter generation (same pattern as Feature 2), shows a preview table, then allows CSV/TSV export — all client-side using Blob and URL.createObjectURL.

The backend generates the filtered dataset. The download happens in the browser. No server-side file generation, no temp file cleanup, no storage costs.

Features 7 & 8: AI Chat — Stats and Live DB Query

The floating chat widget was the most technically interesting feature.

Two modes in one endpoint:

Stats mode — Claude answers questions from aggregate statistics only. "How many records are missing a birthdate?" The answer comes from pre-computed stats, not a live query.
DB query mode — For questions that can't be answered from stats, Claude generates a SQL SELECT statement that the backend executes directly.

The DB query mode required careful safety engineering:

SAFE_QUERY_RULES = """
Generate a PostgreSQL SELECT query. Hard rules:
- Only query table: vatandas
- Never include column: tc_no  
- Always filter: WHERE tenant_id = {tenant_id}
- Always add: LIMIT 100
- No subqueries, no JOINs to other tables
- No aggregations that could expose cross-tenant data
"""

def validate_generated_query(sql: str, tenant_id: int) -> str:
    sql_lower = sql.lower().strip()

    assert sql_lower.startswith("select"), "Must be SELECT"
    assert "tc_no" not in sql_lower, "tc_no is blocked"
    assert f"tenant_id = {tenant_id}" in sql_lower, "tenant_id filter required"
    assert "limit" in sql_lower, "LIMIT required"

    # Extract and enforce limit
    limit_match = re.search(r"limit\s+(\d+)", sql_lower)
    if limit_match and int(limit_match.group(1)) > 100:
        sql = re.sub(r"limit\s+\d+", "LIMIT 100", sql, flags=re.IGNORECASE)

    return sql

The query executes with a 5-second statement timeout at the database level. If Claude generates something slow, it gets cancelled.

Query result display: Results are shown in a collapsible QueryBlock with the SQL visible above the data table. Users can see exactly what ran. Transparency matters when AI is touching your database.

Feature 9: AI Audit Trail Explainer

The audit log records 30+ event types — citizen_created, bulk_import, login_failed, 2fa_enabled, etc. Non-technical admins can't interpret raw event streams.

async def explain_audit_events(events: list[dict]) -> str:
    # Strip internal IDs and technical fields
    safe_events = [
        {"type": e["event_type"], "user": e["user_email"], "time": e["created_at"]}
        for e in events[:20]  # Last 20 events only
    ]

    prompt = f"""Summarise this audit log in plain language for a non-technical admin.
Note any unusual patterns (multiple failed logins, bulk operations, permission changes).
2-4 sentences maximum.

Events: {json.dumps(safe_events)}"""

Cap at 20 events. Don't send the full log. Keep the token budget predictable.

What I'd Do Differently

1. Add structured output from the start.

Anthropic's tool use / structured output API is cleaner than prompt engineering Return JSON only. I retrofitted it on some features. It should have been the default.

2. Cache more aggressively.

The daily briefing doesn't need to be generated every page load. A 6-hour Redis cache with background refresh would cut AI costs significantly. Currently it regenerates on every dashboard open.

3. Separate AI service layer.

All nine features call Claude via a thin wrapper in app/services/ai.py. This is good for centralised error handling and logging. What I should have added sooner: a proper prompt registry so prompts are versioned and testable independently of the endpoint logic.

4. Test AI outputs, not just AI calls.

My test suite mocks the Anthropic client. That's necessary but insufficient — it doesn't catch prompt regressions. A small set of golden-output integration tests against real Claude (run weekly, not on every CI run) would give me confidence that prompt changes don't silently break feature behaviour.

The Result

Nine features, all in production, all credit-metered, all with safety validation on AI outputs. The platform handles real data for real users.

The engineering pattern that made this possible: AI generates, Python validates, humans confirm. The LLM is never the last line of defence.

Building AI features into your product? Let's talk about what that looks like.