kubex.core¶

Auto-generated reference for the kubex.core module.

Exceptions¶

kubex.core.exceptions ¶

ConfigurationValueError ¶

Bases: ConfigurationError, ValueError

A configuration failure that used to raise ValueError.

Also a ValueError, so except ValueError keeps working.

KubexApiError ¶

KubexApiError(
    content: C,
    *args: Any,
    status: HTTPStatus | None = None,
    retry_after: float | None = None,
)

Bases: Generic[C], KubexClientException

Source code in kubex/core/exceptions.py

def __init__(
    self,
    content: C,
    *args: Any,
    status: HTTPStatus | None = None,
    retry_after: float | None = None,
) -> None:
    if status is not None:
        self.status = status
    elif not hasattr(type(self), "status"):
        self.status = HTTPStatus.BAD_REQUEST
    self.content = content
    if retry_after is not None:
        self.retry_after = retry_after
    super().__init__(*(args or (_describe_error(self.status, content),)))

retry_after `class-attribute` `instance-attribute` ¶

retry_after: float | None = None

Seconds to wait before retrying, from the Retry-After response header (or the Status details.retryAfterSeconds) when the server sent one — typically on 429 Too Many Requests and 503 Service Unavailable.

KubexConnectionError ¶

Bases: KubexClientException

The API server could not be reached, or the connection broke mid-response.

KubexTimeoutError ¶

Bases: KubexClientException, TimeoutError

A request exceeded its client-side timeout.

Also a TimeoutError, so except TimeoutError keeps working.

Request and response¶

kubex.core.request ¶

kubex.core.response ¶

API parameters¶

kubex.core.params ¶

AttachOptions ¶

AttachOptions(
    *,
    container: str | None = None,
    stdin: bool = False,
    stdout: bool = True,
    stderr: bool = True,
    tty: bool = False,
)

Options for the Pod attach subresource. Mirrors v1.PodAttachOptions.

Source code in kubex/core/params.py

def __init__(
    self,
    *,
    container: str | None = None,
    stdin: bool = False,
    stdout: bool = True,
    stderr: bool = True,
    tty: bool = False,
) -> None:
    self.container = container
    self.stdin = stdin
    self.stdout = stdout
    self.stderr = stderr
    self.tty = tty

ExecOptions ¶

ExecOptions(
    *,
    command: Sequence[str],
    container: str | None = None,
    stdin: bool = False,
    stdout: bool = True,
    stderr: bool = True,
    tty: bool = False,
)

Options for the Pod exec subresource.

Mirrors the v1.PodExecOptions Kubernetes API. command is required and must be a non-empty sequence of strings — each element serializes as a separate command= query parameter. Boolean flags serialize as the lowercase strings "true" / "false" to match what the Kubernetes API server expects.

Source code in kubex/core/params.py

def __init__(
    self,
    *,
    command: Sequence[str],
    container: str | None = None,
    stdin: bool = False,
    stdout: bool = True,
    stderr: bool = True,
    tty: bool = False,
) -> None:
    # ``str`` satisfies ``Sequence[str]`` (each character is itself a
    # ``str``), so ``list("sh")`` would silently produce ``["s", "h"]``
    # and emit a broken request. Reject it explicitly so callers see a
    # clear local validation error instead of a confusing 400 from the
    # API server.
    if isinstance(command, (str, bytes)):
        raise TypeError(
            "command must be a sequence of strings, not a single "
            f"{type(command).__name__}; pass e.g. ['sh', '-c', 'echo hi']"
        )
    self.command = list(command)
    if not self.command:
        raise ValueError("command must be a non-empty sequence of strings")
    # Each element must be a ``str`` — non-strings would otherwise
    # serialize via ``__str__`` (or fail at the HTTP layer for ``bytes``)
    # and produce a confusing 400 from the API server. Validate here so
    # callers see a clear local error instead.
    for index, element in enumerate(self.command):
        if not isinstance(element, str):
            raise TypeError(
                f"command[{index}] must be str, got {type(element).__name__}"
            )
    self.container = container
    self.stdin = stdin
    self.stdout = stdout
    self.stderr = stderr
    self.tty = tty

