CRUD Operations¶

Api[ResourceType] exposes the full set of Kubernetes CRUD operations as async methods. Every method returns the server-side resource (or a Status for deletions), fully parsed by Pydantic.

get¶

Fetch a single resource by name:

from kubex.api import Api
from kubex.client import create_client
from kubex.k8s.v1_35.core.v1.pod import Pod

client = await create_client()
async with client:
    api: Api[Pod] = Api(Pod, client=client, namespace="default")
    pod = await api.get("my-pod")
    print(pod.metadata.name)

Pass resource_version= to pin the read to a specific version; omit it for the current state.

Names and namespaces¶

Every method that targets a single object (get, replace, patch, delete, and every subresource call) requires a non-empty name: passing None or "" raises ValueError("resource name may not be empty") before any request is sent, so delete() can never turn into a collection delete. Names and namespaces are validated like client-go does — ".", ".." and anything containing / or % raise ValueError — and are percent-encoded, so a name such as weird#name addresses exactly that object. An empty namespace (namespace="") is also rejected; use namespace=None for "all namespaces" on list, watch and delete_collection.

local = Pod(
    metadata=ObjectMetadata(generate_name="example-pod-"),
    spec=PodSpec(containers=[Container(name="example", image="nginx")]),
)
created = await api.create(local)
await api.delete(local.metadata.name)  # ValueError: only the server knows the name
await api.delete(created.metadata.name)  # OK: the server-assigned name

list¶

List all resources in a namespace (or cluster-wide):

pods = await api.list()
for pod in pods.items:
    print(pod.metadata.name)

Filter with label_selector= (e.g. "app=nginx") or field_selector= (e.g. "status.phase=Running"). Use limit= and continue_token= for paginated results:

page = await api.list(label_selector="app=nginx", limit=100)
while page.metadata.continue_:
    page = await api.list(
        label_selector="app=nginx", limit=100, continue_token=page.metadata.continue_
    )

Pass namespace=None to list across all namespaces even when the Api instance has a default namespace set:

all_pods = await api.list(namespace=None)

Objects that do not fit the model¶

By default, list() raises pydantic's ValidationError if any item does not fit the model. Pass invalid=InvalidObjects.INCLUDE (or the string "include") to get a ListWithInvalid instead: items holds the objects that fit, and invalid the others, each an InvalidObject with raw, error and metadata:

from kubex.core.params import InvalidObjects

pods = await api.list(invalid=InvalidObjects.INCLUDE)
for pod in pods.items:
    print(pod.metadata.name)
for bad in pods.invalid:
    print(f"{bad.metadata.name} does not fit the model: {bad.error}")

ListWithInvalid is a ListEntity, so metadata.resource_version and metadata.continue_ work as usual; invalid is not serialized. The list itself is still checked strictly, and an item without a readable metadata.name still raises. See Objects that do not fit the model for when this happens and what other calls do with such objects.

create¶

Create a new resource. The returned object is the server-assigned representation (including metadata.name when generate_name was used):

from kubex.k8s.v1_35.core.v1.container import Container
from kubex.k8s.v1_35.core.v1.pod_spec import PodSpec
from kubex_core.models.metadata import ObjectMetadata

pod = await api.create(
    Pod(
        metadata=ObjectMetadata(generate_name="example-pod-"),
        spec=PodSpec(containers=[Container(name="example", image="nginx")]),
    ),
)
print(pod.metadata.name)  # server-assigned name, e.g. "example-pod-7g4k2"

Pass dry_run=True to validate the request without persisting anything. Pass field_manager= to set the field manager for server-side apply.

replace¶

Replace a resource in its entirety (the get → modify → replace pattern):

current = await api.get("example-pod")
current.metadata.labels = {**(current.metadata.labels or {}), "env": "staging"}
updated = await api.replace("example-pod", current)
print(updated.metadata.labels)

The full object (including resourceVersion) must be present in the payload — the API server uses resourceVersion as an optimistic-concurrency check. If anything else modified the object between the get and the replace (a controller, the kubelet updating a new pod), the server answers 409 Conflict; re-read and retry:

from kubex.core.exceptions import Conflict

for _ in range(5):
    current = await api.get("example-pod")
    current.metadata.labels = {**(current.metadata.labels or {}), "env": "staging"}
    try:
        updated = await api.replace("example-pod", current)
    except Conflict:
        continue
    break

For changes to a few fields, a patch avoids the read-modify-write cycle altogether.

delete¶

Delete a resource by name. The return type is Status | ResourceType: if the resource has finalizers the API server returns the updated object (with deletionTimestamp set) rather than a Status:

result = await api.delete("example-pod")

Control deletion behaviour with optional parameters:

Parameter	Description
`grace_period_seconds`	Override the resource's `terminationGracePeriodSeconds`
`propagation_policy`	`"Foreground"`, `"Background"`, or `"Orphan"`
`preconditions`	`Precondition(uid=..., resource_version=...)` for safe deletes
`dry_run`	Validate without actually deleting (`True` or `"All"`)

from kubex.core.params import Precondition

await api.delete(
    "example-pod",
    grace_period_seconds=0,
    propagation_policy="Foreground",
    preconditions=Precondition(uid=pod.metadata.uid),
)

delete_collection¶

Delete multiple resources matching a selector in a single call:

result = await api.delete_collection(label_selector="app=example-batch")

delete_collection accepts the same filter parameters as list (label_selector, field_selector, limit, continue_token) plus the same deletion options as delete (grace_period_seconds, propagation_policy, preconditions, dry_run).

The return type is Status | ListEntity[ResourceType] — some versions of the Kubernetes API return the deleted list rather than a Status.

Full example from examples/delete_collection.py:

import uuid
from kubex.api import Api
from kubex.client import create_client
from kubex.k8s.v1_35.core.v1.container import Container
from kubex.k8s.v1_35.core.v1.pod import Pod
from kubex.k8s.v1_35.core.v1.pod_spec import PodSpec
from kubex_core.models.metadata import ObjectMetadata


async def main() -> None:
    client = await create_client()
    async with client:
        api: Api[Pod] = Api(Pod, client=client, namespace="default")
        run_id = uuid.uuid4().hex[:8]
        label_selector = f"app=example-batch-{run_id}"

        for i in range(3):
            await api.create(
                Pod(
                    metadata=ObjectMetadata(
                        generate_name=f"example-delete-collection-{i}-",
                        labels={"app": f"example-batch-{run_id}"},
                    ),
                    spec=PodSpec(
                        containers=[Container(name="nginx", image="nginx:latest")]
                    ),
                ),
            )

        result = await api.delete_collection(label_selector=label_selector)
        print(f"delete_collection result: {type(result).__name__}")

Error handling¶

All non-2xx responses raise a subclass of KubexApiError. The most common errors for CRUD operations:

Exception	HTTP status	Typical cause
`NotFound`	404	Resource does not exist
`Conflict`	409	`create` on an existing resource, or `replace` with a stale `resourceVersion`
`Forbidden`	403	RBAC: the service account lacks permission
`UnprocessableEntity`	422	Validation failure (schema mismatch, required field missing)

See Exceptions for the full hierarchy.