gyoza.server#

gyoza server runtime — FastAPI application, repositories, and scheduler.

gyoza.server.start(host=None, port=None, reload=None, workers=None, log_level=None)[source]#

Start the gyoza server.

Parameters passed explicitly take precedence over environment variables, which in turn take precedence over the built-in defaults. This lets the gyoza server start CLI pass Typer-parsed values directly.

Parameters:

host (str | None) – Host to bind to.
port (int | None) – Port to listen on.
reload (bool | None) – Enable uvicorn auto-reload (development only).
workers (int | None) – Number of uvicorn worker processes.
log_level (str | None) – Logging level string.

Return type:

None

API#

Pydantic models for API request/response validation.

class gyoza.server.api.models.InputSpecModel(*, type, required=True, default=None, description=None, example=None)[source]#

Bases: BaseModel

Input specification model.

Parameters:

type (str)
required (bool)
default (Any | None)
description (str | None)
example (Any | None)

type: str#

required: bool#

default: Any | None#

description: str | None#

example: Any | None#

model_config: ClassVar[ConfigDict] = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class gyoza.server.api.models.OutputSpecModel(*, type, description=None, example=None)[source]#

Bases: BaseModel

Output specification model.

Parameters:

type (str)
description (str | None)
example (Any | None)

type: str#

description: str | None#

example: Any | None#

model_config: ClassVar[ConfigDict] = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class gyoza.server.api.models.ConstraintsModel(*, ram_mb=None, vram_mb=None, cpu=None, gpu_id=None, tags=<factory>)[source]#

Bases: BaseModel

Hardware constraints model. An unset field reserves nothing.

Parameters:

ram_mb (int | None)
vram_mb (int | None)
cpu (int | None)
gpu_id (int | None)
tags (list[str])

ram_mb: int | None#

vram_mb: int | None#

cpu: int | None#

gpu_id: int | None#

tags: list[str]#

model_config: ClassVar[ConfigDict] = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class gyoza.server.api.models.RetryPolicyModel(*, max_attempts=3)[source]#

Bases: BaseModel

Retry policy model.

Parameters:: max_attempts (int)

max_attempts: int#

model_config: ClassVar[ConfigDict] = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class gyoza.server.api.models.EventDeliveryModel(*, topic='', attributes=None)[source]#

Bases: BaseModel

Event delivery configuration model.

Parameters:

topic (str)
attributes (dict[str, str] | None)

topic: str#

attributes: dict[str, str] | None#

model_config: ClassVar[ConfigDict] = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class gyoza.server.api.models.OpDefinitionCreate(*, id, version, image, description=None, input_specs=None, output_specs=None, constraints=None, retry_policy=None, event_delivery=None)[source]#

Bases: BaseModel

Request model for creating/updating OpDefinition.

Parameters:

id (str)
version (str)
image (str)
description (str | None)
input_specs (dict[str, InputSpecModel] | None)
output_specs (dict[str, OutputSpecModel] | None)
constraints (ConstraintsModel | None)
retry_policy (RetryPolicyModel | None)
event_delivery (EventDeliveryModel | None)

id: str#

version: str#

image: str#

description: str | None#

input_specs: dict[str, InputSpecModel] | None#

output_specs: dict[str, OutputSpecModel] | None#

constraints: ConstraintsModel | None#

retry_policy: RetryPolicyModel | None#

event_delivery: EventDeliveryModel | None#

model_config: ClassVar[ConfigDict] = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class gyoza.server.api.models.OpDefinitionResponse(*, id, version, image, description=None, input_specs, output_specs, constraints, retry_policy, event_delivery, createdAt, updatedAt)[source]#

Bases: BaseModel

Response model for OpDefinition.

Parameters:

id (str)
version (str)
image (str)
description (str | None)
input_specs (dict[str, Any])
output_specs (dict[str, Any])
constraints (dict[str, Any])
retry_policy (dict[str, Any])
event_delivery (dict[str, Any])
createdAt (str)
updatedAt (str)

id: str#

version: str#

image: str#

description: str | None#

input_specs: dict[str, Any]#

output_specs: dict[str, Any]#

constraints: dict[str, Any]#