InvalidObjects ¶

Bases: str, Enum

What list() and watch() do with an object that does not fit the model.

RAISE (the default) raises pydantic's ValidationError. INCLUDE keeps the object as an InvalidObject instead: list() returns it in ListWithInvalid.invalid and watch() yields an InvalidWatchEvent.

PortForwardOptions ¶

PortForwardOptions(*, ports: Sequence[int])

Options for the Pod portforward subresource.

Validates that ports are unique, non-empty, and within 1..65535. to_query_params() returns repeated ("ports", "<n>") pairs in input order.

Source code in kubex/core/params.py

def __init__(self, *, ports: Sequence[int]) -> None:
    ports_t = tuple(ports)
    if not ports_t:
        raise ValueError("portforward requires at least one port")
    if len(set(ports_t)) != len(ports_t):
        raise ValueError("duplicate ports are not allowed")
    # Channel IDs are a single byte (0..255).  Two channels per port
    # (data=2i, error=2i+1) means index 127 maps to error channel
    # 2*127+1 = 255, which collides with the v5 CHANNEL_CLOSE sentinel.
    # The kubelet portforward currently negotiates v4 only (no
    # CHANNEL_CLOSE sentinel), but we cap at 127 to stay forward-compatible
    # with v5 and to keep error-channel ids inside [0, 254].
    if len(ports_t) > 127:
        raise ValueError(
            f"portforward supports at most 127 ports, got {len(ports_t)}"
        )
    for p in ports_t:
        if isinstance(p, bool) or not isinstance(p, int):
            raise TypeError(f"port must be int, got {type(p).__name__}")
        if not 1 <= p <= 65535:
            raise ValueError(f"port {p} out of range 1..65535")
    self.ports = ports_t

Timeout ¶

Timeout(
    total: float | None = None,
    *,
    connect: float | None = None,
    read: float | None = None,
    write: float | None = None,
    pool: float | None = None,
)

Timeout settings for HTTP requests.

Provides a client-agnostic way to configure timeout settings for the underlying HTTP client. Individual fields may be ignored by a given implementation if the client library does not support them.

Parameters:

Name	Type	Description	Default
`total`	`float \| None`	Total timeout in seconds for the whole request. Used as the default for unset granular fields.	`None`
`connect`	`float \| None`	Timeout in seconds for establishing a connection.	`None`
`read`	`float \| None`	Timeout in seconds for reading the response.	`None`
`write`	`float \| None`	Timeout in seconds for writing the request body. Only honored by the httpx backend.	`None`
`pool`	`float \| None`	Timeout in seconds for acquiring a connection from the pool. Only honored by the httpx backend.	`None`

Source code in kubex/core/params.py

def __init__(
    self,
    total: float | None = None,
    *,
    connect: float | None = None,
    read: float | None = None,
    write: float | None = None,
    pool: float | None = None,
) -> None:
    self.total = total
    self.connect = connect
    self.read = read
    self.write = write
    self.pool = pool

coerce `classmethod` ¶

coerce(value: TimeoutTypes) -> Timeout | None

Normalize a TimeoutTypes value to Timeout or None.

Source code in kubex/core/params.py

@classmethod
def coerce(cls, value: TimeoutTypes) -> Timeout | None:
    """Normalize a ``TimeoutTypes`` value to ``Timeout`` or ``None``."""
    if value is None:
        return None
    if isinstance(value, Timeout):
        return value
    return cls(total=float(value))

format_rfc3339 ¶

format_rfc3339(value: datetime) -> str

Format a timezone-aware datetime as an RFC 3339 UTC timestamp (...Z).

Naive datetimes are rejected rather than guessed at (local time or UTC).

Source code in kubex/core/params.py

def format_rfc3339(value: datetime) -> str:
    """Format a timezone-aware datetime as an RFC 3339 UTC timestamp (`...Z`).

    Naive datetimes are rejected rather than guessed at (local time or UTC).
    """
    if value.tzinfo is None or value.tzinfo.utcoffset(value) is None:
        raise ValueError("datetime must be timezone-aware, e.g. datetime.now(UTC)")
    utc = value.astimezone(timezone.utc)
    if utc.microsecond:
        return utc.strftime("%Y-%m-%dT%H:%M:%S.%fZ")
    return utc.strftime("%Y-%m-%dT%H:%M:%SZ")

