Files
svcforge/services/api/routes/instances.py
T
Nguyen Minh Phuc 50c2fe2a1e
ci / lint (push) Successful in 1m19s
ci / unit (push) Failing after 1m2s
ci / integration (push) Has been skipped
ci / types (push) Successful in 1m37s
ci / security (push) Failing after 38s
ci / dockerfile (push) Successful in 14s
ci / image (api) (push) Has been skipped
ci / image (reconciler) (push) Has been skipped
ci / image (worker) (push) Has been skipped
ci / bump (push) Has been skipped
svcforge: reference implementation
Complete working build of the system learn-python/ teaches.
164 tests, mypy --strict clean, domain coverage 99%.
2026-07-17 10:44:54 +00:00

219 lines
8.5 KiB
Python

"""The tenant-facing API.
Two rules run through every handler here:
* **AuthZ is the WHERE clause.** No handler ever compares `inst.team` to the caller's
team, because the repo never returns another team's row to compare. A wrong-team id is
a 404. 403 would confirm the id exists, which is the leak.
* **The instance and its task commit together.** A committed instance with no task is an
instance that never provisions and that nothing will ever retry.
"""
from __future__ import annotations
from datetime import UTC, datetime, timedelta
from typing import Any
from uuid import UUID, uuid4
from fastapi import APIRouter, Depends, HTTPException, Query, Response, status
from services.api.deps import (
CatalogDep,
InstanceRepoDep,
PoolDep,
TaskRepoDep,
TeamDep,
idempotency_key,
rate_limit,
)
from services.api.models import CreateInstanceRequest, ErrorBody, InstanceResponse
from svcforge_core.domain.models import CatalogEntry, Instance, TaskKind
from svcforge_core.domain.states import IllegalTransition, InstanceState, transition
# Declared on the router so every error shape lands in openapi.json under ErrorBody.
# The exception handler already renders this at runtime; without declaring it, generated
# clients see the contract for 2xx only and invent their own guess for the rest.
ERROR_RESPONSES: dict[int | str, dict[str, Any]] = {
401: {"model": ErrorBody, "description": "Missing or invalid credentials"},
404: {"model": ErrorBody, "description": "No such instance, or not this team's"},
409: {"model": ErrorBody, "description": "Instance is not in a state that allows this"},
422: {"model": ErrorBody, "description": "The body is well-formed but cannot be processed"},
}
router = APIRouter(prefix="/v1/instances", tags=["instances"], responses=ERROR_RESPONSES)
def release_name_for(team: str, service_type: str, instance_id: UUID) -> str:
"""The helm release name. Deterministic, and `unique` in the schema.
This is the idempotency anchor. A worker that dies after `helm install` but before it
marks the task done will retry, compute the same name, and `helm upgrade --install`
onto the same release instead of creating a second one. Derive it from anything that
is not already durable — a timestamp, a random suffix, the retry count — and a retry
provisions a duplicate.
Truncated to the uuid's first 8 chars to stay inside the 53-char limit helm imposes
on release names (Kubernetes label values, minus room for chart-generated suffixes).
"""
return f"{team}-{service_type}-{str(instance_id)[:8]}"
def namespace_for(team: str) -> str:
"""One namespace per tenant. The blast radius of a bad chart is one team."""
return f"tenant-{team}"
def _resolve(catalog: dict[str, CatalogEntry], service_type: str, size: str) -> CatalogEntry:
"""Look up service_type + size, or raise the right 4xx.
The two failures are different HTTP problems and the spec asks for different codes:
an unknown service_type is a resource that does not exist (404); an unknown size for a
real service_type is a body the server understood and cannot process (422).
"""
entry = catalog.get(service_type)
if entry is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail={
"code": "unknown_service_type",
"message": f"no such service_type: {service_type}",
},
)
if size not in entry.sizes:
raise HTTPException(
status_code=422, # starlette renamed the 422 constant; the number never moved
detail={
"code": "unknown_size",
"message": f"{service_type} has no size {size!r}; available: {sorted(entry.sizes)}",
},
)
return entry
@router.post(
"",
status_code=status.HTTP_202_ACCEPTED,
response_model=InstanceResponse,
dependencies=[Depends(rate_limit), Depends(idempotency_key)],
)
async def create_instance(
body: CreateInstanceRequest,
response: Response,
team: TeamDep,
pool: PoolDep,
instances: InstanceRepoDep,
tasks: TaskRepoDep,
catalog: CatalogDep,
) -> Instance:
"""Accept a provisioning request. 202, never 201.
Nothing is provisioned when this returns. The row exists and a task is queued; a
worker will do the work seconds or minutes from now. 201 Created would be a lie about
a resource that does not exist yet, and clients would stop polling.
"""
entry = _resolve(catalog, body.service_type, body.size)
instance_id = uuid4()
now = datetime.now(UTC)
inst = Instance(
id=instance_id,
team=team,
service_type=body.service_type,
size=body.size,
state=InstanceState.REQUESTED,
namespace=namespace_for(team),
release_name=release_name_for(team, body.service_type, instance_id),
# Pinned from the catalog AT CREATION TIME, not read from the catalog later.
# This column records what is actually deployed; bumping catalog.yaml must show up
# as drift the reconciler can see, not silently rewrite history.
chart_version=entry.chart_version,
expires_at=now + timedelta(days=body.ttl_days) if body.ttl_days is not None else None,
created_at=now,
updated_at=now,
)
# The transaction. Both writes go through THIS conn, or the atomicity is theatre.
async with pool.connection() as conn, conn.transaction():
created = await instances.create(conn, inst)
await tasks.enqueue(conn, created.id, TaskKind.PROVISION)
response.headers["Location"] = f"/v1/instances/{created.id}"
return created
@router.get("/{instance_id}", response_model=InstanceResponse)
async def get_instance(
instance_id: UUID,
team: TeamDep,
instances: InstanceRepoDep,
) -> Instance:
"""404 if the repo returns None. A wrong-team id is a 404, not a 403."""
inst = await instances.get(instance_id, team)
if inst is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail={"code": "not_found", "message": f"no such instance: {instance_id}"},
)
return inst
@router.get("", response_model=list[InstanceResponse])
async def list_instances(
team: TeamDep,
instances: InstanceRepoDep,
limit: int = Query(default=50, ge=1, le=200),
) -> list[Instance]:
"""The caller's instances, newest first. Bounded: no endpoint returns 'all rows'."""
return await instances.list(team, limit=limit)
@router.delete(
"/{instance_id}",
status_code=status.HTTP_202_ACCEPTED,
response_model=InstanceResponse,
dependencies=[Depends(rate_limit)],
)
async def delete_instance(
instance_id: UUID,
team: TeamDep,
instances: InstanceRepoDep,
tasks: TaskRepoDep,
) -> Instance:
"""state -> deleting, enqueue deprovision. 202: the helm uninstall has not happened yet.
Ordering note. `InstanceRepo.update_state` owns its own connection, so the CAS and the
enqueue cannot share one transaction without reaching around the repo. Given two
statements, the order is chosen for its failure mode: CAS first, enqueue second. A
crash in between leaves an instance in `deleting` with no task, which the reconciler's
sweep re-enqueues. The other order leaves a deprovision task pointing at a `ready`
instance, and a worker would tear down a live service nobody asked to delete.
"""
inst = await instances.get(instance_id, team)
if inst is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail={"code": "not_found", "message": f"no such instance: {instance_id}"},
)
try:
target = transition(inst.state, InstanceState.DELETING)
except IllegalTransition as exc:
# Already deleting or already deleted. Not an error the tenant can fix.
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail={
"code": "illegal_transition",
"message": f"instance is {inst.state}; delete is not a legal transition",
},
) from exc
if not await instances.update_state(inst.id, expect=inst.state, to=target):
# Someone moved the row between the read and the CAS.
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail={"code": "conflict", "message": "instance changed concurrently; retry"},
)
await tasks.enqueue_standalone(inst.id, TaskKind.DEPROVISION)
return inst.model_copy(update={"state": target})