Adding New Endpoints¶
Follow this pattern for every new endpoint group.
Layer Architecture¶
Route (routes/*.py) ← Thin JSON wrapper, validates input
↓
Service (services/*.py) ← Business logic, orchestration
↓
Repository (db/repositories/*.py) ← SQL queries, data access
↓
PostgreSQL ← The AppMecca database
Each layer has a single responsibility:
- Routes parse HTTP requests, resolve the caller's scope, call the service, return JSON responses
- Services coordinate between repositories, AWS clients, and business rules
- Repositories execute SQL queries and return raw data
There is no environment parameter
The container is bound to one environment at deploy time, and each
environment has its own database. Read the value from
appmecca_api.config.env() if you need it — never accept it from the
caller, and never filter SQL on it (the column was dropped in migration
010).
Step-by-Step¶
1. Create the Repository¶
# src/appmecca_api/db/repositories/apps.py
from typing import Any
from appmecca_api.db.repositories.base import BaseRepository
class AppRepository(BaseRepository):
"""Data access for the apps table."""
async def list_apps(self, group_id: Any = None) -> list[dict]:
"""List apps, optionally narrowed to one group."""
if group_id is None:
rows = await self.fetch_all(
"SELECT * FROM apps ORDER BY app_name",
)
else:
rows = await self.fetch_all(
"SELECT * FROM apps WHERE group_id = $1 ORDER BY app_name",
group_id,
)
return [dict(r) for r in rows]
async def get_app(self, app_name: str, branch: str = "main") -> dict | None:
"""Get a single app by canonical name and branch."""
row = await self.fetch_one(
"SELECT * FROM apps WHERE app_name = $1 AND branch = $2",
app_name, branch,
)
return dict(row) if row else None
BaseRepository provides fetch_one, fetch_all, fetch_val, execute, execute_returning and exists. Every one leases a connection from the shared pool for a single statement.
Splice predicates before the ORDER BY
Appending a WHERE fragment to a query that already ends in ORDER BY is a
syntax error — and if the caller wraps the query in a broad except, it is
a silent one that ships an empty result with no traceback.
2. Create the Service¶
# src/appmecca_api/services/app_service.py
from typing import Any
import structlog
from appmecca_api.db.repositories.apps import AppRepository
logger = structlog.get_logger()
class AppService:
"""Business logic for app management."""
def __init__(self, repo: AppRepository) -> None:
self._repo = repo
async def list_apps(self, group_id: Any = None) -> list[dict]:
"""List apps in the caller's group scope."""
apps = await self._repo.list_apps(group_id=group_id)
logger.info("apps_listed", count=len(apps))
return apps
3. Create Pydantic Models¶
# src/appmecca_api/models/apps.py
from pydantic import BaseModel
class AppResponse(BaseModel):
"""App detail response."""
app_name: str
display_name: str | None
slug: str
branch: str
repository: str
deploy_status: str
latest_build: str | None
4. Create the Route¶
# src/appmecca_api/routes/apps.py
from fastapi import APIRouter, Depends
from appmecca_api.auth.group_scope import GroupScope, resolve_group
from appmecca_api.db.connection import get_pool
from appmecca_api.db.repositories.apps import AppRepository
from appmecca_api.services.app_service import AppService
router = APIRouter(prefix="/api/apps", tags=["apps"])
async def _get_service() -> AppService:
pool = await get_pool()
return AppService(AppRepository(pool))
@router.get("")
async def list_apps(
scope: GroupScope = Depends(resolve_group),
service: AppService = Depends(_get_service),
) -> list[dict]:
"""List apps in the caller's group scope."""
# A platform-wide admin key has no group binding — lift the filter.
effective_group_id = None if scope.platform_wide else scope.group_id
return await service.list_apps(group_id=effective_group_id)
5. Gate the Route¶
Authorization is not optional, and picking the right gate matters:
| Gate | Use for |
|---|---|
resolve_group |
Any list endpoint — yields the caller's GroupScope |
gate_app_scope(...) |
Any route with an {app_name} path parameter; resolves slug → canonical name and 404s (never 403) on a cross-tenant miss |
require_platform_operator() |
Platform-operator endpoints (agents, api-keys, platform config) |
require_user() |
Any authenticated human; also rejects disabled accounts |
require_scope("deploy") |
Tenant-facing routes where the real boundary is still enforced by gate_app_scope |
require_scope(\"admin\") is not an operator gate
Every self-registered tenant OWNS their personal group, and an OWNER maps to
the admin api-scope. A platform-operator endpoint must use
require_platform_operator().
Ask the same question twice
When two decisions both depend on "did the caller supply this?", use the
same test for both. Truthiness (if app_name:) and identity
(app_name is None) disagree on "" — and FastAPI binds a bare ?app_name=
to "", not None. That disagreement has been a live authorization bypass.
Normalise the value once at the route.
6. Declare a Cache Tier¶
Add the route to src/appmecca_api/data/api_cache_behaviours.yaml. CacheHeadersMiddleware reads it to stamp Cache-Control.
Only mark a tier shared: true when the response is provably identical for every caller — that flag is what makes CloudFront cache it cross-caller. A per-caller response in a shared tier is a data leak at the CDN.
7. Mount the Router¶
# In main.py, inside create_app():
from appmecca_api.routes.apps import router as apps_router
app.include_router(apps_router)
Declaration order decides matching
Starlette matches routes in declaration order. A literal path such as
/api/status/stable declared after a /{app_name} catch-all in the same
router is unreachable — the catch-all answers it, silently and with the
wrong body.
8. Write Tests¶
Write tests for each layer:
- Repository tests — mock the pool, verify SQL queries
- Service tests — mock the repository, verify business logic
- Route tests — use the async client, verify HTTP behavior; override gates via
app.dependency_overrides - Scoping tests — a cross-tenant caller must get 404, and a widened predicate must fail a real-PostgreSQL test
Assert the positive
Prefer asserting that the right thing happened over asserting that a string is absent. Negative assertions pass trivially when the failure mode simply never produces that string, which is how several tests have pinned nothing at all. If you must negate, first prove the string appears in the failure mode.
9. Run Quality Checks¶
Then, from the platform repo, run the cross-repo gates before publishing: