poiesis.core.adaptors.kubernetes package

Submodules

poiesis.core.adaptors.kubernetes.kubernetes module

Kubernetes adaptor.

class poiesis.core.adaptors.kubernetes.kubernetes.KubernetesAdapter

Bases: KubernetesPort

Kubernetes adaptor.

core_api

The Kubernetes Core API

batch_api

The Kubernetes Batch API

namespace

The Kubernetes namespace

async create_config_map(configmap)

Create a config map.

Parameters:

configmap (V1ConfigMap) – The ConfigMap object.

Return type:

V1ConfigMap

async create_job(job)

Create a Kubernetes Job.

Parameters:

job (V1Job) – The Kubernetes Job object.

Return type:

V1Job

async create_pod(pod)

Create a task execution pod.

Parameters:

pod (V1Pod) – The pod object.

Return type:

str

async create_pvc(pvc)

Create a Persistent Volume Claim.

Parameters:

pvc (V1PersistentVolumeClaim) – The Persistent Volume Claim object.

Return type:

str

async delete_job(name)

Delete a job.

Parameters:

name (str) – The name of the job.

Return type:

None

async delete_jobs_by_label(label_selector)

Delete jobs by label selector.

Parameters:

label_selector (str) – The label selector.

Return type:

None

async delete_pods_by_label(label_selector)

Delete pods by label selector.

Parameters:

label_selector (str) – The label selector.

Return type:

None

async delete_pvc(name)

Delete a Persistent Volume Claim.

Return type:

None

async delete_pvcs_by_label(label_selector)

Delete PVCs by label selector.

Parameters:

label_selector (str) – The label selector.

Return type:

None

Notes

Uses ‘Foreground’ propagation policy to ensure dependent resources are deleted before the PVC is removed.

async get_job(name)

Get a Kubernetes Job.

Parameters:

name (str) – The name of the Job.

Return type:

V1Job

async get_pod(name)

Get a specific pod.

Parameters:

name (str) – The name of the pod.

Return type:

V1Pod

async get_pod_log(name)

Get log of a pod.

Parameters:

name (str) – The name of the pod.

Return type:

str

async get_pods(label_selector)

Get all pods matching the label selector.

Parameters:

label_selector (str) – The label selector.

Return type:

list[V1Pod]

async list_jobs_by_label(label_selector)

List jobs by label selector.

Parameters:

label_selector (str) – The label selector.

Return type:

list[V1Job]

async list_pods_by_label(label_selector)

List pods by label selector.

Parameters:

label_selector (str) – The label selector.

Return type:

list[V1Pod]

async list_pvcs_by_label(label_selector)

List PVCs by label selector.

Parameters:

label_selector (str) – The label selector.

Return type:

list[V1PersistentVolumeClaim]

async patch_config_map(name, configmap)

Patch a ConfigMap.

Parameters:

• name (str) – The name of the ConfigMap to patch.

• configmap (V1ConfigMap) – The updated ConfigMap object (with changes such as ownerReferences).

Return type:

str

poiesis.core.adaptors.kubernetes.models module

Models for the core services.

class poiesis.core.adaptors.kubernetes.models.V1CapabilitiesPydanticModel(**data)

Bases: BaseModel

Capabilities.

add: list[str] | None

drop: list[str] | None

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

to_k8s_model()

Convert to Kubernetes model.

Return type:

V1Capabilities

class poiesis.core.adaptors.kubernetes.models.V1PodSecurityContextPydanticModel(**data)

Bases: BaseModel

Pod security context.

fs_group: int | None

fs_group_change_policy: str | None

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

run_as_group: int | None

run_as_non_root: bool | None

run_as_user: int | None

se_linux_options: V1SELinuxOptionsPydanticModel | None

seccomp_profile: V1SeccompProfilePydanticModel | None

supplemental_groups: list[int] | None

sysctls: list[V1SysctlPydanticModel] | None

to_k8s_model()

Convert to Kubernetes model.

Return type:

V1PodSecurityContext

windows_options: V1WindowsSecurityContextOptionsPydanticModel | None

class poiesis.core.adaptors.kubernetes.models.V1SELinuxOptionsPydanticModel(**data)

Bases: BaseModel

SELinux options.

level: str | None

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

role: str | None

to_k8s_model()

Convert to Kubernetes model.

Return type:

V1SELinuxOptions

type: str | None

user: str | None

class poiesis.core.adaptors.kubernetes.models.V1SeccompProfilePydanticModel(**data)

Bases: BaseModel

Seccomp profile.

localhost_profile: str | None

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

to_k8s_model()

Convert to Kubernetes model.

Return type:

V1SeccompProfile

type: str

class poiesis.core.adaptors.kubernetes.models.V1SecurityContextPydanticModel(**data)

Bases: BaseModel

Security context.

allow_privilege_escalation: bool | None

capabilities: V1CapabilitiesPydanticModel | None

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

privileged: bool | None

proc_mount: str | None

read_only_root_filesystem: bool | None

run_as_group: int | None

run_as_non_root: bool | None

run_as_user: int | None

se_linux_options: V1SELinuxOptionsPydanticModel | None

seccomp_profile: V1SeccompProfilePydanticModel | None

to_k8s_model()

Convert to Kubernetes model.

Return type:

V1SecurityContext

windows_options: V1WindowsSecurityContextOptionsPydanticModel | None

class poiesis.core.adaptors.kubernetes.models.V1SysctlPydanticModel(**data)

Bases: BaseModel

Sysctl.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str

to_k8s_model()

Convert to Kubernetes model.

Return type:

V1Sysctl

value: str

class poiesis.core.adaptors.kubernetes.models.V1WindowsSecurityContextOptionsPydanticModel(**data)

Bases: BaseModel

Windows security context options.

gmsa_credential_spec: str | None

gmsa_credential_spec_name: str | None

host_process: bool | None

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

run_as_user_name: str | None

to_k8s_model()

Convert to Kubernetes model.

Return type:

V1WindowsSecurityContextOptions

Module contents

Kubernetes adapter.