Queue

Package: flyteplugins.union.remote

Represents a Union scheduling queue.

Parameters

        
    
class Queue(
    pb2: queue_pb2.Queue,
)

Parameter	Type	Description
`pb2`	`queue_pb2.Queue`

Properties

Property	Type	Description
`action_concurrency`	`int`
`cluster_managed`	`bool`	True for the queue a cluster owns — the co-named one created with it. Such a queue’s clusters and cluster pool belong to the cluster and cannot be changed through `Queue.update`; move the cluster with `Cluster.update` instead. Everything else stays editable.
`cluster_pool`	`str`
`clusters`	`list[str]`
`created_at`	`str`
`created_by`	`str`	Who created the object, or "" when the server did not say or sent only a subject.
`deleted_at`	`str`	When the queue was soft-deleted, or "" if it is live.
`depth`	`int`
`domain`	`str`
`fairness`	`str`
`has_authorship`	`bool`	True when at least one of `created_by` / `updated_by` renders as something readable.
`is_deleted`	`bool`
`max_accelerators`	`dict[str, int]`	Per-accelerator caps on the queue’s in-flight GPU devices, as `{selector: count}`. Selectors are `class[/device[/partition]]` with the device by its canonical name (`nvidia_gpu`, `nvidia_gpu/nvidia-t4`, `nvidia_gpu/nvidia-a100/1g.5gb`); a cap stored under an older spelling reads back as its canonical name. An empty dict means no accelerator cap; GPUs are capped only here.
`max_resources`	`dict[str, str]`	Resource cap on the queue’s dispatched, not-yet-completed actions, as `{name: quantity}`. Only the dimensions the cap names are present; an absent dimension is unlimited. An empty dict means the queue has no cap at all.
`name`	`str`
`organization`	`str`
`priority`	`str`
`project`	`str`
`run_concurrency`	`int`
`scheduling`	`str`	What the resource gate does when the head of the queue does not fit. `strict_fifo` holds the line behind it; `greedy_capacity` skips it and keeps packing. An unset field reads as `strict_fifo` — that is what the scheduler applies.
`state_generation`	`int`	Server-owned counter of queue state changes (spec updates don’t touch it). State-changing requests carry the generation the caller observed, so a delayed request cannot apply to a state the caller never saw. Starts at 1 on creation; queues created before the counter existed read 0 until their first transition.
`status`	`str`
`updated_at`	`str`
`updated_by`	`str`	Who last changed the object (spec, state, deletion or undeletion), or "" as above.

Method	Description
`activate()`	Re-activate a draining or drained queue.
`authorship_rows()`	Label / value rows for a detail view; empty when nothing renders.
`create()`	Create a new queue.
`delete()`	Request deletion of a queue.
`details()`	Get a single point-in-time metrics snapshot for a queue.
`drain()`	Begin draining a queue — stops new submissions, lets in-flight work complete.
`get()`	Get a queue by name.
`listall()`	List queues in the organization.
`to_dict()`	Convert the object to a JSON-serializable dictionary.
`to_json()`	Convert the object to a JSON string.
`undelete()`	Restore a soft-deleted queue with the spec it had when it was deleted.
`update()`	Update a queue’s configuration.
`watch()`	Stream real-time queue metrics via gRPC server streaming.
`watch_pool()`	Stream metrics for every live queue of the given pools (all pools when unset).

cluster_pool is optional: when unset, the server places the queue in the pool named default. Use Queue.update to move the queue to a different pool later — that requires the queue to be drained first.

clusters defaults to ["*"], routing to every cluster in the assigned pool; the wildcard cannot be mixed with explicit cluster names.

max_resources caps the summed resource request of the queue’s dispatched, not-yet-completed actions, as {name: quantity} — e.g. {"cpu": "64", "memory": "512Gi"}. Names are cpu, memory and ephemeral_storage; values are Kubernetes quantities. A dimension not named is unlimited (0 is a hard cap, not “unset”); None or {} means no cap. GPUs are not capped here — see max_accelerators.

max_accelerators caps the in-flight GPU devices per accelerator type, as {selector: count} — e.g. {"T4": 2, "A100/1g.5gb": 1, "nvidia_gpu": 8}. A selector is a task-style device shorthand (T4), optionally with a partition, a canonical class/device[/partition], or a class alone to cap every device of that class. A device is charged to every cap covering it; a type not named is unlimited; counts are positive. Every GPU request names its type (a task that omits it gets the run’s default GPU type), so every GPU in flight is charged to the caps covering its type. None or {} means no accelerator cap. Enforced with the same scheduling rule as max_resources.

scheduling is what the resource gate does when the action at the head of the queue does not fit the remaining capacity: strict_fifo (the default, and what an unset field means) holds the line behind it, while greedy_capacity skips it and keeps packing what does fit.

A taken name is rejected as already existing — an existing queue, a soft-deleted one, or a cluster’s co-named implicit queue (see Cluster.create). Temporarily, only organization-scoped queues can be created: the server rejects an id carrying project or domain.

