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#

Main FastAPI application.

gyoza.server.api.app.verify_api_key(api_key=Security(dependency=<fastapi.security.api_key.APIKeyHeader object>, use_cache=True, scope=None, scopes=None))[source]#

Verify API key from header.

Return type:: str
Parameters:: api_key (str | None)

gyoza.server.api.app.health_check()[source]#

Health check endpoint.

Returns:: Health status.
Return type:: dict[str, str]

Pydantic models for API request/response validation.

class gyoza.server.api.models.InputSpecModel(**data)[source]#

Bases: BaseModel

Input specification model.

Parameters:

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

type: str#

required: bool#

default: 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(**data)[source]#

Bases: BaseModel

Hardware constraints model.

Parameters:

ram_mb (int)
vram_mb (int)
cpu (int)

ram_mb: int#

vram_mb: int#

cpu: int#

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

class gyoza.server.api.models.RetryPolicyModel(**data)[source]#

Bases: BaseModel

Retry policy model.

Parameters:

max_attempts (int)
backoff_ms (int)

max_attempts: int#

backoff_ms: 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(**data)[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(**data)[source]#

Bases: BaseModel

Request model for creating/updating OpDefinition.

Parameters:

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

id: str#

version: str#

image: str#

input_specs: dict[str, InputSpecModel] | 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(**data)[source]#

Bases: BaseModel

Response model for OpDefinition.

Parameters:

id (str)
version (str)
image (str)
input_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#

input_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.OpRunCreateFromDefinition(**data)[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:

inputs (dict[str, Any])
priority (int)

inputs: dict[str, Any]#

priority: int#

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

class gyoza.server.api.models.OpRunResponse(**data)[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])
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]#

outputs: dict[str, Any]#

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

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.OpRunCreate(**data)[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])
constraints (ConstraintsModel | None)
retry_policy (RetryPolicyModel | None)
event_delivery (EventDeliveryModel | None)

image: str#

priority: int#

inputs: 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(**data)[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(**data)[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])
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]#

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(**data)[source]#

Bases: BaseModel

Request model for adding an event to an attempt.

Parameters:

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

type: str#

msg: str | int#

payload: 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.EventEntryResponse(**data)[source]#

Bases: BaseModel

Response model for a single event entry.

Parameters:

id (int)
type (str)
t (str)
msg (str | int)
state (str)

id: int#

type: str#

t: str#

msg: str | int#

state: str#

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

