Spaces:
Running
Running
File size: 3,904 Bytes
3493993 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 | from __future__ import annotations
from typing import Annotated
from fastapi import APIRouter, Header, HTTPException, Query, Request, status
from app.generation.schemas.requests import (
GenerationJobView,
GenerationProviderView,
GenerationRequestCreate,
GenerationRequestView,
)
from app.generation.model_registry import GenerationModelView
router = APIRouter(prefix="/v1/generation", tags=["generation"])
def _generation(request: Request):
service = request.app.state.container.generation
service.ensure_ready()
return service
def _identity(request: Request) -> tuple[str, str]:
context = request.state.auth
if not context.workspace_id or not context.user_id:
# The credential itself is never a workspace. API-key middleware
# resolves the authoritative membership before this route runs.
raise HTTPException(status_code=403, detail="No active workspace membership.")
return context.workspace_id, context.user_id
@router.get("/providers", response_model=list[GenerationProviderView])
async def list_providers(request: Request) -> list[GenerationProviderView]:
return _generation(request).list_providers()
@router.get(
"/providers/{provider}", response_model=GenerationProviderView
)
async def get_provider(request: Request, provider: str) -> GenerationProviderView:
return _generation(request).get_provider(provider)
@router.get("/models", response_model=list[GenerationModelView])
async def list_models(
request: Request,
provider: str | None = Query(default=None, min_length=1, max_length=64),
) -> list[GenerationModelView]:
return _generation(request).list_models(provider=provider)
@router.get("/providers/{provider}/models/{model_id}", response_model=GenerationModelView)
async def get_model(
request: Request, provider: str, model_id: str
) -> GenerationModelView:
return _generation(request).get_model(provider, model_id)
@router.post(
"/requests",
response_model=GenerationRequestView,
status_code=status.HTTP_202_ACCEPTED,
)
async def create_request(
request: Request,
payload: GenerationRequestCreate,
idempotency_key: Annotated[
str, Header(alias="Idempotency-Key", min_length=8, max_length=255)
],
) -> GenerationRequestView:
workspace_id, user_id = _identity(request)
return await _generation(request).create(
workspace_id=workspace_id,
user_id=user_id,
payload=payload,
idempotency_key=idempotency_key,
)
@router.get("/requests", response_model=list[GenerationRequestView])
async def list_requests(
request: Request,
offset: int = Query(default=0, ge=0),
limit: int = Query(default=100, ge=1, le=500),
) -> list[GenerationRequestView]:
workspace_id, user_id = _identity(request)
return await _generation(request).list_requests(
workspace_id, user_id, offset=offset, limit=limit
)
@router.get("/requests/{generation_request_id}", response_model=GenerationRequestView)
async def get_request(
request: Request, generation_request_id: str
) -> GenerationRequestView:
workspace_id, user_id = _identity(request)
return await _generation(request).get_request(
workspace_id, user_id, generation_request_id
)
@router.get("/jobs/{generation_job_id}", response_model=GenerationJobView)
async def get_job(request: Request, generation_job_id: str) -> GenerationJobView:
workspace_id, user_id = _identity(request)
return await _generation(request).get_job(
workspace_id, user_id, generation_job_id
)
@router.post(
"/jobs/{generation_job_id}/cancel", response_model=GenerationJobView
)
async def cancel_job(request: Request, generation_job_id: str) -> GenerationJobView:
workspace_id, user_id = _identity(request)
return await _generation(request).cancel(
workspace_id, user_id, generation_job_id
)
|