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
NEXT_PUBLIC_API_URL environment variable in the web app.Tech Stack
Project Structure
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 # DependenciesAuthentication
The API uses JWT bearer tokens for authentication. Tokens are issued by the auth service and validated on every request via a FastAPI dependency:
# 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 userStandard Response Format
All endpoints return a consistent response envelope:
{
"success": true,
"data": { ... },
"message": "Operation completed successfully"
}{
"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:
# 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
Alert Management
POST/GET/PATCH/DELETECRUD operations for alert rules, incidents, and alert lifecycle management.
Escalation Policies
POST/GET/PUT/DELETEMulti-stage escalation workflows with configurable delays and notification targets.
Notification Channels
POST/GET/PUT/DELETEEmail, SMS, Slack, and webhook channel configuration.
Identity & Access
POST/GET/PUT/DELETEUser, group, role, and environment management endpoints.
FinOps Analytics
GETCost and usage data endpoints powered by Athena/CUR integration.