Skip to content

Python package reference

API reference for the cloud_fpga_orchestrator package, generated directly from the source with mkdocstrings. It stays in sync with the code and grows richer as docstrings are added.

Domain models

The core state types shared across the service.

API request/response models

The Pydantic models that define the REST contract.

RunRequest

Bases: BaseModel

Payload for POST /fpga/{id}/run.

Source code in _code/orchestrator/src/cloud_fpga_orchestrator/api/models.py 
13
14
15
16
17
18
19
20
21
22
23
class RunRequest(BaseModel):
    """Payload for POST /fpga/{id}/run."""

    op: int = Field(description="Wishbone opcode: 1 = write, 2 = read")
    address: int = Field(description="Wishbone byte address")
    data: list[int] = Field(
        default_factory=list, description="32-bit data words (write only)"
    )
    count: int = Field(
        default=1, description="number of 32-bit words to read (read only)"
    )

SessionResponse

Bases: BaseModel

Active session information for an FPGA.

Source code in _code/orchestrator/src/cloud_fpga_orchestrator/api/models.py 
31
32
33
34
35
36
37
class SessionResponse(BaseModel):
    """Active session information for an FPGA."""

    session_id: UUID
    fpga_id: int
    owner: str
    expires_at: datetime

JobResponse

Bases: BaseModel

Status and metadata for a single job.

Source code in _code/orchestrator/src/cloud_fpga_orchestrator/api/models.py 
40
41
42
43
44
45
46
47
48
class JobResponse(BaseModel):
    """Status and metadata for a single job."""

    job_id: UUID
    fpga_id: int
    type: JobType
    status: JobStatus
    created_at: datetime
    updated_at: datetime

FPGASummary

Bases: BaseModel

State snapshot for a single FPGA node.

Source code in _code/orchestrator/src/cloud_fpga_orchestrator/api/models.py 
51
52
53
54
55
56
57
class FPGASummary(BaseModel):
    """State snapshot for a single FPGA node."""

    fpga_id: int
    state: FPGAState
    session: SessionResponse | None
    current_job_id: UUID | None

SubmitResponse

Bases: BaseModel

Returned on a successful POST /fpga/{id}/submit (HTTP 202).

Source code in _code/orchestrator/src/cloud_fpga_orchestrator/api/models.py 
60
61
62
63
64
65
class SubmitResponse(BaseModel):
    """Returned on a successful POST /fpga/{id}/submit (HTTP 202)."""

    job_id: UUID
    fpga_id: int
    status: JobStatus = JobStatus.QUEUED

RunResponse

Bases: BaseModel

Returned on a successful POST /fpga/{id}/run (HTTP 202).

Source code in _code/orchestrator/src/cloud_fpga_orchestrator/api/models.py 
68
69
70
71
72
73
class RunResponse(BaseModel):
    """Returned on a successful POST /fpga/{id}/run (HTTP 202)."""

    job_id: UUID
    fpga_id: int
    status: JobStatus = JobStatus.QUEUED

RunResultResponse

Bases: BaseModel

Result data available once a run job completes.

Source code in _code/orchestrator/src/cloud_fpga_orchestrator/api/models.py 
76
77
78
79
80
class RunResultResponse(BaseModel):
    """Result data available once a run job completes."""

    ok: bool
    data: list[int]

HealthResponse

Bases: BaseModel

System health snapshot.

Source code in _code/orchestrator/src/cloud_fpga_orchestrator/api/models.py 
83
84
85
86
87
class HealthResponse(BaseModel):
    """System health snapshot."""

    status: str
    redis: bool

ErrorResponse

Bases: BaseModel

Standard error envelope returned on all non-2xx responses.

Source code in _code/orchestrator/src/cloud_fpga_orchestrator/api/models.py 
90
91
92
93
94
95
class ErrorResponse(BaseModel):
    """Standard error envelope returned on all non-2xx responses."""

    error: str = Field(description="Machine-readable error code")
    message: str = Field(description="Human-readable description")
    job_id: UUID | None = None

ExchangeRequest

Bases: BaseModel

Payload for POST /auth/github/exchange.

Source code in _code/orchestrator/src/cloud_fpga_orchestrator/api/models.py 
 98
 99
100
101
class ExchangeRequest(BaseModel):
    """Payload for POST /auth/github/exchange."""

    github_token: str = Field(description="GitHub access token from device flow or PAT")

ExchangeResponse

Bases: BaseModel

Returned on a successful GitHub token exchange.

Source code in _code/orchestrator/src/cloud_fpga_orchestrator/api/models.py 
104
105
106
107
108
109
110
class ExchangeResponse(BaseModel):
    """Returned on a successful GitHub token exchange."""

    api_key: str = Field(
        description="Opaque API key — pass as X-API-Key on all requests"
    )
    github_username: str

Application factory

lifespan async 

lifespan(app: FastAPI) -> AsyncGenerator[None, None]

Initialize resources on startup and clean up on shutdown.

Source code in _code/orchestrator/src/cloud_fpga_orchestrator/api/app.py 
16
17
18
19
20
21
@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
    """Initialize resources on startup and clean up on shutdown."""
    init_pool()
    yield
    await close_pool()

create_app 

create_app() -> FastAPI

Create and configure the FastAPI application.

Source code in _code/orchestrator/src/cloud_fpga_orchestrator/api/app.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
def create_app() -> FastAPI:
    """Create and configure the FastAPI application."""
    app = FastAPI(
        title="Cloud FPGA Orchestrator",
        description=(
            "REST API for submitting HDL designs and running jobs on remote"
            " FPGAs."
        ),
        version="0.1.0",
        lifespan=lifespan,
    )

    app.include_router(health_router)
    app.include_router(auth_router)
    app.include_router(fpga_router)
    app.include_router(jobs_router)
    app.include_router(session_router)

    return app