retry_policy: dict[str, Any]#

event_delivery: dict[str, Any]#

createdAt: str#

updatedAt: str#

model_config: ClassVar[ConfigDict] = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class gyoza.server.api.models.OpDefinitionDeleteResponse(*, id, version=None, deleted)[source]#

Bases: BaseModel

Response model for an OpDefinition deletion.

Parameters:

id (str)
version (str | None)
deleted (int)

id: str#

version: str | None#

deleted: int#

model_config: ClassVar[ConfigDict] = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class gyoza.server.api.models.LogicalSubmission(*, inputs, outputs)[source]#

Bases: BaseModel

Stable resources/parameters supplied by the caller, excluding signed URLs.

Parameters:

inputs (dict[str, JsonValue])
outputs (dict[str, JsonValue])

model_config: ClassVar[ConfigDict] = {'extra': 'forbid'}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

inputs: dict[str, JsonValue]#

outputs: dict[str, JsonValue]#

finite_json()[source]#

class gyoza.server.api.models.OpRunCreateFromDefinition(*, idempotency_key=None, logical_payload=None, inputs=<factory>, outputs=<factory>, priority=0, event_delivery=None)[source]#

Bases: BaseModel

Request model for creating OpRun from OpDefinition.

This is used when creating a run from a template (OpDefinition). For creating runs from scratch, use OpRunCreate (to be implemented).

Parameters:

idempotency_key (str | None)
logical_payload (LogicalSubmission | None)
inputs (dict[str, Any])
outputs (dict[str, Any])
priority (int)
event_delivery (EventDeliveryModel | None)

idempotency_key: str | None#

logical_payload: LogicalSubmission | None#

paired_idempotency()[source]#

inputs: dict[str, Any]#

outputs: dict[str, Any]#

priority: int#

event_delivery: EventDeliveryModel | None#

model_config: ClassVar[ConfigDict] = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class gyoza.server.api.models.OpRunResponse(*, id, state, priority, progress, current_attempt, image, inputs, outputs, events, execution_summary, constraints, reservation, retry_policy, event_delivery, op_definition, op_definition_version=None, idempotency_key=None, createdAt, updatedAt)[source]#

Bases: BaseModel

Response model for OpRun.

Parameters:

id (str)
state (str)
priority (int)
progress (int)
current_attempt (int)
image (str)
inputs (dict[str, Any])
outputs (dict[str, Any])
events (list[dict[str, Any]])
execution_summary (dict[str, Any])
constraints (dict[str, Any])
reservation (dict[str, int])
retry_policy (dict[str, Any])
event_delivery (dict[str, Any])
op_definition (str | None)
op_definition_version (str | None)
idempotency_key (str | None)
createdAt (str)
updatedAt (str)

id: str#

state: str#

priority: int#

progress: int#

current_attempt: int#

image: str#

inputs: dict[str, Any]#

outputs: dict[str, Any]#

events: list[dict[str, Any]]#

execution_summary: dict[str, Any]#

constraints: dict[str, Any]#

reservation: dict[str, int]#

retry_policy: dict[str, Any]#

event_delivery: dict[str, Any]#

op_definition: str | None#

op_definition_version: str | None#

idempotency_key: str | None#

createdAt: str#

updatedAt: str#

model_config: ClassVar[ConfigDict] = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class gyoza.server.api.models.OpRunListItem(*, id, state, priority, progress, current_attempt, image, inputs=None, outputs=None, events=None, execution_summary, constraints, retry_policy, event_delivery, op_definition, createdAt, updatedAt)[source]#

Bases: BaseModel

OpRun as returned by the list endpoint.

inputs, outputs and events can be arbitrarily large (GeoJSON inputs, one event per progress line), so the listing omits them unless explicitly requested via ?expand=.

Parameters:

id (str)
state (str)
priority (int)
progress (int)
current_attempt (int)
image (str)
inputs (dict[str, Any] | None)
outputs (dict[str, Any] | None)
events (list[dict[str, Any]] | None)
execution_summary (dict[str, Any])
constraints (dict[str, Any])
retry_policy (dict[str, Any])
event_delivery (dict[str, Any])
op_definition (str | None)
createdAt (str)
updatedAt (str)

