main.py

"""
A todo API on FastAPI and managed Postgres.

The point of this template is the wiring: how an app on velixir finds its database, creates
its tables, stays up while the database is out of reach, and reports health honestly. The
todos themselves are the least interesting part.

Deploy it from the velixir gallery and a managed Postgres is created and bound for you, so
DATABASE_URL is in the environment on the first boot. Nothing here hardcodes a host, so the
same code runs locally against a container and in production against the managed instance.

The Procfile starts it with uvicorn on 0.0.0.0:$PORT. --forwarded-allow-ips "*" trusts the
X-Forwarded-For and X-Forwarded-Proto headers set by velixir's edge, so the app sees the
visitor's address and scheme. That is safe because nothing else can reach the container.
"""

import html
import os
from contextlib import asynccontextmanager
from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException, Path, Query, Request, status
from fastapi.responses import HTMLResponse, JSONResponse, PlainTextResponse
from sqlalchemy import func, select, text
from sqlalchemy.exc import OperationalError
from sqlalchemy.orm import Session

import db
from models import Todo, TodoCreate, TodoOut, TodoUpdate

SETUP_MESSAGE = (
    "No database is bound to this app yet. On velixir: create a managed Postgres, bind it to "
    "this app as DATABASE_URL, then redeploy. Deploying this template from the velixir "
    "gallery does all of that for you."
)


@asynccontextmanager
async def lifespan(_app):
    if db.engine is None:
        # Setup mode. Rather than exiting (a crash loop reads as "the platform is broken"),
        # serve a page that says exactly what is missing and keep the health check green.
        print(
            "DATABASE_URL is not set, so this app is serving its setup page.\n"
            "On velixir: create a managed Postgres, bind it to this app as DATABASE_URL, then redeploy.\n"
            "Locally: DATABASE_URL=postgres://user:pass@localhost:5432/todos uvicorn main:app --port 8080",
            flush=True,
        )
    else:
        # In the background, so the server is already answering (and explaining the wait)
        # while this retries.
        db.connect_in_background()
    yield


app = FastAPI(
    title="Todos on velixir",
    summary="A small CRUD API on FastAPI and managed Postgres.",
    version="1.0.0",
    lifespan=lifespan,
)


# ─── Database access ────────────────────────────────────────────────────────


def get_session():
    """A database session for one request, or a 503 that says why there cannot be one."""
    if db.engine is None:
        raise HTTPException(status.HTTP_503_SERVICE_UNAVAILABLE, detail=SETUP_MESSAGE)
    if not db.state.ready:
        # The driver's error stays in the log: it can name the database's internal address and
        # user, which is nobody else's business.
        raise HTTPException(
            status.HTTP_503_SERVICE_UNAVAILABLE,
            detail="Waiting for the database. The app log has the details.",
            headers={"Retry-After": "5"},
        )
    with db.Session() as session:
        yield session


DbSession = Annotated[Session, Depends(get_session)]
TodoId = Annotated[int, Path(ge=1, le=2**63 - 1, description="The todo's id.")]


# The database went away after start-up (a failover, a restart). Say so with a 503 rather
# than a bare 500. Requests succeed again on their own once it is back.
@app.exception_handler(OperationalError)
async def database_unreachable(_request: Request, exc: OperationalError):
    print(f"database unreachable: {exc.orig or exc}", flush=True)
    return JSONResponse(
        {"detail": "The database is unreachable right now. Try again in a few seconds."},
        status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
        headers={"Retry-After": "5"},
    )


# ─── API ────────────────────────────────────────────────────────────────────

# Plain `def` handlers, not `async def`: SQLAlchemy's session is synchronous, and FastAPI runs
# sync handlers in a thread pool, so a slow query never blocks other requests.


@app.get("/api/todos", response_model=list[TodoOut], tags=["todos"])
def list_todos(
    session: DbSession,
    done: Annotated[bool | None, Query(description="Only todos that are, or are not, done.")] = None,
):
    """The newest 100 todos."""
    query = select(Todo).order_by(Todo.created_at.desc(), Todo.id.desc()).limit(100)
    if done is not None:
        query = query.where(Todo.done == done)
    return session.scalars(query).all()