Watch events¶

kubex.core.watch_event ¶

INITIAL_EVENTS_END_ANNOTATION `module-attribute` ¶

INITIAL_EVENTS_END_ANNOTATION = 'k8s.io/initial-events-end'

Annotation on the bookmark that ends the initial events of a watch started with send_initial_events=True.

Bookmark ¶

Bases: BaseEntity

Bookmark is a pointer to a resource in a stream.

EventType ¶

Bases: str, Enum

EventType is the type of the watch event.

InvalidWatchEvent `dataclass` ¶

InvalidWatchEvent(type: EventType, object: InvalidObject)

A watch event whose object does not fit the resource model.

Yielded by Api.watch(invalid=InvalidObjects.INCLUDE) in place of a WatchEvent for an ADDED, MODIFIED or DELETED event; the stream goes on. Like a WatchEvent it has type and object.metadata, so code that only needs an object's name, namespace or resource version works on both.

Attributes:

Name	Type	Description
`type`	`EventType`	`ADDED`, `MODIFIED` or `DELETED`.
`object`	`InvalidObject`	The object, with its raw data, the validation error and its metadata.

is_initial_events_end `property` ¶

is_initial_events_end: bool

Always False: only a bookmark ends the initial events.

WatchEvent ¶

WatchEvent(
    resource_type: Type[ResourceType],
    raw_event: dict[str, Any],
)

Bases: Generic[ResourceType]

WatchEvent represents a single event from a watch stream.

Source code in kubex/core/watch_event.py

def __init__(
    self, resource_type: Type[ResourceType], raw_event: dict[str, Any]
) -> None:
    self._resource_type = resource_type
    self.type = EventType(raw_event["type"])
    self.object: ResourceType | Bookmark
    match self.type:
        case EventType.BOOKMARK:
            self.object = Bookmark.model_validate(raw_event["object"])
        case EventType.ERROR:
            try:
                status = Status.model_validate(raw_event["object"])
            except ValidationError:
                raise build_exception(0, str(raw_event["object"]))
            raise build_exception(status.code, status)
        case _:
            self.object = self._resource_type.model_validate(raw_event["object"])

is_initial_events_end `property` ¶

is_initial_events_end: bool

Whether this is the bookmark that ends the initial events.

A watch started with send_initial_events=True first sends an ADDED event for every existing object, then this bookmark. From here on the stream carries live changes, and the bookmark's object.metadata.resource_version is the first one a watch can resume from. Other bookmarks are periodic checkpoints and return False.

Objects that do not fit the model¶

kubex.core.invalid_object ¶

Objects that do not fit their resource model.

Api.list() and Api.watch() raise pydantic's ValidationError when an object from the API server does not fit the model, for example a custom resource stored before its CRD schema changed. With invalid=InvalidObjects.INCLUDE they keep such an object as an InvalidObject instead, so one bad object neither fails a whole list nor ends a watch.

InvalidObject `dataclass` ¶

InvalidObject(
    raw: dict[str, Any],
    error: ValidationError,
    metadata: ObjectMetadata,
)

An object from the API server that does not fit the resource model.

Attributes:

Name	Type	Description
`raw`	`dict[str, Any]`	The object as the server sent it.
`error`	`ValidationError`	Why it does not fit the model.
`metadata`	`ObjectMetadata`	The object's metadata. It always parsed and always has a `name`, so the object can still be identified, e.g. to drop it from a cache on a `DELETED` event.

ListWithInvalid ¶

Bases: ListEntity[ResourceType], Generic[ResourceType]

A list whose objects that do not fit the model are kept apart.

Returned by Api.list(invalid=InvalidObjects.INCLUDE). items holds the objects that fit the model, in server order; invalid holds the rest. The list's metadata (resource_version, continue_) is the server's, so paging and watching from the list work as usual. invalid is not serialized.

Patch types¶

kubex.core.patch ¶

ApplyPatch ¶

ApplyPatch(body: P)

Bases: Patch, Generic[P]

