APIFastAPIPython 3.10+PostgreSQL

Quper API

The Quper API is a comprehensive multi-tenant alerting, escalation, and monitoring backend built with FastAPI and PostgreSQL. It provides sophisticated alert rule management, notification routing, escalation policies, role-based access control, and FinOps analytics capabilities.

Base URL

The API is served via FastAPI with Uvicorn. In production deployments it runs behind an nginx reverse proxy. Configure the base URL via the NEXT_PUBLIC_API_URL environment variable in the web app.

Tech Stack

Framework
FastAPI 0.116.0
Async Python web framework
ORM
SQLAlchemy 1.4
Async + sync support
Database
PostgreSQL
Primary data store
Migrations
Alembic
Schema versioning
AWS SDK
boto3 1.38.46
Redshift & S3 integration
Scheduler
APScheduler
Background job management
Auth
PyJWT
JWT token handling
Driver
asyncpg + psycopg2
Async + sync PostgreSQL

Project Structure

Directory Layout
cricut.quper.api/
├── main.py                    # FastAPI app initialization & router registration
├── start_server.py            # Uvicorn entry point
├── app/
│   ├── api/
│   │   └── v1/
│   │       ├── alert_rules/   # Alert rule CRUD endpoints
│   │       ├── alerts/        # Alert lifecycle endpoints
│   │       ├── escalation/    # Escalation policy endpoints
│   │       ├── notifications/ # Notification channel endpoints
│   │       ├── schedules/     # On-call schedule endpoints
│   │       ├── users/         # User management endpoints
│   │       ├── groups/        # Group management endpoints
│   │       ├── roles/         # Role management endpoints
│   │       ├── environments/  # Environment endpoints
│   │       └── finops/        # FinOps analytics endpoints
│   ├── models/                # SQLAlchemy ORM models
│   ├── schemas/               # Pydantic request/response schemas
│   ├── services/              # Business logic layer
│   ├── repositories/          # Database access layer
│   └── core/
│       ├── database.py        # DB connection & session factory
│       ├── config.py          # Environment variable config
│       └── auth.py            # JWT auth dependencies
├── alembic/                   # Database migrations
└── pyproject.toml             # Dependencies

Authentication

The API uses JWT bearer tokens for authentication. Tokens are issued by the auth service and validated on every request via a FastAPI dependency:

python
# FastAPI auth dependency
async def get_current_user(
    token: str = Depends(oauth2_scheme),
    db: AsyncSession = Depends(get_db)
) -> User:
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
        user_id = payload.get("sub")
        if user_id is None:
            raise HTTPException(status_code=401)
    except JWTError:
        raise HTTPException(status_code=401)

    user = await user_repository.get_by_id(db, user_id)
    if user is None:
        raise HTTPException(status_code=401)
    return user

Standard Response Format

All endpoints return a consistent response envelope:

Success Response
{
  "success": true,
  "data": { ... },
  "message": "Operation completed successfully"
}
Error Response
{
  "success": false,
  "error": "Resource not found",
  "detail": "Alert rule with id 'abc123' does not exist"
}

Multi-tenancy

Every request is scoped to a tenant (environment). The tenant ID is extracted from the JWT payload and used to filter all database queries:

python
# All queries include tenant_id filter
async def get_alert_rules(
    db: AsyncSession,
    tenant_id: str,  # Extracted from JWT
    skip: int = 0,
    limit: int = 100,
) -> list[AlertRule]:
    result = await db.execute(
        select(AlertRule)
        .where(AlertRule.tenant_id == tenant_id)
        .offset(skip)
        .limit(limit)
    )
    return result.scalars().all()

API Modules