stroblmeandClaude Opus 5
60d7ec81c0
A wheel whose top-level module is `app` collides with anything else in a user's venv, so the package that is about to be published takes the name it is published under. Only the Python package moves; the repo, the Docker WORKDIR and the compose project keep theirs. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
92 lines
3.0 KiB
Python
92 lines
3.0 KiB
Python
"""Artifacts over HTTP: the one way bytes get in and out of the store.
|
|
|
|
A node on this host could reach the directory itself, but a node on a remote
|
|
worker cannot — and having one path rather than two is what keeps a flow's
|
|
code the same wherever it runs.
|
|
"""
|
|
|
|
from typing import Any
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Query, Request
|
|
from fastapi.responses import StreamingResponse
|
|
from jwt.exceptions import InvalidTokenError
|
|
from pydantic import BaseModel
|
|
from sqlmodel import Session
|
|
|
|
from fluksio.api.deps import user_from_token
|
|
from fluksio.core import security
|
|
from fluksio.core.db import engine
|
|
from fluksio.flow.artifacts import ArtifactStore
|
|
|
|
|
|
def artifact_caller(request: Request) -> str:
|
|
"""Who may move artifacts: a signed-in person, or an attached worker.
|
|
|
|
A worker's node stores its checkpoints through this endpoint, so its own
|
|
credential has to open it — and only it. The token is no use anywhere else
|
|
in the API, which is why this check is here rather than in the shared
|
|
dependency every other route uses.
|
|
"""
|
|
header = request.headers.get("Authorization", "")
|
|
token = header[7:] if header.lower().startswith("bearer ") else ""
|
|
if not token:
|
|
raise HTTPException(status_code=401, detail="Not authenticated")
|
|
try:
|
|
claims = security.decode_worker_token(token)
|
|
except InvalidTokenError:
|
|
pass
|
|
else:
|
|
return f"worker:{claims.get('sub')}"
|
|
with Session(engine) as session:
|
|
# With the request, so a credential that is scoped by route — a wall
|
|
# panel's — is judged against this one rather than waved through.
|
|
user = user_from_token(session, token, request)
|
|
if user is None:
|
|
raise HTTPException(status_code=401, detail="Not authenticated")
|
|
return user.email
|
|
|
|
|
|
router = APIRouter(
|
|
prefix="/artifacts", tags=["artifacts"], dependencies=[Depends(artifact_caller)]
|
|
)
|
|
|
|
|
|
class ArtifactRef(BaseModel):
|
|
digest: str
|
|
size: int
|
|
media_type: str
|
|
name: str = ""
|
|
|
|
|
|
def _store(request: Request) -> ArtifactStore:
|
|
store: ArtifactStore | None = getattr(request.app.state, "artifact_store", None)
|
|
if store is None:
|
|
raise HTTPException(status_code=503, detail="The artifact store is not ready")
|
|
return store
|
|
|
|
|
|
@router.put("", response_model=ArtifactRef)
|
|
async def put_artifact(
|
|
request: Request,
|
|
name: str = Query(default=""),
|
|
media_type: str = Query(default=""),
|
|
) -> Any:
|
|
"""Store the request body and answer with the reference to it."""
|
|
store = _store(request)
|
|
body = await request.body()
|
|
return store.put([body], name=name, media_type=media_type)
|
|
|
|
|
|
@router.get("/{digest}")
|
|
def get_artifact(digest: str, request: Request) -> Any:
|
|
"""Stream one artifact back."""
|
|
store = _store(request)
|
|
path = store.path(digest)
|
|
if path is None:
|
|
raise HTTPException(status_code=404, detail="No such artifact")
|
|
return StreamingResponse(
|
|
store.read(digest),
|
|
media_type="application/octet-stream",
|
|
headers={"Content-Length": str(path.stat().st_size)},
|
|
)
|