elements SDK

Every dependency below follows the same idea: connect to the right thing, the right way, automatically. You import a function, add it to a route with Depends(), and the connection — to the database, to Redis, to the permission check — is already pointed at the correct host, port, and pool for your deployment. You never construct a connection string yourself.

Database — FastAPI Routes

In FastAPI routes, use Depends() injection. The connection pool is initialized once and shared across all requests.

# your_service/core/databases.py
from elements.db.db_dependencies import create_db_dependency

get_db      = create_db_dependency("your_db", use_replica=False)  # writes
get_read_db = create_db_dependency("your_db", use_replica=True)   # reads

# your_service/api/routes.py
from fastapi import Depends
from ..core.databases import get_db, get_read_db

@router.post("/items")
async def create_item(data: ItemSchema, db=Depends(get_db)):
    await db.execute(
        "INSERT INTO items (id, name) VALUES ($1, $2)",
        uuid.uuid4(), data.name
    )

@router.get("/items")
async def list_items(db=Depends(get_read_db)):
    return await db.fetch("SELECT * FROM items ORDER BY created_at DESC")

Database — Async Workers

In background workers (not FastAPI), use the async for generator pattern. The connection is scoped to the block and released automatically.

from elements.db.db_dependencies import get_database_connection

async for conn in get_database_connection("your_db", use_replica=False):
    await conn.execute("UPDATE items SET status=$1 WHERE id=$2", "done", item_id)
    rows = await conn.fetch("SELECT * FROM queue WHERE status='pending'")

Database — Connection Details

Pools are created on first use per database name and pool type, then cached and shared for the rest of the process lifetime — calling create_db_dependency or get_database_connection again for the same database doesn't open a second pool. Pool sizes come from environment config: MIN_POOL_SIZE/MAX_POOL_SIZE (default 5–20) for primary, half that (default 2–10) for replica.

Built-in auth database helpers

The auth database is used so often that elements ships ready-made dependencies — no factory call needed:

from elements.db.db_dependencies import get_auth_db, get_auth_read_db

@router.get("/profile")
async def profile(db=Depends(get_auth_db)):
    return await db.fetchrow("SELECT * FROM auth.users WHERE id = $1", user_id)

Troubleshooting — bypassing HAProxy

If HAProxy or PgBouncer is misbehaving and you need to confirm whether the problem is at that layer, elements also exposes direct connections straight to each PostgreSQL node. Think of it as a "skip the middlemen" escape hatch — useful for diagnosis, not for normal application code:

from elements.db.db_dependencies import get_direct_primary_db, get_direct_replica_db

# Connects straight to db1, bypassing management:5432 / PgBouncer
async for conn in get_direct_primary_db("your_db"):
    rows = await conn.fetch("SELECT 1")

Health checks & shutdown

Every pool can report its own health, and all pools can be closed cleanly together — useful for /health endpoints and lifespan shutdown:

from elements.db.db_dependencies import check_db_health, shutdown_async_db

@router.get("/health")
async def health():
    return await check_db_health()
    # {"database": true, "database_your_db": {"primary": true, "replica": true}}

# In lifespan shutdown:
await shutdown_async_db()

Redis — FastAPI Routes

from elements.redis.redis_dependencies import get_redis

@router.post("/cache")
async def cache_item(redis=Depends(get_redis)):
    await redis.set("key", "value", ex=3600)
    value = await redis.get("key")
    await redis.lpush("my:queue", json.dumps(payload))

Redis — Async Workers

from elements.redis.redis_dependencies import init_discovery_redis

class MyWorker:
    async def start(self):
        self.redis_manager = await init_discovery_redis()

    async def process(self):
        async with self.redis_manager.get_connection() as redis:
            item = await redis.brpop("my:queue", timeout=5)
            if item:
                await self.handle(json.loads(item[1]))

Redis — Available Pools

elements wires up three separate Redis connection pools, each pointing at a different Redis instance for a different purpose. You don't choose a host or DB number yourself — you just import the dependency for the job:

from elements.redis.redis_dependencies import init_webhook_redis, shutdown_webhook_redis

# In a worker's startup (not a FastAPI route):
webhook_redis = await init_webhook_redis()

async with webhook_redis.get_connection() as conn:
    job = await conn.brpop("webhook:stripe", timeout=5)

# In the worker's shutdown:
await shutdown_webhook_redis()

Health checks & shutdown

Each pool reports its own status, the same way the database pools do:

from elements.redis.redis_dependencies import (
    check_redis_health, check_session_redis_health,
    shutdown_redis, shutdown_session_redis,
)

@router.get("/health")
async def health():
    discovery = await check_redis_health()
    session = await check_session_redis_health()
    return {"discovery": discovery, "session": session}

# In lifespan shutdown:
await shutdown_redis()
await shutdown_session_redis()

FastAPI Middleware

elements provides the common middleware every service needs — CORS and request logging — wired up with one call in your FastAPI app's startup.

# your_service/main.py
from fastapi import FastAPI
from contextlib import asynccontextmanager
from elements.fastapi.middleware import setup_common_middleware
from .core.databases import shutdown_async_db
from .api import router

@asynccontextmanager
async def lifespan(app):
    yield
    await shutdown_async_db()

app = FastAPI(lifespan=lifespan)
setup_common_middleware(app)
app.include_router(router, prefix="/api/your-service")

setup_common_middleware() adds two things:

• CORS — preconfigured for your registered domains. Pass cors_allowed_origins=[...] to override the defaults if your service needs a different set.
• Request logging — logs every request with method, path, status code, and duration, and adds X-Process-Time / X-Request-ID response headers. /, /health, /healthz, and /ready are skipped to keep logs quiet. Pass log_payloads=True to also log request bodies/query params while debugging.

