DEV Community: Jayanth

Multi-tenant PostgreSQL: row-level security vs schema-per-tenant & when to use which

Jayanth — Sun, 24 May 2026 10:40:30 +0000

If you're building a multi-tenant SaaS, this is the first real architecture

decision that will haunt you if you get it wrong.
I've implemented both approaches in production. Here's the honest trade-off.

Option A: Shared schema with row-level security (RLS)

Every tenant's data lives in the same tables. A tenant_id column on every
row. PostgreSQL RLS policies enforce that queries only ever return rows
belonging to the current tenant.

-- Enable RLS on the table
ALTER TABLE orders ENABLE ROW LEVEL SECURITY;

-- Policy: users only see their tenant's rows
CREATE POLICY tenant_isolation ON orders
    USING (tenant_id = current_setting('app.current_tenant_id')::uuid);

# Set the tenant context before every query
async def set_tenant(conn, tenant_id: str):
    await conn.execute(
        "SELECT set_config('app.current_tenant_id', $1, true)",
        tenant_id
    )

Works well when: You have many small tenants. Hundreds or thousands.
Schema-per-tenant at that scale is unmanageable — migrations alone would take hours.

Breaks when: A noisy tenant runs heavy queries and degrades performance for others. You can't easily move one tenant's data to a separate DB. You need different retention policies per tenant.

Option B: Schema per tenant

Each tenant gets their own PostgreSQL schema — effectively a namespace.
tenant_abc.orders, tenant_xyz.orders. Same tables, different schema.

-- Create schema for a new tenant
CREATE SCHEMA tenant_abc;

-- Set search path at connection time
SET search_path TO tenant_abc, public;

# Alembic migration across all tenant schemas
from alembic import command
from alembic.config import Config

def migrate_all_tenants(tenant_schemas: list[str]):
    for schema in tenant_schemas:
        alembic_cfg = Config("alembic.ini")
        alembic_cfg.set_main_option("sqlalchemy.url", db_url)
        alembic_cfg.set_section_option("alembic", "version_table_schema", schema)
        command.upgrade(alembic_cfg, "head")

Works well when: You have fewer, larger tenants. Enterprise customers
who need data isolation guarantees, custom retention, or the ability to
export their entire dataset cleanly.
Breaks when: You have 500+ tenants. Running migrations across 500
schemas sequentially is slow. Connection pool overhead grows.

What I actually use
For most SaaS products at early stage: start with RLS. It's simpler to
operate, migrations are trivial, and you can always move to schema-per-tenant
for specific large customers later by routing them to a dedicated schema
or even a dedicated database.
The hybrid approach — RLS for SMB tenants, dedicated schema for enterprise —
is what I've settled on. Your connection string is the router.

def get_db_url(tenant: Tenant) -> str:
    if tenant.tier == "enterprise":
        return tenant.dedicated_db_url
    return f"{shared_db_url}?options=-csearch_path={tenant.schema}"

One thing nobody tells you: test your RLS policies with a superuser disabled.
PostgreSQL superusers bypass RLS by default. Your staging environment running
as a superuser will never catch a broken policy. Use a restricted role in tests.

Distributed tracing across FastAPI and Celery with OpenTelemetry the part nobody shows you

Jayanth — Sun, 24 May 2026 10:33:02 +0000

Every OpenTelemetry tutorial ends at "add the middleware to FastAPI." That gets
you traces for HTTP requests. It tells you nothing about what happens inside
your Celery workers.
The hard part is propagating trace context from your FastAPI handler into the
Celery task, so a single request trace shows the full chain: HTTP → queue →
worker → database.
Here's the exact setup I use in production.
Why context propagation breaks by default
When FastAPI handles a request, the OTEL SDK creates a span and stores the
trace context in Python's contextvars. When you call .delay() to queue a
Celery task, that context is not automatically passed — Celery serialises
the task arguments to JSON and ships them to Redis. The trace context lives
in memory, not in the message. The worker starts with no idea what trace
it belongs to.
The fix: manually inject the trace context into the task headers, then
extract it in the worker before creating child spans.

Step 1: Inject context when queuing

from opentelemetry import propagate
from opentelemetry.trace import get_current_span

def queue_task_with_context(payload: dict):
    carrier = {}
    propagate.inject(carrier)  # pulls current trace context into dict

    process_order.apply_async(
        kwargs={"payload": payload},
        headers={"otel_context": carrier},
    )

Step 2: Extract context in the worker

from celery.signals import task_prerun
from opentelemetry import propagate, trace

tracer = trace.get_tracer(__name__)

@task_prerun.connect
def on_task_prerun(task_id, task, args, kwargs, **kw):
    headers = task.request.get("otel_context", {})
    ctx = propagate.extract(headers)  # restore parent trace context

    span = tracer.start_span(
        name=f"celery.{task.name}",
        context=ctx,
        kind=trace.SpanKind.CONSUMER,
    )
    task.request.otel_span = span
    task.request.otel_token = trace.use_span(span, end_on_exit=False).__enter__()