id: str#

state: str#

priority: int#

progress: int#

current_attempt: int#

image: str#

inputs: dict[str, Any] | None#

outputs: dict[str, Any] | None#

events: list[dict[str, Any]] | None#

execution_summary: dict[str, Any]#

constraints: dict[str, Any]#

retry_policy: dict[str, Any]#

event_delivery: dict[str, Any]#

op_definition: str | None#

createdAt: str#

updatedAt: str#

model_config: ClassVar[ConfigDict] = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class gyoza.server.api.models.OpRunListResponse(*, object='list', data, has_more, url='/runs')[source]#

Bases: BaseModel

Stripe-style list response with cursor-based pagination.

Parameters:

object (str)
data (list[OpRunListItem])
has_more (bool)
url (str)

object: str#

data: list[OpRunListItem]#

has_more: bool#

url: str#

model_config: ClassVar[ConfigDict] = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class gyoza.server.api.models.RunStatsWindow(*, start, end, bucket_seconds, buckets)[source]#

Bases: BaseModel

The resolved time window a stats response covers.

Parameters:

start (int)
end (int)
bucket_seconds (int)
buckets (int)

start: int#

end: int#

bucket_seconds: int#

buckets: int#

model_config: ClassVar[ConfigDict] = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class gyoza.server.api.models.RunStatsTotals(*, all, by_state=<factory>)[source]#

Bases: BaseModel

Run counts over the whole window.

Parameters:

all (int)
by_state (dict[str, int])

all: int#

by_state: dict[str, int]#

model_config: ClassVar[ConfigDict] = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class gyoza.server.api.models.RunStatsDefinition(*, op_definition, count, buckets)[source]#

Bases: BaseModel

Per-definition activity within the window.

Parameters:

op_definition (str)
count (int)
buckets (list[int])

op_definition: str#

count: int#

buckets: list[int]#

model_config: ClassVar[ConfigDict] = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class gyoza.server.api.models.RunStatsResponse(*, object='run_stats', window, totals, definitions)[source]#

Bases: BaseModel

Aggregated run activity, computed server-side.

Lets a dashboard render totals and per-definition sparklines from one small response instead of paginating every run in the window.

Parameters:

object (str)
window (RunStatsWindow)
totals (RunStatsTotals)
definitions (list[RunStatsDefinition])

object: str#

window: RunStatsWindow#

totals: RunStatsTotals#

definitions: list[RunStatsDefinition]#

model_config: ClassVar[ConfigDict] = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class gyoza.server.api.models.OpRunCreate(*, image, priority=0, inputs=<factory>, outputs=<factory>, constraints=None, retry_policy=None, event_delivery=None)[source]#

Bases: BaseModel

Request model for creating OpRun directly (ad-hoc).

This is used when creating a run from scratch without a template. For creating runs from OpDefinition, use OpRunCreateFromDefinition.

Parameters:

image (str)
priority (int)
inputs (dict[str, Any])
outputs (dict[str, Any])
constraints (ConstraintsModel | None)
retry_policy (RetryPolicyModel | None)
event_delivery (EventDeliveryModel | None)

image: str#

priority: int#

inputs: dict[str, Any]#

outputs: dict[str, Any]#

constraints: ConstraintsModel | None#

retry_policy: RetryPolicyModel | None#

event_delivery: EventDeliveryModel | None#

model_config: ClassVar[ConfigDict] = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class gyoza.server.api.models.OpRunUpdate(*, state=None, outputs=None, priority=None)[source]#

Bases: BaseModel

Request model for updating OpRun via PATCH.

Parameters:

state (str | None)
outputs (dict[str, Any] | None)
priority (int | None)

state: str | None#

outputs: dict[str, Any] | None#

priority: int | None#

model_config: ClassVar[ConfigDict] = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class gyoza.server.api.models.OpAttemptResponse(*, id, op_run_id, attempt, state, progress, events, inputs, outputs, execution_summary, constraints, reservation, started_at, finished_at)[source]#

Bases: BaseModel

Response model for OpAttempt.

Parameters:

id (str)
op_run_id (str)
attempt (int)
state (str)
progress (int)
events (list[dict[str, Any]])
inputs (dict[str, Any])
outputs (dict[str, Any])
execution_summary (dict[str, Any])
constraints (dict[str, Any])
reservation (dict[str, int])
started_at (str | None)
finished_at (str | None)

id: str#

op_run_id: str#

attempt: int#

state: str#

progress: int#

events: list[dict[str, Any]]#

inputs: dict[str, Any]#

outputs: dict[str, Any]#

execution_summary: dict[str, Any]#

constraints: dict[str, Any]#

reservation: dict[str, int]#

started_at: str | None#

finished_at: str | None#

model_config: ClassVar[ConfigDict] = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class gyoza.server.api.models.AddEventRequest(*, type, msg, payload=None, attempt=None, worker_id=None)[source]#

Bases: BaseModel

Request model for adding an event to an attempt.

Parameters:

type (str)
msg (str | int)
payload (dict[str, Any] | None)
attempt (int | None)
worker_id (str | None)

type: str#

msg: str | int#

payload: dict[str, Any] | None#

attempt: int | None#

worker_id: str | None#

model_config: ClassVar[ConfigDict] = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class gyoza.server.api.models.EventEntryResponse(*, id, type, t, msg, state, payload=<factory>)[source]#

Bases: BaseModel

Response model for a single event entry.

Parameters:

id (int)
type (str)
t (str)
msg (str | int)
state (str)
payload (dict[str, Any])

id: int#

type: str#

t: str#

msg: str | int#

state: str#

payload: dict[str, Any]#

model_config: ClassVar[ConfigDict] = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class gyoza.server.api.models.EventsResponse(*, events)[source]#

Bases: BaseModel

Response model for polling events.

Parameters:: events (list[EventEntryResponse])

events: list[EventEntryResponse]#

model_config: ClassVar[ConfigDict] = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class gyoza.server.api.models.TopicEventResponse(*, cursor, run_id, attempt, type, msg, t, state, payload=<factory>)[source]#

Bases: BaseModel

A single event in the topic feed.

Parameters:

cursor (str)
run_id (str)
attempt (int)
type (str)
msg (str | int)
t (str)
state (str)
payload (dict[str, Any])

cursor: str#

run_id: str#

attempt: int#

type: str#

msg: str | int#

t: str#

state: str#

payload: dict[str, Any]#

model_config: ClassVar[ConfigDict] = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class gyoza.server.api.models.TopicEventsResponse(*, events, next_cursor=None, has_more=False)[source]#

Bases: BaseModel

Cursor-paginated event feed for a topic.

Parameters:

events (list[TopicEventResponse])
next_cursor (str | None)
has_more (bool)

events: list[TopicEventResponse]#

next_cursor: str | None#

has_more: bool#

model_config: ClassVar[ConfigDict] = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class gyoza.server.api.models.ErrorResponse(*, detail, errors=None)[source]#

Bases: BaseModel

Error response model.

Parameters:

detail (str)
errors (list[str] | None)

detail: str#

errors: list[str] | None#

model_config: ClassVar[ConfigDict] = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class gyoza.server.api.models.GPUModel(*, id, vram_mb, tags=<factory>)[source]#

Bases: BaseModel

GPU resource model.

Parameters:

id (int)
vram_mb (int)
tags (list[str])

id: int#

vram_mb: int#

tags: list[str]#

model_config: ClassVar[ConfigDict] = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class gyoza.server.api.models.ResourcesModel(*, cpu_cores, ram_mb, gpus=<factory>)[source]#

Bases: BaseModel

Hardware resources model.

Parameters:

cpu_cores (int)
ram_mb (int)
gpus (list[GPUModel])

cpu_cores: int#

ram_mb: int#

gpus: list[GPUModel]#

model_config: ClassVar[ConfigDict] = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class gyoza.server.api.models.HeartbeatRequest(*, worker_id, resources, tags=<factory>, running_ops=None)[source]#

Bases: BaseModel

Request model for worker heartbeat.

Parameters:

worker_id (str)
resources (ResourcesModel)
tags (list[str])
running_ops (list[dict[str, Any]] | None)

worker_id: str#

resources: ResourcesModel#

tags: list[str]#