# Override CORS origins and enable payload logging for debugging:
setup_common_middleware(
    app,
    cors_allowed_origins=["https://yourdomain.com"],
    log_payloads=True,
)

Authentication & Permissions (RBAC)

Login itself (Google OAuth, magic links, sessions) is handled by auth_service — your app doesn't implement login. What your app does do is check, on each request, whether the logged-in user is allowed to do the thing they're asking for. elements gives you two ready-made dependencies for that, and the actual decision is made by a PostgreSQL function — not by Python logic.

Think of it like a bouncer at a club who radios the manager (the database) and asks "is this person on the list, and what can they do?" — your route code never has to know the rules itself.

require_role — "are they at least a member?"

Roles are ranked: guest < member < premium < enterprise < admin. require_role checks the caller's role for the current domain is at or above the one you specify.

from elements.auth.permissions import require_role

@router.post("/articles")
async def create_article(user_id: str = Depends(require_role("premium"))):
    # Only premium, enterprise, or admin users reach this line.
    # Anything lower gets a 403 automatically.
    ...

require_permission — "can they do this specific thing?"

For finer-grained checks than a role tier — e.g. "can moderate articles" vs "can create articles" — use require_permission. It calls a SQL permission-check function and caches the result in session Redis for 5 minutes so repeated requests don't hit the database every time.

from elements.auth.permissions import require_permission

@router.post("/articles/{id}/moderate")
async def moderate_article(
    id: str,
    user_id: str = Depends(require_permission("articles.moderate"))
):
    ...

Both dependencies read the session cookie, look up the session in session Redis, then ask the database auth.get_user_tier() or auth.check_user_permission_domain() for the answer — whichever raises a 401/403 first short-circuits the route, so your handler body only runs for authorized requests.

Multi-domain aware: get_current_domain

A single deployment can serve more than one website/domain from the same backend. get_current_domain reads the incoming Host header, checks it against known active domains, and is what require_role / require_permission use to scope "what role does this user have on this domain" — the same user can be an admin on one domain and a guest on another.

from elements.auth.dependencies import get_current_domain

@router.get("/whoami")
async def whoami(domain: str = Depends(get_current_domain)):
    return {"domain": domain}

Service-to-service: API keys

For machine-to-machine calls (no browser session), use API key authentication instead. The key is verified — and its usage stats updated — by a single SQL function call:

from elements.auth.dependencies import get_user_from_api_key, require_api_permission

@router.post("/webhook/ingest")
async def ingest(caller=Depends(require_api_permission("webhook.write"))):
    ...

Logging

Structured JSON logging. All logs are written to /home/ubuntu/code/logs/{service}-{component}.log, uploaded to S3 by cron, and indexed in management Redis for full-text search via the ops dashboard.

from elements.logger import get_logger

logger = get_logger(component="my_component")

logger.info("User created", context={"user_id": str(user_id)})
logger.error("Payment failed", context={"intent_id": intent_id, "error": str(e)})
logger.debug("Cache hit", context={"key": cache_key})

# Also available: logger.warning(), logger.critical()
# logger.exception() auto-attaches the current traceback to context["traceback"]
try:
    risky_call()
except Exception:
    logger.exception("risky_call failed")

Log output (JSON, one line per entry):

{"timestamp": "2026-05-13T10:14:29Z", "level": "info", "service": "your_service",
 "component": "my_component", "hostname": "app1", "message": "User created",
 "context": {"user_id": "6eb083f1-..."}}

Email — Queue-Based

Don't call SES directly from your request handler — a slow SES call would block the response, and a failed send would be silently lost. Instead, enqueue_email drops the email job onto a Redis queue (on openresty-lb) and returns immediately. A dedicated email worker picks it up, sends via SES, and retries on failure with a dead-letter queue for jobs that exhaust their retries.

from elements.email.email_queue import enqueue_email

# Returns immediately with a job id — actual send happens async
job_id = await enqueue_email(
    to_email="user@example.com",
    subject="Welcome to Acme",
    html_content="<p>Thanks for signing up!</p>",
    text_content="Thanks for signing up!",
    from_email="noreply@yourdomain.com",
    service="your_service",
)

For emails that must never be sent twice (e.g. "your subscription was charged"), pass idempotency_key — the worker records a short-lived marker after sending and skips duplicate jobs with the same key:

await enqueue_email(
    to_email=user_email,
    subject="Payment receipt",
    html_content=receipt_html,
    text_content=receipt_text,
    idempotency_key=f"receipt:{invoice_id}",
    service="payment_service",
)

Config Manager

All environment variables are available through ConfigManager — a singleton backed by lru_cache that reads /etc/environment once.

from elements.config.manager import get_config_manager

config = get_config_manager()

# DB config
config.db_config.haproxy_host        # "management"
config.db_config.haproxy_primary_port # 5432
config.db_config.user                # "postgres"
config.db_config.password            # from PGPASSWORD env

# App config
config.app_env      # "prod"
config.aws_region   # "us-east-1"
config.s3_bucket    # "uploads-123456789012"

Building & Deploying Your Service

# your_service/pyproject.toml defines the wheel
# build-and-deploy.sh does the rest:

cd your-service-repo/
./build-and-deploy.sh

  Building wheel...        OK
  Uploading to S3...       OK
  Publishing to Redis...   OK
  [app1] Installing...     OK
  [app2] Installing...     OK
  Deployment complete.

The deployment subscriber on each app node picks up the Redis message, downloads the wheel from S3, installs it, and restarts the service — all in under 30 seconds, in parallel across all nodes.