Parameter	Type	Description
`cls`
`name`	`str`
`org`	`str \| None`
`domain`	`str`
`project`	`str`
`run_concurrency`	`int`
`action_concurrency`	`int`
`depth`	`int`
`priority`	`str`
`fairness`	`str`
`clusters`	`list[str] \| None`
`cluster_pool`	`str \| None`
`max_resources`	`Mapping[str, object] \| None`
`max_accelerators`	`Mapping[str, object] \| None`
`scheduling`	`str`

delete()

This method can be called both synchronously or asynchronously.

Default invocation is sync and will block. To call it asynchronously, use the function .aio() on the method name itself, e.g.,: result = await Queue.delete.aio().

        
    
def delete(
    cls,
    name: str,
    org: str | None = None,
    domain: str = '',
    project: str = '',
) -> Queue

Request deletion of a queue. What it does depends on the queue’s state:

drained: the system has already confirmed the queue holds no work, so it is soft-deleted right away and comes back deleted;
draining: the queue may still hold work, so it comes back deleting — it stays listed, and the system moves it to deleted once its remaining state is cleaned up. Deletion cannot be cancelled; a deleted queue can be brought back with Queue.undelete;
active: rejected — a queue that is accepting work must first be drained (see Queue.drain), so deleting a serving queue always takes two deliberate calls;
deleting or deleted: rejected, the deletion is already under way or done.

A deleted queue disappears from Queue.listall and is no longer scheduled on — Queue.get still returns it, carrying deleted_at — but it keeps its name reserved: Queue.create with the same name is rejected until the queue is undeleted. Use Queue.listall(deleted=True) to find deleted queues and Queue.undelete to restore one.

A queue referenced as run.default_queue in settings at any scope cannot be deleted until those settings are updated or unset. The reserved default queue is no exception: it can be drained and deleted like any other queue, but while it is deleted, runs that name no queue (and have no run.default_queue setting) are rejected at creation.

Returns the queue as the request left it (deleted or deleting).

Parameter	Type	Description
`cls`
`name`	`str`
`org`	`str \| None`
`domain`	`str`
`project`	`str`

details()

This method can be called both synchronously or asynchronously.

Default invocation is sync and will block. To call it asynchronously, use the function .aio() on the method name itself, e.g.,: result = await Queue.details.aio().

        
    
def details(
    cls,
    name: str,
    org: str | None = None,
    domain: str = '',
    project: str = '',
) -> dict

Get a single point-in-time metrics snapshot for a queue.

Reads the first message from the WatchQueueMetrics stream. For continuous updates, use Queue.watch. Works for a queue in any state, drained included.

Parameter	Type	Description
`cls`
`name`	`str`
`org`	`str \| None`
`domain`	`str`
`project`	`str`

drain()

This method can be called both synchronously or asynchronously.

Default invocation is sync and will block. To call it asynchronously, use the function .aio() on the method name itself, e.g.,: result = await Queue.drain.aio().

        
    
def drain(
    cls,
    name: str,
    org: str | None = None,
    domain: str = '',
    project: str = '',
) -> Queue

Begin draining a queue — stops new submissions, lets in-flight work complete.

Any live queue can be drained, including one referenced as run.default_queue in settings at any scope: a draining or drained queue still resolves, and runs that land on a queue that is not accepting work are rejected at creation with an actionable error. Only deleting such a queue is refused (see Queue.delete). A deleting or deleted queue cannot be drained. Use Queue.watch to follow the drain to completion.

Parameter	Type	Description
`cls`
`name`	`str`
`org`	`str \| None`
`domain`	`str`
`project`	`str`

get()

This method can be called both synchronously or asynchronously.

Default invocation is sync and will block. To call it asynchronously, use the function .aio() on the method name itself, e.g.,: result = await Queue.get.aio().

        
    
def get(
    cls,
    name: str,
    org: str | None = None,
    domain: str = '',
    project: str = '',
) -> Queue

Get a queue by name.

Soft-deleted queues are returned too, carrying deleted_at, so a deleted queue’s details stay reachable.

Parameter	Type	Description
`cls`
`name`	`str`
`org`	`str \| None`
`domain`	`str`
`project`	`str`

listall()

This method can be called both synchronously or asynchronously.

Default invocation is sync and will block. To call it asynchronously, use the function .aio() on the method name itself, e.g.,: result = await Queue.listall.aio().

        
    
def listall(
    cls,
    org: str | None = None,
    limit: int = 100,
    deleted: bool = False,
    state: str | None = None,
) -> AsyncIterator[Queue]

List queues in the organization.

With deleted=True the server returns only soft-deleted queues (it filters on the deleted_at marker, which live queues never carry) — that is how a deleted queue is discovered before Queue.undelete.

state (“active”, “draining”, “drained”, “deleting” or “deleted”) narrows the listing to queues in that state, server-side. A deleting queue is still live and listed like any other; a deleted one shows up only under deleted=True.

Parameter	Type	Description
`cls`
`org`	`str \| None`
`limit`	`int`
`deleted`	`bool`
`state`	`str \| None`

to_dict()

        
    
def to_dict()

Convert the object to a JSON-serializable dictionary.