running_ops: list[dict[str, Any]] | None#

model_config: ClassVar[ConfigDict] = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class gyoza.server.api.models.WorkerResponse(*, id, resources, tags, running_ops, created_at, last_heartbeat_at, is_active)[source]#

Bases: BaseModel

Response model for Worker.

Parameters:

id (str)
resources (dict[str, Any])
tags (list[str])
running_ops (list[dict[str, Any]])
created_at (str)
last_heartbeat_at (str)
is_active (bool)

id: str#

resources: dict[str, Any]#

tags: list[str]#

running_ops: list[dict[str, Any]]#

created_at: str#

last_heartbeat_at: str#

is_active: bool#

model_config: ClassVar[ConfigDict] = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class gyoza.server.api.models.WorkerOpRunResponse(*, id, image, inputs, outputs=<factory>, constraints, attempt=None, reservation=None)[source]#

Bases: BaseModel

Response model for WorkerOpRun.

Parameters:

id (str)
image (str)
inputs (dict[str, Any])
outputs (dict[str, Any])
constraints (dict[str, Any])
attempt (int | None)
reservation (dict[str, int] | None)

id: str#

image: str#

inputs: dict[str, Any]#

outputs: dict[str, Any]#

constraints: dict[str, Any]#

attempt: int | None#

reservation: dict[str, int] | None#

model_config: ClassVar[ConfigDict] = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class gyoza.server.api.models.ClaimOpsResponse(*, ops)[source]#

Bases: BaseModel

Response model for claim ops endpoint.

Parameters:: ops (list[WorkerOpRunResponse])

ops: list[WorkerOpRunResponse]#

model_config: ClassVar[ConfigDict] = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Op Definition#

Repository for OpDefinition persistence.

This is the only place that knows about MongoDB. It handles serialization and deserialization of OpDefinition entities. OpDefinitions are identified by a composite key (id + version). Multiple versions of the same definition can coexist.

class gyoza.server.op_definition.repository.OpDefinitionRepository(collection)[source]#

Bases: object

Repository for OpDefinition persistence.

Supports versioning - definitions are identified by (id, version). Same id + version = replace. Same id + different version = new entry.

Parameters:: collection (Collection) – MongoDB collection for OpDefinitions.

get(definition_id, version=None)[source]#

Retrieve an OpDefinition by its ID and optionally version.

Parameters:

definition_id (str) – The unique identifier (name) of the OpDefinition.
version (str | None) – The version of the definition. If None, returns the latest version.

Returns:

The OpDefinition if found, None otherwise.

Return type:

OpDefinition | None

upsert(definition)[source]#

Create or update an OpDefinition.

If an OpDefinition with the same id AND version exists, it will be replaced. If the version is different, a new entry is created (preserving old versions).

Parameters:: definition (OpDefinition) – The OpDefinition to upsert.
Returns:: The upserted OpDefinition with updated timestamps.
Return type:: OpDefinition

save(definition)[source]#

Persist an OpDefinition (insert or update).

Parameters:: definition (OpDefinition) – The OpDefinition to persist.
Return type:: None

delete(definition_id, version=None)[source]#

Delete an OpDefinition by its ID and optionally version.

Parameters:

definition_id (str) – The unique identifier of the OpDefinition.
version (str | None) – The version to delete. If None, deletes all versions.

Returns:

Number of definitions deleted.

Return type:

int

list_all()[source]#

List all OpDefinitions (all versions).

Returns:: All OpDefinitions in the collection.
Return type:: list[OpDefinition]

list_versions(definition_id)[source]#

List all versions of an OpDefinition.

Parameters:: definition_id (str) – The unique identifier of the OpDefinition.
Returns:: All versions of the definition, sorted by created_at descending.
Return type:: list[OpDefinition]

exists(definition_id, version=None)[source]#

Check if an OpDefinition exists.

Parameters:

definition_id (str) – The unique identifier of the OpDefinition.
version (str | None) – The version to check. If None, checks if any version exists.

Returns:

True if the definition exists, False otherwise.

Return type:

bool

MongoDB connection for the op_definition subdomain.

This module is responsible for: - Creating the MongoClient (with internal connection pool) - Selecting the database - Exposing ready-to-use collections

