"""The FastAPI application behind ``vibe-sketch serve``.
The API (``/api``): version and liveness, the document JSON schema and
validation, the revisioned document store with import, report, render and
export (:mod:`vibesketch.server.documents`), and the input writers. Each
route mirrors an SDK call and a CLI command (AGENTS.md, ground rule 4).
OpenAPI is served at ``/openapi.json``, so a client can read the contract
instead of guessing it. Everything else is the browser shell, served from
its built bundle.
"""
from __future__ import annotations
import os
from collections.abc import Iterable
from pathlib import Path
from typing import Annotated, Any, Self
from fastapi import Body, FastAPI, HTTPException
from fastapi.responses import HTMLResponse
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel, ConfigDict, model_validator
from vibesketch import __version__
from vibesketch.codenames import codename_for_current, version_label
from vibesketch.document import Document, JobSpec, document_json_schema
from vibesketch.document.io import DocumentError, from_data
from vibesketch.server.store import Store
__all__ = ["create_app", "web_bundle"]
#: Shown at ``/`` when the browser shell has not been built.
_NO_BUNDLE = """<!doctype html><html lang="en"><head><meta charset="utf-8">
<title>vibe-sketcher</title></head><body style="font-family: sans-serif; margin: 3em">
<h1>vibe-sketcher server</h1>
<p>The API is running (<a href="/docs">interactive API documentation</a>), but the
browser shell has not been built. From the checkout, run:</p>
<pre>cd web && npm ci && npm run build</pre>
<p>or reinstall with <code>./scripts/install.sh --force</code>, which builds it when
Node.js is available. Then reload this page.</p></body></html>"""
[docs]
def web_bundle() -> Path | None:
"""The built browser shell: ``VIBESKETCH_WEB_DIST``, the package, or the checkout."""
configured = os.environ.get("VIBESKETCH_WEB_DIST")
candidates = [Path(configured)] if configured else []
here = Path(__file__).resolve().parent
candidates += [here / "static", here.parents[2] / "web" / "dist"]
for candidate in candidates:
if (candidate / "index.html").is_file():
return candidate
return None
class ValidationReport(BaseModel):
"""Whether a document is valid, and every problem when it is not."""
valid: bool
errors: list[str]
class WriterInfo(BaseModel):
name: str
description: str
tasks: list[str]
fields: list[str]
class InputRequest(BaseModel):
"""Which input to write: a job of the document, or a species and a writer."""
model_config = ConfigDict(extra="forbid")
document: Document
job: str | None = None
species: str | None = None
program: str | None = None
spec: JobSpec | None = None
#: Embed the species in 3D first when it has no conformer (not stored).
embed: bool = False
@model_validator(mode="after")
def _job_or_species_and_program(self) -> Self:
if self.job is not None:
if self.species is not None or self.program is not None or self.spec is not None:
raise ValueError("give a job, or a species and a program, not both")
elif self.species is None or self.program is None:
raise ValueError("give a job, or a species and a program")
return self
class InputFileOut(BaseModel):
name: str
media_type: str
content: str
class InputResponse(BaseModel):
program: str
species: str
files: list[InputFileOut]
[docs]
def create_app(
store: Store | None = None,
*,
web_root: Path | None = None,
hosts: Iterable[str] | None = None,
max_body: int | None = None,
) -> FastAPI:
"""Build the application. One instance per server process.
``store`` holds the documents (a fresh one by default); ``web_root`` is
the built browser shell (found with :func:`web_bundle` by default).
``hosts`` are the only names the server answers for, and the only
origins it takes requests from (:mod:`vibesketch.server.guard`); a
loopback server passes :data:`~vibesketch.server.guard.LOOPBACK`.
Request bodies are bounded by ``max_body`` bytes, by default
:data:`vibesketch.limits.MAX_REQUEST_BYTES`.
"""
from vibesketch.limits import MAX_REQUEST_BYTES
from vibesketch.server.documents import make_router
from vibesketch.server.guard import Guard
app = FastAPI(
title="vibe-sketcher",
version=__version__,
description="2D chemical structure and reaction sketcher.",
)
limit = MAX_REQUEST_BYTES if max_body is None else max_body
app.add_middleware(Guard, hosts=hosts, max_body=limit)
app.state.store = store if store is not None else Store()
app.include_router(make_router(app.state.store))
@app.get("/api/version", tags=["meta"])
def version() -> dict[str, str | None]:
"""The running version, its codename when one is registered, and the
label the user-facing surfaces print."""
return {
"version": __version__,
"codename": codename_for_current(),
"label": version_label(),
}
@app.get("/health", tags=["meta"])
def health() -> dict[str, str]:
"""Liveness probe for the desktop shell and for deployments."""
return {"status": "ok"}
@app.get("/api/schema/document", tags=["documents"])
def document_schema() -> dict[str, Any]:
"""The JSON schema (draft 2020-12) of a version-1 ``.vsk`` document."""
return document_json_schema()
@app.post("/api/documents/validate", tags=["documents"])
def validate_document(document: Annotated[Any, Body()]) -> ValidationReport:
"""Check any JSON value as a document; never fails, reports instead."""
try:
from_data(document)
except DocumentError as error:
return ValidationReport(valid=False, errors=error.errors)
return ValidationReport(valid=True, errors=[])
@app.get("/api/writers", tags=["inputs"])
def list_writers() -> list[WriterInfo]:
"""The input writers, the tasks each can express and its fields."""
from vibesketch.inputs import writers
return [
WriterInfo(
name=writer.name,
description=writer.description,
tasks=sorted(writer.tasks),
fields=sorted(writer.fields),
)
for writer in writers()
]
@app.post(
"/api/inputs",
tags=["inputs"],
responses={
404: {"description": "The document has no such job or species."},
422: {"description": "The writer refused, or the request is invalid."},
},
)
def write_input_route(request: InputRequest) -> InputResponse:
"""Write the input for a job of the document, or for a species and a writer.
A writer that cannot express the request answers 422 with
``{"detail": {"program": ..., "reasons": [...]}}`` and writes nothing.
"""
from vibesketch.inputs import (
UnknownReference,
UnknownWriter,
WriterRefusal,
write_input,
)
document = request.document
if request.embed:
from vibesketch.document.embed import UnknownSpecies, embed
target = request.species
if request.job is not None:
target = next((j.species for j in document.jobs if j.id == request.job), None)
if target is not None:
try:
document, _ = embed(document, [target])
except UnknownSpecies as error:
raise HTTPException(404, str(error)) from None
try:
bundle = write_input(
document,
job=request.job,
species=request.species,
program=request.program,
spec=request.spec,
)
except UnknownReference as error:
raise HTTPException(404, str(error)) from None
except UnknownWriter as error:
raise HTTPException(422, {"program": error.name, "reasons": [str(error)]}) from None
except WriterRefusal as error:
detail = {"program": error.program, "reasons": list(error.reasons)}
raise HTTPException(422, detail) from None
return InputResponse(
program=bundle.program,
species=bundle.species,
files=[
InputFileOut(name=f.name, media_type=f.media_type, content=f.content)
for f in bundle.files
],
)
bundle = web_root if web_root is not None else web_bundle()
if bundle is not None:
app.mount("/", StaticFiles(directory=bundle, html=True), name="web")
else:
@app.get("/", include_in_schema=False)
def no_bundle() -> HTMLResponse:
return HTMLResponse(_NO_BUNDLE)
return app