Returns: dict: A dictionary representation of the object.

to_json()

        
    
def to_json()

Convert the object to a JSON string.

Returns: str: A JSON string representation of the object.

undelete()

This method can be called both synchronously or asynchronously.

Default invocation is sync and will block. To call it asynchronously, use the function .aio() on the method name itself, e.g.,: result = await Queue.undelete.aio().

        
    
def undelete(
    cls,
    name: str,
    org: str | None = None,
    domain: str = '',
    project: str = '',
) -> Queue

Restore a soft-deleted queue with the spec it had when it was deleted.

The queue always comes back in the drained state, whatever state its deletion started from — call Queue.activate to make it accept work again. A deleting queue cannot be undeleted: its deletion has to finish first. Undeleting a queue that is not deleted is rejected by the server.

Every cluster the queue routes to must be live and in the queue’s pool, and the pool itself must be live — a restored queue must never point at something that is gone. A cluster’s co-named queue therefore cannot be undeleted while its cluster is deleted: undelete the cluster instead, which brings the queue back with it (see Cluster.undelete). While the cluster is live, it undeletes like any other queue.

Parameter	Type	Description
`cls`
`name`	`str`
`org`	`str \| None`
`domain`	`str`
`project`	`str`

update()

This method can be called both synchronously or asynchronously.

Default invocation is sync and will block. To call it asynchronously, use the function .aio() on the method name itself, e.g.,: result = await Queue.update.aio().

        
    
def update(
    cls,
    name: str,
    org: str | None = None,
    domain: str = '',
    project: str = '',
    run_concurrency: int | None = None,
    action_concurrency: int | None = None,
    depth: int | None = None,
    priority: str | None = None,
    fairness: str | None = None,
    clusters: list[str] | None = None,
    cluster_pool: str | None = None,
    max_resources: Mapping[str, object] | None = None,
    max_accelerators: Mapping[str, object] | None = None,
    scheduling: str | None = None,
) -> Queue

Update a queue’s configuration. Unset fields are read from the current spec.

max_resources follows the same rule: None leaves the current cap exactly as it is (an unrelated edit never drops it), a {name: quantity} mapping replaces it wholesale, and an empty {} removes it — the queue becomes unlimited again. See Queue.create for the accepted names and values. max_accelerators follows the same three-way rule (None keeps, a mapping replaces, {} clears), independently of max_resources. scheduling behaves like every other field: None keeps the queue’s current policy.

The server requires an explicit cluster pool on every UpdateQueue call and rejects requests without one, so the queue’s current pool is resent when cluster_pool is not given.

Passing a different cluster_pool moves the queue: the target pool must already exist, the queue’s clusters must be members of it, and the queue must be in the drained state (see Queue.drain) — the server rejects the move otherwise.

A cluster’s co-named implicit queue is the exception: its clusters and cluster_pool are managed by that cluster and any request changing either is rejected — move the cluster with Cluster.update instead. Its other fields (concurrency, depth, priority, fairness) stay editable.

A soft-deleted queue cannot be updated at all — undelete it first.

Parameter	Type	Description
`cls`
`name`	`str`
`org`	`str \| None`
`domain`	`str`
`project`	`str`
`run_concurrency`	`int \| None`
`action_concurrency`	`int \| None`
`depth`	`int \| None`
`priority`	`str \| None`
`fairness`	`str \| None`
`clusters`	`list[str] \| None`
`cluster_pool`	`str \| None`
`max_resources`	`Mapping[str, object] \| None`
`max_accelerators`	`Mapping[str, object] \| None`
`scheduling`	`str \| None`

watch()

This method can be called both synchronously or asynchronously.

Default invocation is sync and will block. To call it asynchronously, use the function .aio() on the method name itself, e.g.,: result = await Queue.watch.aio().

        
    
def watch(
    cls,
    name: str,
    org: str | None = None,
    domain: str = '',
    project: str = '',
) -> AsyncIterator[dict]

Stream real-time queue metrics via gRPC server streaming.

Yields dicts shaped for the CLI watch UI: the queue’s caps merged with each QueueMetrics snapshot pushed by the server. The server controls the cadence of updates. Works for a queue in any state — watch a draining queue to see its in-flight runs and actions reach zero, and a drained one to confirm it is idle before deleting or moving it.

Parameter	Type	Description
`cls`
`name`	`str`
`org`	`str \| None`
`domain`	`str`
`project`	`str`

watch_pool()

This method can be called both synchronously or asynchronously.

Default invocation is sync and will block. To call it asynchronously, use the function .aio() on the method name itself, e.g.,: result = await Queue.watch_pool.aio().

        
    
def watch_pool(
    cls,
    org: str | None = None,
    pools: list[str] | None = None,
) -> AsyncIterator[list[dict]]

Stream metrics for every live queue of the given pools (all pools when unset).

Yields the latest snapshot of all watched queues whenever any stream reports. Pools are discovered from the queue listing; each pool is one server stream, merged here.

Parameter	Type	Description
`cls`
`org`	`str \| None`
`pools`	`list[str] \| None`