Server-side apply patch (application/apply-patch+yaml).

The body is serialized as JSON (valid YAML, and what kubectl apply --server-side sends), so timestamps and other non-primitive fields use their JSON wire format.

apiVersion and kind are always sent, even though exclude_unset would drop them when they are only model defaults: the server rejects an apply patch without them. Request builders pass the target resource's values explicitly (so e.g. a PartialObjectMetadata body applied through api.metadata.patch() is sent as the target kind); otherwise they come from the body, falling back to its __RESOURCE_CONFIG__. metadata.managedFields is always dropped since the server rejects any non-nil value. metadata.resourceVersion is kept and acts as an optimistic-concurrency precondition, so when applying an object fetched earlier, clear it (obj.metadata.resource_version = None) unless a Conflict on concurrent changes is what you want.

Unlike MergePatch/StrategicMergePatch, exclude_none defaults to True here: a field management conflict is reported by field ownership, not by an explicit null, so there is no RFC 7396-style "null deletes the field" convention to preserve.

Source code in kubex/core/patch.py

def __init__(self, body: P) -> None:
    self.body = body

JsonPatch ¶

Bases: RootModel[list[JsonPatchOperation]]

RFC 6902 JSON Patch document.

Can be constructed directly or built incrementally via chaining::

patch = JsonPatch().add("/a", 1).remove("/b").replace("/c", "new")

JsonPointer ¶

Bases: str

RFC 6901 JSON Pointer.

A JSON Pointer is either the empty string (referencing the whole document) or a sequence of reference tokens each prefixed by /.

Construct from a raw string, from individual tokens, or by chaining the / operator::

ptr = JsonPointer("/foo/0")
ptr = JsonPointer.from_tokens("foo", 0)
ptr = JsonPointer() / "foo" / 0

tokens `property` ¶

tokens: list[str]

Decompose into unescaped reference tokens.

from_tokens `classmethod` ¶

from_tokens(*tokens: str | int) -> JsonPointer

Build a pointer from unescaped reference tokens.

Tokens are escaped per RFC 6901 (~ -> ~0, / -> ~1).

Source code in kubex/core/json_pointer.py

@classmethod
def from_tokens(cls, *tokens: str | int) -> JsonPointer:
    """Build a pointer from unescaped reference tokens.

    Tokens are escaped per RFC 6901 (``~`` -> ``~0``, ``/`` -> ``~1``).
    """
    if not tokens:
        return cls("")
    return cls("/" + "/".join(_escape_token(str(t)) for t in tokens))

MergePatch ¶

MergePatch(body: P)

Bases: Patch, Generic[P]

RFC 7396 JSON merge patch (application/merge-patch+json).

exclude_none defaults to False: RFC 7396 uses an explicit null to delete a key, so a field the caller deliberately set to None must reach the server as null rather than being stripped before serialization. exclude_unset still limits the patch to fields the caller actually touched, so fields left at their default remain distinguishable from explicitly null ones.

Source code in kubex/core/patch.py

def __init__(self, body: P) -> None:
    self.body = body

StrategicMergePatch ¶

StrategicMergePatch(body: P)

Bases: Patch, Generic[P]

Kubernetes strategic merge patch (application/strategic-merge-patch+json).

exclude_none defaults to False for the same reason as MergePatch: an explicit null is meaningful (it deletes the field) and must not be stripped before serialization.

Source code in kubex/core/patch.py

def __init__(self, body: P) -> None:
    self.body = body

JSON Patch (RFC 6902)¶

kubex.core.json_patch ¶

JsonPatch ¶

Bases: RootModel[list[JsonPatchOperation]]

RFC 6902 JSON Patch document.

Can be constructed directly or built incrementally via chaining::

patch = JsonPatch().add("/a", 1).remove("/b").replace("/c", "new")

JSON Pointer (RFC 6901)¶

kubex.core.json_pointer ¶

JsonPointer ¶

Bases: str

RFC 6901 JSON Pointer.

A JSON Pointer is either the empty string (referencing the whole document) or a sequence of reference tokens each prefixed by /.

Construct from a raw string, from individual tokens, or by chaining the / operator::

