KubeClient
Minimal API client authenticated as the pod's service account.
Attributes
attributenamespace= namespaceThe namespace every object is created in.
attributetoken_path= token_pathattributeverify= verifyattributehttp= httpx.AsyncClient(base_url=base_url, verify=verify, timeout=30.0, transport=transport)attributemountslist[tuple[str, dict[str, Any], str]]= []The server container's mounts as (mount path, volume, subPath),
longest path first.
attributelabelsdict[str, str]= {}The server pod's labels, which node-bound sessions follow.
attributeownerdict[str, Any] | None= NoneOwner reference given to session pods, or None.
attributeshared_claimsset[str]= set()Claims every node can mount (ReadWriteMany).
Functions
constructor__init__(base_url, namespace, verify=True, token_path=os.path.join(SERVICE_ACCOUNT_DIR, 'token'), transport=None) -> NoneCreate a client for base_url scoped to namespace.
parambase_urlstrparamnamespacestrparamverifyssl.SSLContext | bool= Trueparamtoken_pathstr= os.path.join(SERVICE_ACCOUNT_DIR, 'token')paramtransporthttpx.AsyncBaseTransport | None= NoneReturns
Nonefuncin_cluster() -> KubeClientBuild the client from the service account the kubelet mounts.
Returns
app.providers.kubernetes_provider.KubeClientfuncrequest(method, path, **kwargs) -> dict[str, Any]Send one request and return the decoded body.
parammethodstrparampathstrparamkwargsAny= {}Returns
dict[str, typing.Any]functext(method, path, **kwargs) -> strSend one request and return the body as text, for a pod's log.
parammethodstrparampathstrparamkwargsAny= {}Returns
strfuncpath(resource, name='', group='') -> strReturn the URL path of a namespaced resource (collection when name is empty).
paramresourcestrparamnamestr= ''paramgroupstr= ''Returns
strfuncexec(pod, container, command, timeout=30.0) -> strRun a command in a container and return its output, over the API's WebSocket exec channel.
parampodstrparamcontainerstrparamcommandlist[str]paramtimeoutfloat= 30.0Returns
str