docs: improve openapi documentation

closes #6

app/web/dtos.py

@@ -0,0 +1,33 @@
+from typing import Any, Dict, Optional
+
+from pydantic import AnyHttpUrl, BaseModel, Field
+
+import app.shared.db.schemas as schemas
+
+
+class DetailResponse(BaseModel):
+    detail: str
+
+
+DEFAULT_RESPONSES: Dict[int | str, Dict[str, Any]] = {
+    401: {"model": DetailResponse, "description": "Not authenticated"}
+}
+
+
+class PostJobPayload(BaseModel):
+    url: AnyHttpUrl = Field(
+        description=(
+            "URL where the media file is available. This needs to be a direct link."
+        )
+    )
+
+    type: schemas.JobType = Field(description="Type of this job.")
+
+    # TODO: limit to locales selected by whisper.
+    language: Optional[str] = Field(
+        description=(
+            "Spoken language in the media file."
+            "While optional, this can improve output "
+            "by selecting a language-specific model. (applies to 'en')"
+        )
+    )

app/web/main.py

@@ -3,20 +3,24 @@ from typing import List, Optional
 from uuid import UUID
 
 from fastapi import APIRouter, Depends, FastAPI, HTTPException, Path
-from pydantic import AnyHttpUrl, BaseModel
 from sqlalchemy import or_
 from sqlalchemy.orm import Session
 
-import app.shared.db.schemas as schemas
 import app.shared.db.models as models
+import app.shared.db.schemas as schemas
 from app.shared.celery import get_celery_binding
 from app.shared.db.base import get_session
+from app.web.dtos import DEFAULT_RESPONSES, DetailResponse, PostJobPayload
 from app.web.security import authenticate_api_key
 
 app = FastAPI()
 celery = get_celery_binding()
 
-api_router = APIRouter(prefix="/api/v1", dependencies=[Depends(authenticate_api_key)])
+api_router = APIRouter(
+    prefix="/api/v1",
+    dependencies=[Depends(authenticate_api_key)],
+    responses={**DEFAULT_RESPONSES},
+)
 
 
 def queue_task(job: models.Job) -> None:
@@ -28,17 +32,11 @@ def queue_task(job: models.Job) -> None:
     transcribe.delay(job.id)
 
 
-@api_router.get("/")
+@api_router.get("/", response_model=None, status_code=204)
 def api_root() -> None:
     return None
 
 
-class PostJobPayload(BaseModel):
-    url: AnyHttpUrl
-    type: schemas.JobType
-    language: Optional[str]
-
-
 @api_router.post("/jobs", response_model=schemas.Job, status_code=201)
 def create_job(
     payload: PostJobPayload,
@@ -77,7 +75,11 @@ def get_transcripts(
     return query.all()
 
 
-@api_router.get("/jobs/{id}", response_model=schemas.Job)
+@api_router.get(
+    "/jobs/{id}",
+    response_model=schemas.Job,
+    responses={404: {"model": DetailResponse, "description": "Not authenticated"}},
+)
 def get_transcript(
     id: UUID = Path(), session: Session = Depends(get_session)
 ) -> Optional[models.Job]:
@@ -95,9 +97,6 @@ def get_artifacts_for_job(
         session.query(models.Artifact).filter(models.Artifact.job_id == str(id))
     ).all()
 
-    if not len(artifacts):
-        raise HTTPException(status_code=404)
-
     return artifacts
 
 

app/web/security.py

@@ -1,16 +1,16 @@
 from hmac import compare_digest
 
 from fastapi import Depends, HTTPException
-from fastapi.security import OAuth2PasswordBearer
+from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
 
 from app.shared.settings import settings
 
 
 def authenticate_api_key(
-    token: str = Depends(OAuth2PasswordBearer(tokenUrl="token")),
+    credentials: HTTPAuthorizationCredentials = Depends(HTTPBearer(auto_error=False)),
 ) -> None:
-    if not token:
-        raise HTTPException(status_code=422)
     # use compare_digest to counter timing attacks.
-    if not compare_digest(settings.API_SECRET, token):
+    if not credentials or not compare_digest(
+        settings.API_SECRET, credentials.credentials
+    ):
         raise HTTPException(status_code=401)