@app.post("/api/todos", response_model=TodoOut, status_code=status.HTTP_201_CREATED, tags=["todos"])
def create_todo(body: TodoCreate, session: DbSession):
    """Add a todo. It starts out not done."""
    todo = Todo(title=body.title)
    session.add(todo)
    session.commit()
    session.refresh(todo)
    return todo


@app.get("/api/todos/{todo_id}", response_model=TodoOut, tags=["todos"])
def get_todo(todo_id: TodoId, session: DbSession):
    """One todo."""
    return find(session, todo_id)


@app.patch("/api/todos/{todo_id}", response_model=TodoOut, tags=["todos"])
def update_todo(todo_id: TodoId, body: TodoUpdate, session: DbSession):
    """Rename a todo, tick it off, or both. Fields left out (or null) stay as they are."""
    todo = find(session, todo_id)
    for field, value in body.model_dump(exclude_unset=True, exclude_none=True).items():
        setattr(todo, field, value)
    session.commit()
    session.refresh(todo)
    return todo


@app.delete("/api/todos/{todo_id}", status_code=status.HTTP_204_NO_CONTENT, tags=["todos"])
def delete_todo(todo_id: TodoId, session: DbSession):
    """Delete a todo for good."""
    session.delete(find(session, todo_id))
    session.commit()


def find(session, todo_id):
    todo = session.get(Todo, todo_id)
    if todo is None:
        raise HTTPException(status.HTTP_404_NOT_FOUND, detail="No todo with that id.")
    return todo


# ─── Health ─────────────────────────────────────────────────────────────────


@app.get("/healthz", include_in_schema=False)
def healthz():
    """Liveness and readiness in one: this app is useless without its database, so a failed
    query takes it out of rotation rather than letting it serve errors."""
    if db.engine is None:
        return PlainTextResponse("ok (setup mode: no database bound)")
    if not db.state.ready:
        return PlainTextResponse("waiting for the database", status_code=503)
    try:
        with db.engine.connect() as conn:
            conn.execute(text("SELECT 1"))
    except Exception as exc:
        print(f"health check failed: {exc}", flush=True)
        return PlainTextResponse("database unreachable", status_code=503)
    return PlainTextResponse("ok")


# ─── Pages ──────────────────────────────────────────────────────────────────


@app.get("/", response_class=HTMLResponse, include_in_schema=False)
def home(request: Request):
    if db.engine is None:
        return HTMLResponse(page("Connect a database", SETUP_BODY))
    if not db.state.ready:
        return HTMLResponse(page("Waiting for the database", WAITING_BODY), 503)
    try:
        with db.Session() as session:
            count = session.scalar(select(func.count()).select_from(Todo))
    except OperationalError as exc:
        print(f"database unreachable: {exc.orig or exc}", flush=True)
        return HTMLResponse(page("Waiting for the database", WAITING_BODY), 503)
    return HTMLResponse(page("Todos on velixir", home_body(public_url(request), count)))


def public_url(request):
    """Where this app lives: PUBLIC_URL when set (a custom domain), else its velixir address,
    else whatever the request says (local development)."""
    if os.environ.get("PUBLIC_URL"):
        return os.environ["PUBLIC_URL"].rstrip("/")
    if os.environ.get("VELIXIR_APP_SLUG"):
        return f"https://{os.environ['VELIXIR_APP_SLUG']}.velixir.run"
    return str(request.base_url).rstrip("/")


def home_body(base, count):
    base = html.escape(base)
    stored = "1 todo" if count == 1 else f"{count} todos"
    return f"""
  <h1>Todos</h1>
  <p class="sub">A FastAPI app on managed Postgres, reached through <code>DATABASE_URL</code>.
    {stored} stored.</p>

  <ul class="links">
    <li><a href="/docs">Interactive docs</a><span>Try every endpoint from the browser.</span></li>
    <li><a href="/redoc">API reference</a><span>The same API as a page to read.</span></li>
    <li><a href="/openapi.json">OpenAPI schema</a><span>For generating a client.</span></li>
  </ul>

  <h2>From a terminal</h2>
  <pre>curl -X POST {base}/api/todos \\
  -H 'content-type: application/json' \\
  -d '{{"title": "Ship it"}}'

curl {base}/api/todos</pre>"""