from celery.signals import task_postrun, task_failure

@task_postrun.connect
def on_task_postrun(task, **kw):
    span = getattr(task.request, "otel_span", None)
    if span:
        span.set_status(trace.StatusCode.OK)
        span.end()

@task_failure.connect
def on_task_failure(task, exception, **kw):
    span = getattr(task.request, "otel_span", None)
    if span:
        span.record_exception(exception)
        span.set_status(trace.StatusCode.ERROR)
        span.end()

# SDK setup in your FastAPI app
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.trace.export import BatchSpanProcessor

exporter = OTLPSpanExporter(endpoint="http://otel-collector:4317", insecure=True)
provider.add_span_processor(BatchSpanProcessor(exporter))

What you get
One trace ID that follows a request from the HTTP layer through the queue
into the Celery worker and down into your database query. When something
breaks at 2 AM, you open Grafana, search by trace ID, and see exactly
where it failed and how long each step took.
Without this, you have logs from three different services with no way to
connect them. With this, you have one timeline.
The full runnable repo with docker-compose, FastAPI app, Celery worker,
and Grafana Tempo is at github.com/itsjayanth/observability-starter-kit.

How I built an idempotent async task queue with Celery, Redis, and FastAPI

Jayanth — Sun, 24 May 2026 10:30:30 +0000

Most tutorials show you how to send a task to Celery. None of them show you what
happens when that task runs twice.
In production, tasks run twice constantly — network retries, worker crashes,
accidental re-queues. If your task isn't idempotent, you'll process the same order
twice, send the same email twice, or charge a customer twice.
Here's exactly how I built idempotent task execution in a multi-tenant workload.
The problem
A simplified version of what I was dealing with:

Webhook comes in → task queued → worker processes it
Worker dies mid-execution → task requeued → second worker picks it up
Same task runs twice → duplicate side effects

The naive fix is "just check if it already ran." But that check itself has a race
condition if two workers pick up the same task simultaneously.
The solution: idempotency keys + atomic locking
Every task gets an idempotency key derived from its input — not a random UUID,
but a deterministic hash of the payload. If the same payload comes in twice, it
produces the same key.

import hashlib
import json

def make_idempotency_key(task_name: str, payload: dict) -> str:
    payload_hash = hashlib.sha256(
        json.dumps(payload, sort_keys=True).encode()
    ).hexdigest()[:16]
    return f"task:{task_name}:{payload_hash}"

Before executing, the worker tries to acquire a Redis lock using SET NX EX
(set if not exists, with expiry). Only one worker wins the lock. The others
see the key exists and exit cleanly.

import redis
from celery import Task

r = redis.Redis(host="localhost", port=6379)

class IdempotentTask(Task):
    def __call__(self, *args, **kwargs):
        key = make_idempotency_key(self.name, kwargs)
        lock_acquired = r.set(key, "processing", nx=True, ex=300)

        if not lock_acquired:
            # Another worker is handling this — safe to skip
            return {"status": "skipped", "reason": "duplicate"}

        try:
            result = self.run(*args, **kwargs)
            r.set(key, "completed", ex=86400)  # mark done for 24hrs
            return result
        except Exception:
            r.delete(key)  # release lock on failure so it can retry
            raise

What about the dead-letter queue?
Tasks that fail repeatedly (after your retry limit) go to a DLQ — a separate
Redis list or SQS queue depending on your broker. The DLQ is where you debug,
not where you panic.

# celery config
task_acks_late = True          # only ack after success — not on receipt
task_reject_on_worker_lost = True  # requeue if worker dies mid-task
task_serializer = "json"

# per-task retry config
@app.task(
    bind=True,
    max_retries=3,
    default_retry_delay=60,
    queue="default",
)
def process_webhook(self, payload: dict):
    try:
        handle(payload)
    except TransientError as e:
        raise self.retry(exc=e, countdown=2 ** self.request.retries)
    except PermanentError:
        # don't retry — send to DLQ manually
        send_to_dlq(payload)

What I learned
Three things that aren't obvious until you've been burned:
task_acks_late = True is not optional. Default Celery behaviour acknowledges
the task the moment a worker receives it. If the worker crashes before finishing,
the task is gone. Late acks mean the message stays in the queue until the task
succeeds.
Lock expiry must be longer than your task timeout. If your task takes 4
minutes and your lock expires in 3, a second worker will pick it up while the
first is still running. Set ex to at least 2× your p99 task duration.
The idempotency key must be deterministic from the payload, not the task ID.
Celery generates a new task ID on every retry. If you key on task ID, retries
won't detect duplicates. Key on the business payload instead.
This setup has been running in production handling clinical imaging workloads
for over a year with zero duplicate executions.

Hey, There!

Jayanth — Sun, 24 May 2026 10:13:36 +0000

Test post