class gyoza.server.api.models.EventsResponse(**data)[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.ErrorResponse(**data)[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(**data)[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(**data)[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(**data)[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(**data)[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)

id: str#

resources: dict[str, Any]#

tags: list[str]#

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

created_at: str#

last_heartbeat_at: str#

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

class gyoza.server.api.models.WorkerOpRunResponse(**data)[source]#

Bases: BaseModel

Response model for WorkerOpRun.

Parameters:

id (str)
image (str)
inputs (dict[str, Any])
constraints (dict[str, Any])

id: str#

image: str#

inputs: dict[str, Any]#

constraints: 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.ClaimOpsResponse(**data)[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].

OpDefinition router - template management endpoints.

gyoza.server.api.routers.definitions.upsert_op_definition(request)[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.

Parameters:: request (OpDefinitionCreate) – OpDefinition data.
Returns:: Created or updated OpDefinition.
Return type:: dict[str, Any]

gyoza.server.api.routers.definitions.get_op_definition(name, version=None)[source]#

Get an OpDefinition by name and optionally version.

Parameters:

name (str) – OpDefinition identifier.
version (str | None) – Specific version to retrieve. If None, returns latest version.

Returns:

OpDefinition data.

Return type:

dict[str, Any]

Raises:

HTTPException – If OpDefinition not found.

gyoza.server.api.routers.definitions.list_op_definitions()[source]#

List all OpDefinitions (all versions).

Returns:: List of all OpDefinitions.
Return type:: list[dict[str, Any]]

gyoza.server.api.routers.definitions.create_op_run_from_definition(name, request, version=None)[source]#

Create an OpRun from an OpDefinition.

Parameters:

name (str) – OpDefinition identifier.
request (OpRunCreate) – OpRun creation parameters.
version (str | None) – Specific version to use. If None, uses latest version.

Returns:

Created OpRun data.

Return type:

dict[str, Any]

Raises:

HTTPException – If OpDefinition not found or inputs are invalid.

OpRun router - execution run management endpoints.

gyoza.server.api.routers.runs.create_op_run(request)[source]#

Create an OpRun directly (ad-hoc).

This endpoint allows creating execution runs without a parent OpDefinition. Useful for ad-hoc usage and development testing.

Parameters:: request (OpRunCreate) – Run creation parameters including image, inputs, and constraints.
Returns:: The created OpRun with auto-generated ID and default values.
Return type:: dict[str, Any]

Examples

>>> POST /runs
>>> {
...   "image": "geoiahub/my_product:latest",
...   "priority": 4,
...   "inputs": {"param": "value"},
...   "constraints": {"ram_mb": 4096}
... }

gyoza.server.api.routers.runs.get_op_run(id)[source]#

Get an OpRun by ID.

Parameters:: id (str) – The unique identifier of the OpRun.
Returns:: The OpRun object with all its data.
Return type:: dict[str, Any]
Raises:: HTTPException – 404 if OpRun not found.

gyoza.server.api.routers.runs.update_op_run(id, request)[source]#

Update an OpRun.

Allows updating state, outputs, and priority of a run.

Parameters:

id (str) – The unique identifier of the OpRun.
request (OpRunUpdate) – Fields to update.

Returns:

The updated OpRun object.

Return type:

dict[str, Any]

Raises:

HTTPException – 404 if OpRun not found. 400 if invalid state or update.

gyoza.server.api.routers.runs.retry_op_run(id)[source]#

Retry an OpRun.

Creates a new OpAttempt, increments the current_attempt counter, and resets the state to PENDING.

Parameters:: id (str) – The unique identifier of the OpRun.
Returns:: The OpRun with state=PENDING and incremented attempt counter.
Return type:: dict[str, Any]
Raises:: HTTPException – 404 if OpRun not found. 400 if max attempts reached.

gyoza.server.api.routers.runs.poll_op_run_events(id, after=None)[source]#

Poll OpRun events.

Returns events using a polling-based mechanism. You can provide the starting event index to receive only events after that index.

Parameters:

id (str) – The unique identifier of the OpRun.
after (int | None) – Optional: Only events after this index will be returned.

Returns:

Object containing list of events with their IDs and current state.

Return type:

dict[str, Any]

Raises:

HTTPException – 404 if OpRun not found.

gyoza.server.api.routers.runs.get_op_attempts(id)[source]#

Get all attempts for an OpRun.

Returns the full history of execution attempts for the run.

Parameters:: id (str) – The unique identifier of the OpRun.
Returns:: List of all OpAttempts for this run.
Return type:: list[dict[str, Any]]
Raises:: HTTPException – 404 if OpRun not found.

gyoza.server.api.routers.runs.add_event_to_run(id, request)[source]#

Add an event to the current attempt of an OpRun.

This endpoint is typically called by the gyoza worker during execution. The event is added to the current attempt’s timeline.

Special event types with payloads:

STARTED — payload: {"worker_id": str, "gpu_id": int} (gpu_id optional).
PROGRESS — msg: int (0-100), updates the progress field automatically.
COMPLETED — payload: {"outputs": dict} (optional).
FAILED — payload: {"error_message": str} (optional).
CANCELLED — no payload.

Parameters:

id (str) – The unique identifier of the OpRun.
request (AddEventRequest) – Event details (type, message, and optional payload).

Returns:

The updated OpRun object.

Return type:

dict[str, Any]

Raises:

HTTPException – 404 if OpRun not found, 400 if run is locked or event is invalid.

Worker router - worker management endpoints.

gyoza.server.api.routers.workers.list_workers()[source]#

List all workers in the pool.

Returns all registered workers regardless of their active status.

Returns:: List of all workers with their resources, tags, and status.
Return type:: list[dict[str, Any]]

gyoza.server.api.routers.workers.heartbeat(request)[source]#

Workers send heartbeats to announce their presence and update their status. If the worker exists, it will be updated with the new resources, tags, and running ops. If it’s a new worker, it will be registered.

Parameters:: request (HeartbeatRequest) – Heartbeat payload containing worker_id, resources, tags, and optionally running ops.
Returns:: The created or updated worker object.
Return type:: dict[str, Any]

Examples

>>> POST /workers/heartbeat
>>> {
...   "worker_id": "worker-gpu-01",
...   "resources": {
...     "cpu_cores": 8,
...     "ram_mb": 32768,
...     "gpus": [{"id": 0, "vram_mb": 8192, "tags": ["cuda"]}]
...   },
...   "tags": ["gpu", "high-mem"],
...   "running_ops": []
... }

gyoza.server.api.routers.workers.claim_ops(worker_id)[source]#

Claim ops for a worker.

Workers call this endpoint to request work allocation. The scheduler will determine which pending OpRuns should be allocated to this worker based on the configured scheduling strategy.

Parameters:: worker_id (str) – ID of the worker requesting ops.
Returns:: Object containing list of claimed ops (WorkerOpRun objects).
Return type:: dict[str, Any]

Examples

>>> POST /workers/worker-gpu-01/claim
>>> Response:
>>> {
...   "ops": [
...     {
...       "id": "run_abc123",
...       "image": "geoiahub/product:v1",
...       "inputs": {"param": "value"},
...       "constraints": {"ram_mb": 4096, "vram_mb": 2048}
...     }
...   ]
... }

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#

Repository for OpRun aggregate persistence.

This is the only place that knows about MongoDB. It handles serialization and deserialization of the OpRun aggregate (including its attempts).

class gyoza.server.op_run.repository.OpRunRepository(runs_collection, attempts_collection)[source]#

Bases: object

Repository for OpRun aggregate persistence.

Handles both OpRun and its attempts as a single aggregate. Attempts are stored in a separate collection but loaded together.

Parameters:

runs_collection (Collection) – MongoDB collection for OpRuns.
attempts_collection (Collection) – MongoDB collection for OpAttempts.

get(run_id)[source]#

Retrieve an OpRun aggregate by its ID.

Loads the run and all its attempts, reconstructing the full aggregate.

Parameters:: run_id (str) – The unique identifier of the OpRun.
Returns:: The complete OpRun aggregate if found, None otherwise.
Return type:: OpRun | None

save(run)[source]#

Persist an OpRun aggregate (insert or update).

Saves both the run and all its attempts.

Parameters:: run (OpRun) – The OpRun aggregate to persist.
Return type:: None

delete(run_id)[source]#

Delete an OpRun and all its attempts.

Parameters:: run_id (str) – The unique identifier of the OpRun to delete.
Returns:: True if the run was deleted, False if not found.
Return type:: bool

list_by_state(state)[source]#

List all OpRuns with a specific state.

Parameters:: state (OpRunState) – The state to filter by.
Returns:: List of OpRuns matching the state.
Return type:: list[OpRun]

list_pending_ordered()[source]#

List pending OpRuns ordered by priority (descending).

Returns:: Pending OpRuns sorted by priority (highest first).
Return type:: list[OpRun]

MongoDB connection for the op_run 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.

Scheduler#

Scheduler for allocating OpRuns to workers.

class gyoza.server.scheduler.scheduler.Scheduler(strategy, op_run_repo, worker_pool, worker_timeout_seconds=60)[source]#

Bases: object

Orchestrates run allocation using a pluggable strategy.

The scheduler provides the repository and active workers to the strategy, which decides how to allocate runs.

Parameters:

strategy (SchedulingStrategy) – The scheduling algorithm to use.
op_run_repo (OpRunRepository) – Repository for querying OpRuns.
worker_pool (WorkerPool) – Pool for fetching active workers.
worker_timeout_seconds (int) – Timeout in seconds for considering workers active.

allocate_for_worker(worker_id)[source]#

Get OpRuns that should be allocated to a worker.

Provides the repository and active workers to the strategy, which determines which runs to allocate. Then transitions the selected runs from PENDING to ASSIGNED state.

Parameters:: worker_id (str) – ID of the worker requesting work.
Returns:: OpRuns allocated to the worker. Empty if none available or worker not found.
Return type:: list[WorkerOpRun]

Scheduling strategy base interface.

Defines the interface for pluggable scheduling algorithms.

class gyoza.server.scheduler.strategies.base.SchedulingStrategy(*args, **kwargs)[source]#

Bases: Protocol

Protocol for scheduling strategies.

Implementations determine how to allocate pending runs to workers. Different strategies can implement different allocation policies (e.g., balanced load, priority-based, affinity-based).

allocate(worker_id, op_run_repo, workers)[source]#

Select runs to allocate to the given worker.

Parameters:

worker_id (str) – ID of the worker requesting work.
op_run_repo (OpRunRepository) – Repository for querying OpRuns (pending, running, etc.).
workers (list[Worker]) – List of all active workers with their current state.

Returns:

Op runs that should be allocated to the requesting worker.

Return type:

list[WorkerOpRun]

Direct scheduling strategy.

Simple strategy that directly assigns the first available pending run to the worker.

class gyoza.server.scheduler.strategies.direct.DirectSchedulingStrategy[source]#

Bases: object

Simple scheduling strategy that directly assigns pending runs.

This strategy assigns the first pending run (highest priority) directly to the requesting worker. It’s deterministic and suitable for testing.

allocate(worker_id, op_run_repo, workers)[source]#

Select runs to allocate to the given worker.

Directly assigns the first pending run (highest priority) to the worker. Returns empty list if no pending runs are available.

Parameters:

worker_id (str) – ID of the worker requesting work.
op_run_repo (OpRunRepository) – Repository for querying OpRuns.
workers (list[Worker]) – List of all active workers (unused in this strategy).

Returns:

First pending Op run, or empty list if none available.

Return type:

list[WorkerOpRun]

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]#

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=<factory>, running_ops=<factory>, 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

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=<factory>)[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=<factory>)[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=<factory>, constraints=<factory>)[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.
constraints (Constraints) – Hardware requirements.

id: str#

image: str#

inputs: dict[str, Any]#

constraints: Constraints#

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=<factory>, 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:

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