This runs ONCE when the subdomain is imported.

Op Run#

Scheduler#

Worker Pool#

WorkerPool for in-memory worker management.

Workers register via heartbeats and are stored in memory. No database persistence since workers are ephemeral.

class gyoza.server.worker.pool.WorkerPool[source]#

Bases: object

In-memory pool of workers.

Workers register themselves via heartbeats containing their full configuration. Each heartbeat creates or updates the worker entry.

heartbeat(hb)[source]#

Register or update a worker via heartbeat.

If worker exists, updates resources, tags and last_heartbeat_at. If worker is new, creates it with current timestamp.

Parameters:: hb (Heartbeat) – Heartbeat payload from the worker.
Returns:: The created or updated worker.
Return type:: Worker

get(worker_id)[source]#

Get a worker by id.

Parameters:: worker_id (str) – Worker name/identifier.
Returns:: The worker if found, None otherwise.
Return type:: Worker | None

list_active(timeout_seconds=60)[source]#

List workers with recent heartbeats.

Parameters:: timeout_seconds (int) – Maximum seconds since last heartbeat to be considered active.
Returns:: Workers with heartbeat within timeout.
Return type:: list[Worker]

list_all()[source]#

List all workers regardless of status.

Returns:: All registered workers.
Return type:: list[Worker]

remove(worker_id)[source]#

Remove a worker from the pool.

Parameters:: worker_id (str) – Worker name/identifier.
Returns:: The removed worker if found, None otherwise.
Return type:: Worker | None

clear()[source]#

Remove all workers from the pool.

Return type:: None

Worker entity.

Workers are ephemeral resources that execute tasks and send heartbeats.

class gyoza.server.worker.worker.Worker(id, resources, tags=list(), running_ops=list(), created_at=<factory>, last_heartbeat_at=<factory>)[source]#

Bases: object

Representation of a worker that can execute tasks.

Workers are identified by their id (name) and register themselves via heartbeats containing their resources and tags.

Parameters:

id (str) – Worker name/identifier.
resources (Resources) – Hardware resources available.
tags (list[str]) – Worker-level tags.
running_ops (list[WorkerOpRun]) – Current running Op runs on the worker.
created_at (datetime) – First heartbeat time.
last_heartbeat_at (datetime) – Most recent heartbeat time.

id: str#

resources: Resources#

tags: list[str]#

running_ops: list[WorkerOpRun]#

created_at: datetime#

last_heartbeat_at: datetime#

is_active(timeout_seconds=30)[source]#

Check if worker is active (heartbeat within timeout).

Parameters:: timeout_seconds (int) – Maximum seconds since last heartbeat to be considered active.
Returns:: True if last heartbeat is within timeout.
Return type:: bool

can_run(constraints)[source]#

Check if a run asking for constraints fits this worker when idle.

Parameters:: constraints (Constraints) – Requirements of the run.
Returns:: True if the worker has every required tag, enough RAM and a GPU with enough VRAM.
Return type:: bool

update_heartbeat(resources, tags, running_ops=None)[source]#

Update worker with new heartbeat data.

Parameters:

resources (Resources) – Updated hardware resources.
tags (list[str]) – Updated worker tags.
running_ops (list[WorkerOpRun] | None) – Running Op runs, if provided by the heartbeat.

Return type:

None

to_dict()[source]#

Convert to dictionary representation.

Return type:: dict[str, Any]

classmethod from_dict(data)[source]#

Create from dictionary representation.

Return type:: Worker
Parameters:: data (dict[str, Any])

Value objects for the Worker subdomain.

These are simple data structures representing hardware resources and execution tasks.

class gyoza.server.worker.resources.GPU(id, vram_mb, tags=list())[source]#

Bases: object

Representation of a single GPU.

Parameters:

id (int) – Numeric GPU identifier (0, 1, 2…).
vram_mb (int) – VRAM in megabytes.
tags (list[str]) – GPU tags (e.g., “cuda”, “nvidia-a100”, “tensor-cores”).

id: int#

vram_mb: int#

tags: list[str]#

to_dict()[source]#

Convert to dictionary representation.

Return type:: dict[str, Any]