SETUP_BODY = """
  <h1>Connect a database</h1>
  <p class="sub">This app keeps its todos in Postgres, and no database is bound to it yet.</p>
  <ol class="steps">
    <li><strong>Create a managed Postgres</strong> from the Databases page, or run
      <code>velixir db create</code>.</li>
    <li><strong>Bind it to this app</strong> as <code>DATABASE_URL</code>: on the database's page
      under Bound apps, or <code>velixir db bind &lt;database-id&gt; --app &lt;app-id&gt;</code>.</li>
    <li><strong>Redeploy</strong> with the Redeploy button on the live release (Deploys tab).
      The table is created on the next boot.</li>
  </ol>
  <p class="note">Deploying this template from the velixir gallery does all three for you.
    Until then the API answers 503, and <a href="/docs">the docs</a> already work.</p>"""


# No error detail here: the driver's message can name the database's internal address and user,
# and this page is public. db.py prints it to the app log instead.
WAITING_BODY = """
  <h1>Waiting for the database</h1>
  <p class="sub">The app is running and retrying the connection every few seconds. The app log says why it cannot connect.</p>
  <p class="note">If the database password was rotated, redeploy the app to pick up the new one.</p>"""


def page(title, body):
    return f"""<!doctype html>
<html lang="en">
<meta charset="utf-8">
<meta name="viewport" content="width=device-width,initial-scale=1">
<title>{html.escape(title)}</title>
<style>
  :root {{ color-scheme: dark; }}
  * {{ box-sizing: border-box; }}
  body {{ margin:0; min-height:100vh; background:#0b0b10; color:#e8e8ef;
         font:15px/1.6 ui-sans-serif,system-ui,-apple-system,Segoe UI,sans-serif; }}
  .wrap {{ max-width:38rem; margin:0 auto; padding:3rem 1.25rem 4rem; }}
  h1 {{ font-size:1.5rem; margin:0 0 .25rem; letter-spacing:-.02em; }}
  h2 {{ font-size:1rem; margin:2rem 0 .75rem; }}
  .sub {{ color:#7a7a8c; font-size:.8125rem; margin:0 0 2rem; }}
  a {{ color:#a5b4fc; }}
  code {{ background:#16161f; border:1px solid #26263a; border-radius:5px;
         padding:.1rem .35rem; color:#c7d2fe; font-size:.8125rem; overflow-wrap:anywhere; }}
  pre {{ background:#13131c; border:1px solid #26263a; border-radius:10px; padding:1rem;
        color:#c7d2fe; font-size:.8125rem; overflow-x:auto; margin:0; }}
  ul.links {{ list-style:none; margin:0; padding:0; }}
  ul.links li {{ display:flex; flex-wrap:wrap; align-items:baseline; gap:.25rem .75rem;
                border:1px solid #1f1f2e; border-radius:10px; padding:.875rem 1rem; margin-bottom:.5rem; }}
  ul.links a {{ font-weight:600; text-decoration:none; }}
  ul.links span {{ color:#7a7a8c; font-size:.8125rem; }}
  ol.steps {{ padding-left:1.25rem; margin:0 0 1.5rem; }}
  ol.steps li {{ margin-bottom:.9rem; }}
  .note {{ color:#7a7a8c; font-size:.8125rem; }}
  footer {{ color:#4b4b5c; font-size:.75rem; margin-top:2.5rem; text-align:center; }}
  footer a {{ color:#7a7a8c; }}
</style>
<div class="wrap">
  {body}
  <footer>
    <a href="/docs">API docs</a> &middot; running on <a href="https://velixir.net">velixir</a>
  </footer>
</div>
</html>"""