kubex.core

Auto-generated reference for the kubex.core module.

Exceptions

kubex.core.exceptions

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

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))

Patch types

kubex.core.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")

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))

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

Request builder

kubex.core.request_builder.builder