classmethod from_dict(data)[source]#

Create from dictionary representation.

Return type:: GPU
Parameters:: data (dict[str, Any])

class gyoza.server.worker.resources.Resources(cpu_cores, ram_mb, gpus=list())[source]#

Bases: object

Hardware resources available on a worker.

Parameters:

cpu_cores (int) – Number of CPU cores.
ram_mb (int) – RAM in megabytes.
gpus (list[GPU]) – List of available GPUs.

cpu_cores: int#

ram_mb: int#

gpus: list[GPU]#

to_dict()[source]#

Convert to dictionary representation.

Return type:: dict[str, Any]

classmethod from_dict(data)[source]#

Create from dictionary representation.

Return type:: Resources
Parameters:: data (dict[str, Any])

class gyoza.server.worker.resources.WorkerOpRun(id, image, inputs=dict(), outputs=dict(), constraints=Constraints(), attempt=None, reservation=None)[source]#

Bases: object

Simplified OpRun representation for worker execution. Contains only the essential fields needed for a worker to execute a task.

Parameters:

id (str) – Unique identifier for the run.
image (str) – Docker image identifier.
inputs (dict[str, Any]) – Input parameters for execution.
outputs (dict[str, Any]) – Partial outputs template (fields pre-filled with upload URLs).
constraints (Constraints) – Hardware requirements.
attempt (int | None) – Attempt number the worker was handed. Its events carry it so the server can refuse the ones an abandoned attempt sends late.
reservation (dict[str, int] | None)

id: str#

image: str#

inputs: dict[str, Any]#

outputs: dict[str, Any]#

constraints: Constraints#

attempt: int | None = None#

reservation: dict[str, int] | None = None#

to_dict()[source]#

Convert to dictionary representation.

Return type:: dict[str, Any]

classmethod from_dict(data)[source]#

Create from dictionary representation.

Return type:: WorkerOpRun
Parameters:: data (dict[str, Any])

classmethod from_op_run(op_run)[source]#

Create WorkerOpRun from OpRun.

Extracts only the essential fields needed for worker execution.

Parameters:: op_run (OpRun) – The OpRun to convert.
Returns:: Simplified OpRun representation for execution.
Return type:: WorkerOpRun

class gyoza.server.worker.resources.Heartbeat(worker_id, resources, tags=list(), running_ops=None)[source]#

Bases: object

Heartbeat sent by a worker to register or update itself.

This is the payload workers send via REST to announce their presence.

Parameters:

worker_id (str) – Worker name/identifier.
resources (Resources) – Hardware resources available.
tags (list[str]) – Worker-level tags.
running_ops (list[WorkerOpRun] | None) – Running Op runs reported at heartbeat time.

worker_id: str#

resources: Resources#

tags: list[str]#

running_ops: list[WorkerOpRun] | None = None#

to_dict()[source]#

Convert to dictionary representation.

Return type:: dict[str, Any]

classmethod from_dict(data)[source]#

Create from dictionary representation.

Return type:: Heartbeat
Parameters:: data (dict[str, Any])

Runner#

Server runner.

Entry point called by gyoza server start. Verifies connections and starts the uvicorn server with configuration from environment variables.

Environment variables#

GYOZA_HOST: Host to bind to. Defaults to 127.0.0.1.
GYOZA_PORT: Port to bind to. Defaults to 5555.
GYOZA_RELOAD: Enable auto-reload for development. Defaults to false.
GYOZA_WORKERS: Number of uvicorn worker processes. Defaults to 1.
GYOZA_LOG_LEVEL: Logging level (debug, info, warning, error). Defaults to info.

gyoza.server.runner.start(host=None, port=None, reload=None, workers=None, log_level=None)[source]#

Start the gyoza server.

Parameters passed explicitly take precedence over environment variables, which in turn take precedence over the built-in defaults. This lets the gyoza server start CLI pass Typer-parsed values directly.

Parameters:

host (str | None) – Host to bind to.
port (int | None) – Port to listen on.
reload (bool | None) – Enable uvicorn auto-reload (development only).
workers (int | None) – Number of uvicorn worker processes.
log_level (str | None) – Logging level string.

Return type:

None