ptr = JsonPointer("/foo/0")
ptr = JsonPointer.from_tokens("foo", 0)
ptr = JsonPointer() / "foo" / 0

tokens `property` ¶

tokens: list[str]

Decompose into unescaped reference tokens.

from_tokens `classmethod` ¶

from_tokens(*tokens: str | int) -> JsonPointer

Build a pointer from unescaped reference tokens.

Tokens are escaped per RFC 6901 (~ -> ~0, / -> ~1).

Source code in kubex/core/json_pointer.py

@classmethod
def from_tokens(cls, *tokens: str | int) -> JsonPointer:
    """Build a pointer from unescaped reference tokens.

    Tokens are escaped per RFC 6901 (``~`` -> ``~0``, ``/`` -> ``~1``).
    """
    if not tokens:
        return cls("")
    return cls("/" + "/".join(_escape_token(str(t)) for t in tokens))

Subresource definitions¶

kubex.core.subresource ¶

WebSocket channel protocol¶

kubex.core.exec_channels ¶

port_prefix_encode ¶

port_prefix_encode(port: int) -> bytes

Inverse of port_prefix_decode.

Outbound port-forward writes carry no port prefix (the channel id alone addresses the kubelet); only the first inbound frame on each channel is prefixed, which PortForwardSession strips via port_prefix_decode. Kubex itself has no reason to construct a prefix, so this exists for the test suite to build fake inbound frames and round-trip against port_prefix_decode.

Source code in kubex/core/exec_channels.py

def port_prefix_encode(port: int) -> bytes:
    """Inverse of `port_prefix_decode`.

    Outbound port-forward writes carry no port prefix (the channel id alone
    addresses the kubelet); only the *first inbound* frame on each channel is
    prefixed, which `PortForwardSession` strips via `port_prefix_decode`.
    Kubex itself has no reason to construct a prefix, so this exists for the
    test suite to build fake inbound frames and round-trip against
    `port_prefix_decode`.
    """
    return port.to_bytes(2, byteorder="little")

kubex.core¶

Exceptions¶

kubex.core.exceptions ¶

ConfigurationValueError ¶

KubexApiError ¶

retry_after class-attribute instance-attribute ¶

KubexConnectionError ¶

KubexTimeoutError ¶

Request and response¶

kubex.core.request ¶

kubex.core.response ¶

API parameters¶

kubex.core.params ¶

AttachOptions ¶

ExecOptions ¶

InvalidObjects ¶

PortForwardOptions ¶

Timeout ¶

coerce classmethod ¶

format_rfc3339 ¶

Watch events¶

kubex.core.watch_event ¶

INITIAL_EVENTS_END_ANNOTATION module-attribute ¶

Bookmark ¶

EventType ¶

InvalidWatchEvent dataclass ¶

is_initial_events_end property ¶

WatchEvent ¶

is_initial_events_end property ¶

Objects that do not fit the model¶

kubex.core.invalid_object ¶

InvalidObject dataclass ¶

ListWithInvalid ¶

Patch types¶

kubex.core.patch ¶

ApplyPatch ¶

JsonPatch ¶

JsonPointer ¶

tokens property ¶

from_tokens classmethod ¶

MergePatch ¶

StrategicMergePatch ¶

JSON Patch (RFC 6902)¶

kubex.core.json_patch ¶

JsonPatch ¶

JSON Pointer (RFC 6901)¶

kubex.core.json_pointer ¶

JsonPointer ¶

tokens property ¶

from_tokens classmethod ¶

Subresource definitions¶

kubex.core.subresource ¶

WebSocket channel protocol¶

kubex.core.exec_channels ¶

port_prefix_encode ¶

Request builder¶

kubex.core.request_builder.builder ¶

retry_after `class-attribute` `instance-attribute` ¶

coerce `classmethod` ¶

INITIAL_EVENTS_END_ANNOTATION `module-attribute` ¶

InvalidWatchEvent `dataclass` ¶

is_initial_events_end `property` ¶

is_initial_events_end `property` ¶

InvalidObject `dataclass` ¶

tokens `property` ¶

from_tokens `classmethod` ¶

tokens `property` ¶

from_tokens `classmethod` ¶