# cloudcoil
🚀 Cloud native operations made beautifully simple with Python
[](https://pypi.python.org/pypi/cloudcoil)
[](https://pepy.tech/project/cloudcoil)
[](https://opensource.org/license/apache-2-0/)
[](https://github.com/cloudcoil/cloudcoil/actions/workflows/ci.yml)
> Modern, async-first Kubernetes client with elegant Pythonic syntax and full type safety
## ✨ Features
- 🔥 **Elegant, Pythonic API** - Feels natural to Python developers including fluent and context manager style resource builders
- ⚡ **Async First** - Native async/await support for high performance
- 🛡️ **Type Safe** - Full mypy support and runtime validation
- 🧪 **Testing Ready** - Built-in pytest fixtures for K8s integration tests
- 📦 **Zero Config** - Works with your existing kubeconfig
- 🪶 **Minimal Dependencies** - Only requires httpx, pydantic, and pyyaml
## 🔧 Installation
> [!NOTE]
> For versioning information and compatibility, see the [Versioning Guide](https://github.com/cloudcoil/cloudcoil/blob/main/VERSIONING.md).
Using [uv](https://github.com/astral-sh/uv) (recommended):
```bash
# Install with Kubernetes support
uv add cloudcoil[kubernetes]
# Install with specific Kubernetes version compatibility
uv add cloudcoil[kubernetes-1-29]
uv add cloudcoil[kubernetes-1-30]
uv add cloudcoil[kubernetes-1-31]
uv add cloudcoil[kubernetes-1-32]
```
Using pip:
```bash
pip install cloudcoil[kubernetes]
```
## 🔌 Integrations
Discover more Cloudcoil model integrations for popular Kubernetes operators and CRDs at [cloudcoil-models on GitHub](https://github.com/topics/cloudcoil-models).
Current first-class integrations include:
| Name | Github | PyPI |
| ------- | ------- | ------- |
| [cert-manager](https://github.com/cert-manager/cert-manager) | [models-cert-manager](https://github.com/cloudcoil/models-cert-manager) | [cloudcoil.models.cert_manager](https://pypi.org/project/cloudcoil.models.cert-manager) |
| [fluxcd](https://github.com/fluxcd/flux2) | [models-fluxcd](https://github.com/cloudcoil/models-fluxcd) | [cloudcoil.models.fluxcd](https://pypi.org/project/cloudcoil.models.fluxcd) |
| [istio](https://github.com/istio/istio) | [models-istio](https://github.com/cloudcoil/models-istio) | [cloudcoil.models.istio](https://pypi.org/project/cloudcoil.models.istio) |
| [kyverno](https://github.com/kyverno/kyverno) | [models-kyverno](https://github.com/cloudcoil/models-kyverno) | [cloudcoil.models.kyverno](https://pypi.org/project/cloudcoil.models.kyverno) |
| [prometheus-operator](https://github.com/prometheus-operator/prometheus-operator) | [models-kyverno](https://github.com/cloudcoil/models-prometheus-operator) | [cloudcoil.models.prometheus-operator](https://pypi.org/project/cloudcoil.models.prometheus_operator) |
You can install these integrations using
```bash
uv add cloudcoil[kyverno]
# You can also install multiple dependencies at once
uv add cloudcoil[cert-manager,fluxcd,kyverno]
```
> Missing an integration you need? [Open a model request](https://github.com/cloudcoil/cloudcoil/issues/new?template=%F0%9F%94%8C-model-request.md) to suggest a new integration!
## 💡 Examples
### Reading Resources
```python
from cloudcoil.client import Config
import cloudcoil.models.kubernetes as k8s
# Get a resource - as simple as that!
service = k8s.core.v1.Service.get("kubernetes")
# List resources with elegant pagination
for pod in k8s.core.v1.Pod.list(namespace="default"):
print(f"Found pod: {pod.metadata.name}")
# Async support out of the box
async for pod in await k8s.core.v1.Pod.async_list():
print(f"Found pod: {pod.metadata.name}")
```
### Building resources
#### Using Models
```python
from cloudcoil import apimachinery
import cloudcoil.models.kubernetes.core.v1 as k8score
import cloudcoil.models.kubernetes.apps.v1 as k8sapps
# Create a Deployment
deployment = k8sapps.Deployment(
metadata=apimachinery.ObjectMeta(name="nginx"),
spec=k8sapps.DeploymentSpec(
replicas=3,
selector=apimachinery.LabelSelector(
match_labels={"app": "nginx"}
),
template=k8score.PodTemplateSpec(
metadata=apimachinery.ObjectMeta(
labels={"app": "nginx"}
),
spec=k8score.PodSpec(
containers=[
k8score.Container(
name="nginx",
image="nginx:latest",
ports=[k8score.ContainerPort(container_port=80)]
)
]
)
)
)
).create()
# Create a Service
service = k8score.Service(
metadata=apimachinery.ObjectMeta(name="nginx"),
spec=k8score.ServiceSpec(
selector={"app": "nginx"},
ports=[k8score.ServicePort(port=80, target_port=80)]
)
).create()
# List Deployments
for deploy in k8sapps.Deployment.list():
print(f"Found deployment: {deploy.metadata.name}")
# Update a Deployment
deployment.spec.replicas = 5
deployment.save()
# Delete resources
k8score.Service.delete("nginx")
k8sapps.Deployment.delete("nginx")
```
#### Using the Fluent Builder API
Cloudcoil provides a powerful fluent builder API for Kubernetes resources with full IDE support and rich autocomplete capabilities:
```python
from cloudcoil.models.kubernetes.apps.v1 import Deployment
from cloudcoil.models.kubernetes.core.v1 import Service
# Create a Deployment using the fluent builder
# The fluent style is great for one-liners and simple configurations
nginx_deployment = (
Deployment.builder()
# Metadata can be configured in a single chain for simple objects
.metadata(lambda metadata: metadata
.name("nginx")
.namespace("default")
)
# Complex nested structures can be built using nested lambda functions
.spec(lambda deployment_spec: deployment_spec
.replicas(3)
# Each level of nesting gets its own lambda for clarity
.selector(lambda label_selector: label_selector
.match_labels({"app": "nginx"})
)
.template(lambda pod_template: pod_template
.metadata(lambda pod_metadata: pod_metadata
.labels({"app": "nginx"})
)
.spec(lambda pod_spec: pod_spec
# Lists can be built using array literals with lambda items
.containers([
lambda container: container
.name("nginx")
.image("nginx:latest")
# Nested collections can use the add() helper
.ports(lambda port_list: port_list.add(
lambda port: port.container_port(80)
))
])
)
)
)
.build()
)
# Create a Service using the builder
service = (
Service.builder()
.metadata(lambda m: m
.name("nginx")
.namespace("default")
)
.spec(lambda s: s
.selector({"app": "nginx"})
.ports(lambda ports: ports.add(lambda p: p.container_port(80)))
)
.build()
)
```
The fluent builder provides:
- ✨ Full IDE support with detailed type information
- 🔍 Rich autocomplete for all fields and nested objects
- ⚡ Compile-time validation of your configuration
- 🎯 Clear and chainable API that guides you through resource creation
#### Using the Context Manager Builder API
For complex nested resources, Cloudcoil also provides a context manager-based builder pattern that can make the structure more clear:
```python
from cloudcoil.models.kubernetes.apps.v1 import Deployment
from cloudcoil.models.kubernetes.core.v1 import Service
# Create a deployment using context managers
# Context managers are ideal for deeply nested structures
with Deployment.new() as nginx_deployment:
# Each context creates a clear visual scope
with nginx_deployment.metadata() as deployment_metadata:
deployment_metadata.name("nginx")
deployment_metadata.namespace("default")
with nginx_deployment.spec() as deployment_spec:
# Simple fields can be set directly
deployment_spec.replicas(3)
# Each nested object gets its own context
with deployment_spec.selector() as label_selector:
label_selector.match_labels({"app": "nginx"})
with deployment_spec.template() as pod_template:
with pod_template.metadata() as pod_metadata:
pod_metadata.labels({"app": "nginx"})
with pod_template.spec() as pod_spec:
# Collections use a parent context for the list
with pod_spec.containers() as container_list:
# And child contexts for each item
with container_list.add() as nginx_container:
nginx_container.name("nginx")
nginx_container.image("nginx:latest")
# Ports can be added one by one
with nginx_container.add_port() as container_port:
container_port.container_port(80)
final_deployment = nginx_deployment.build()
# Create a service using context managers
with Service.new() as nginx_service:
# Context managers make the structure very clear
with nginx_service.metadata() as service_metadata:
service_metadata.name("nginx")
service_metadata.namespace("default")
with nginx_service.spec() as service_spec:
# Simple fields can still be set directly
service_spec.selector({"app": "nginx"})
# Port configuration is more readable with contexts
with service_spec.add_port() as service_port:
service_port.port(80)
service_port.target_port(80)
final_service = nginx_service.build()
```
The context manager builder provides:
- 🎭 Clear visual nesting of resource structure
- 🔒 Automatic resource cleanup
- 🎯 Familiar Python context manager pattern
- ✨ Same great IDE support as the fluent builder
#### Mixing Builder Styles
CloudCoil's intelligent builder system automatically detects which style you're using and provides appropriate IDE support:
```python
from cloudcoil.models.kubernetes.apps.v1 import Deployment
from cloudcoil import apimachinery
# Mixing styles lets you choose the best approach for each part
# The IDE automatically adapts to your chosen style at each level
with Deployment.new() as nginx_deployment:
# Direct object initialization with full type checking
nginx_deployment.metadata(apimachinery.ObjectMeta(
name="nginx",
namespace="default",
labels={"app": "nginx"}
))
with nginx_deployment.spec() as deployment_spec:
# IDE shows all available fields with types
deployment_spec.replicas(3)
# Fluent style with rich autocomplete
deployment_spec.selector(lambda sel: sel.match_labels({"app": "nginx"}))
# Context manager style with full type hints
with deployment_spec.template() as pod_template:
# Mix and match freely - IDE adjusts automatically
pod_template.metadata(apimachinery.ObjectMeta(labels={"app": "nginx"}))
with pod_template.spec() as pod_spec:
with pod_spec.containers() as container_list:
with container_list.add() as nginx_container:
# Complete IDE support regardless of style
nginx_container.name("nginx")
nginx_container.image("nginx:latest")
# Switch styles any time
nginx_container.ports(lambda ports: ports
.add(lambda p: p.container_port(80))
.add(lambda p: p.container_port(443))
)
final_deployment = nginx_deployment.build()
```
This flexibility allows you to:
- 🔀 Choose the most appropriate style for each part of your configuration
- 📖 Maximize readability for both simple and complex structures
- 🎨 Format your code according to your team's preferences
- 🧠 Get full IDE support with automatic style detection
- ✨ Enjoy rich autocomplete in all styles
- ⚡ Benefit from type checking across mixed styles
- 🎯 Receive immediate feedback on type errors
- 🔍 See documentation for all fields regardless of style
### Creating Resources
```python
# Create with Pythonic syntax
namespace = k8s.core.v1.Namespace(
metadata=dict(name="dev")
).create()
# Generate names automatically
test_ns = k8s.core.v1.Namespace(
metadata=dict(generate_name="test-")
).create()
```
### Modifying Resources
```python
# Update resources fluently
deployment = k8s.apps.v1.Deployment.get("web")
deployment.spec.replicas = 3
deployment.update()
# Or use the save method which handles both create and update
configmap = k8s.core.v1.ConfigMap(
metadata=dict(name="config"),
data={"key": "value"}
)
configmap.save() # Creates the ConfigMap
configmap.data["key"] = "new-value"
configmap.save() # Updates the ConfigMap
```
### Deleting Resources
```python
# Delete by name
k8s.core.v1.Pod.delete("nginx", namespace="default")
# Or remove the resource instance
pod = k8s.core.v1.Pod.get("nginx")
pod.remove()
```
### Watching Resources
```python
for event_type, resource in k8s.core.v1.Pod.watch(field_selector="metadata.name=mypod"):
# Wait for the pod to be deleted
if event_type == "DELETED":
break
# You can also use the async watch
async for event_type, resource in await k8s.core.v1.Pod.async_watch(field_selector="metadata.name=mypod"):
# Wait for the pod to be deleted
if event_type == "DELETED":
break
```
### Waiting for Resources
```python
# Wait for a resource to reach a desired state
pod = k8s.core.v1.Pod.get("nginx")
pod.wait_for(lambda _, pod: pod.status.phase == "Running", timeout=300)
# You can also check of the resource to be deleted
await pod.async_wait_for(lambda event, _: event == "DELETED", timeout=300)
# You can also supply multiple conditions. The wait will end when the first condition is met.
# It will also return the key of the condition that was met.
test_pod = k8s.core.v1.Pod.get("tests")
status = await test_pod.async_wait_for({
"succeeded": lambda _, pod: pod.status.phase == "Succeeded",
"failed": lambda _, pod: pod.status.phase == "Failed"
}, timeout=300)
assert status == "succeeded"
```
### Dynamic Resources
```python
from cloudcoil.resources import get_dynamic_resource
# Get a dynamic resource class for any CRD or resource without a model
DynamicJob = get_dynamic_resource("Job", "batch/v1")
# Create using dictionary syntax
job = DynamicJob(
metadata={"name": "dynamic-job"},
spec={
"template": {
"spec": {
"containers": [{"name": "job", "image": "busybox"}],
"restartPolicy": "Never"
}
}
}
)
# Create on the cluster
created = job.create()
# Access fields using dict-like syntax
assert created["spec"]["template"]["spec"]["containers"][0]["image"] == "busybox"
# Update fields
created["spec"]["template"]["spec"]["containers"][0]["image"] = "alpine"
updated = created.update()
# Get raw dictionary representation
raw_dict = updated.raw
```
### Resource Parsing
```python
from cloudcoil import resources
# Parse YAML files
deployment = resources.parse_file("deployment.yaml")
# Parse multiple resources
resources = resources.parse_file("k8s-manifests.yaml", load_all=True)
# Get resource class by GVK if its an existing resource model class
Job = resources.get_model("Job", api_version="batch/v1")
```
### Context Management
```python
# Temporarily switch namespace
with Config(namespace="kube-system"):
pods = k8s.core.v1.Pod.list()
# Custom configs
with Config(kubeconfig="dev-cluster.yaml"):
services = k8s.core.v1.Service.list()
```
## 🧪 Testing Integration
Cloudcoil provides powerful pytest fixtures for Kubernetes integration testing:
### Installation
> uv add cloudcoil[test]
### Basic Usage
```python
import pytest
from cloudcoil.models.kubernetes import core, apps
@pytest.mark.configure_test_cluster
def test_deployment(test_config):
with test_config:
# Creates a fresh k3d cluster for testing
deployment = apps.v1.Deployment.get("app")
assert deployment.spec.replicas == 3
```
### Advanced Configuration
```python
@pytest.mark.configure_test_cluster(
cluster_name="my-test-cluster", # Custom cluster name
k3d_version="v5.7.5", # Specific k3d version
k8s_version="v1.31.4", # Specific K8s version
k8s_image="custom/k3s:latest", # Custom K3s image
remove=True # Auto-remove cluster after tests
)
async def test_advanced(test_config):
with test_config:
# Async operations work too!
service = await core.v1.Service.async_get("kubernetes")
assert service.spec.type == "ClusterIP"
```
### Shared Clusters
Reuse clusters across tests for better performance:
```python
@pytest.mark.configure_test_cluster(
cluster_name="shared-cluster",
remove=False # Keep cluster after tests
)
def test_first(test_config):
with test_config:
# Uses existing cluster if available
namespace = core.v1.Namespace.get("default")
assert namespace.status.phase == "Active"
@pytest.mark.configure_test_cluster(
cluster_name="shared-cluster", # Same cluster name
remove=True # Last test removes the cluster
)
def test_second(test_config):
with test_config:
# Uses same cluster as previous test
pods = core.v1.Pod.list(namespace="kube-system")
assert len(pods) > 0
```
### Parallel Testing
The fixtures are compatible with pytest-xdist for parallel testing:
```bash
# Run tests in parallel
pytest -n auto tests/
# Or specify number of workers
pytest -n 4 tests/
```
### Testing Fixtures API
The testing module provides two main fixtures:
- `test_cluster`: Creates and manages k3d clusters
- Returns path to kubeconfig file
- Handles cluster lifecycle
- Supports cluster reuse
- Compatible with parallel testing
- `test_config`: Provides configured `Config` instance
- Uses test cluster kubeconfig
- Manages client connections
- Handles cleanup automatically
- Context manager support
## 🛡️ MyPy Integration
cloudcoil provides a mypy plugin that enables type checking for dynamically loaded kinds from the scheme. To enable the plugin, add this to your pyproject.toml:
```toml
# pyproject.toml
[tool.mypy]
plugins = ['cloudcoil.mypy']
```
This plugin enables full type checking for scheme.get() calls when the kind name is a string literal:
```py
from cloudcoil import resources
# This will be correctly typed as k8s.batch.v1.Job
job_class = resources.get_model("Job")
# Type checking works on the returned class
job = job_class(
metadata={"name": "test"}, # type checked!
spec={
"template": {
"spec": {
"containers": [{"name": "test", "image": "test"}],
"restartPolicy": "Never"
}
}
}
)
```
## 🏗️ Model Generation
Cloudcoil supports generating typed models from CustomResourceDefinitions (CRDs). You can either use the provided cookiecutter template or set up model generation manually.
### Using the Cookiecutter Template
The fastest way to get started is using our cookiecutter template: [cloudcoil-models-cookiecutter](https://github.com/cloudcoil/cloudcoil/tree/main/cookiecutter)
### Codegen Config
Cloudcoil includes a CLI tool, cloudcoil-model-codegen, which reads configuration from your pyproject.toml under [tool.cloudcoil.codegen.models]. It supports options such as:
• namespace: The Python package name for generated models
• input: Path or URL to CRD (YAML/JSON) or OpenAPI schema
• output: Output directory for the generated code
• mode: Either "resource" (default) or "base" for the generated class hierarchy
• crd-namespace: Inject a namespace for CRD resources
• transformations / updates: Modify the schema before generation
• exclude-unknown: Exclude definitions that cannot be mapped
• aliases: Aliases for properties
• additional-datamodel-codegen-args: Pass extra flags to the underlying generator
Example pyproject.toml config -
```toml
[[tool.cloudcoil.codegen.models]]
# Unique name for the models
# This will be used as the name for the setuptools entrypoints
namespace = "cloudcoil.models.fluxcd"
input = "https://github.com/fluxcd/flux2/releases/download/v2.4.0/install.yaml"
crd-namespace = "io.fluxcd.toolkit"
```
For more examples, check out the [cloudcoil-models](https://github.com/topics/cloudcoil-models) topic on Github.
If you are building a models package to be used with cloudcoil, please make sure to tag it with this topic for discovery.
## 📚 Documentation
For complete documentation, visit [cloudcoil.github.io/cloudcoil](https://cloudcoil.github.io/cloudcoil)
## 📜 License
Apache License, Version 2.0 - see [LICENSE](LICENSE)
Raw data
{
"_id": null,
"home_page": null,
"name": "cloudcoil",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.10",
"maintainer_email": "Sambhav Kothari <sambhavs.email@gmail.com>",
"keywords": "async, cloud-native, kubernetes, pydantic, python",
"author": null,
"author_email": "Sambhav Kothari <sambhavs.email@gmail.com>",
"download_url": "https://files.pythonhosted.org/packages/b1/36/d84b26393b75712d498bcd1ec6963bffc10291a03109d716a052feb2e9be/cloudcoil-0.4.1.tar.gz",
"platform": null,
"description": "# cloudcoil\n\n\ud83d\ude80 Cloud native operations made beautifully simple with Python\n\n[](https://pypi.python.org/pypi/cloudcoil)\n[](https://pepy.tech/project/cloudcoil)\n[](https://opensource.org/license/apache-2-0/)\n[](https://github.com/cloudcoil/cloudcoil/actions/workflows/ci.yml)\n\n> Modern, async-first Kubernetes client with elegant Pythonic syntax and full type safety\n\n## \u2728 Features\n\n- \ud83d\udd25 **Elegant, Pythonic API** - Feels natural to Python developers including fluent and context manager style resource builders\n- \u26a1 **Async First** - Native async/await support for high performance\n- \ud83d\udee1\ufe0f **Type Safe** - Full mypy support and runtime validation\n- \ud83e\uddea **Testing Ready** - Built-in pytest fixtures for K8s integration tests\n- \ud83d\udce6 **Zero Config** - Works with your existing kubeconfig\n- \ud83e\udeb6 **Minimal Dependencies** - Only requires httpx, pydantic, and pyyaml\n\n## \ud83d\udd27 Installation\n\n> [!NOTE]\n> For versioning information and compatibility, see the [Versioning Guide](https://github.com/cloudcoil/cloudcoil/blob/main/VERSIONING.md).\n\nUsing [uv](https://github.com/astral-sh/uv) (recommended):\n\n```bash\n# Install with Kubernetes support\nuv add cloudcoil[kubernetes]\n\n# Install with specific Kubernetes version compatibility\nuv add cloudcoil[kubernetes-1-29]\nuv add cloudcoil[kubernetes-1-30]\nuv add cloudcoil[kubernetes-1-31]\nuv add cloudcoil[kubernetes-1-32]\n```\n\nUsing pip:\n\n```bash\npip install cloudcoil[kubernetes]\n```\n\n## \ud83d\udd0c Integrations\n\nDiscover more Cloudcoil model integrations for popular Kubernetes operators and CRDs at [cloudcoil-models on GitHub](https://github.com/topics/cloudcoil-models).\n\nCurrent first-class integrations include:\n\n| Name | Github | PyPI | \n| ------- | ------- | ------- | \n| [cert-manager](https://github.com/cert-manager/cert-manager) | [models-cert-manager](https://github.com/cloudcoil/models-cert-manager) | [cloudcoil.models.cert_manager](https://pypi.org/project/cloudcoil.models.cert-manager) |\n| [fluxcd](https://github.com/fluxcd/flux2) | [models-fluxcd](https://github.com/cloudcoil/models-fluxcd) | [cloudcoil.models.fluxcd](https://pypi.org/project/cloudcoil.models.fluxcd) |\n| [istio](https://github.com/istio/istio) | [models-istio](https://github.com/cloudcoil/models-istio) | [cloudcoil.models.istio](https://pypi.org/project/cloudcoil.models.istio) |\n| [kyverno](https://github.com/kyverno/kyverno) | [models-kyverno](https://github.com/cloudcoil/models-kyverno) | [cloudcoil.models.kyverno](https://pypi.org/project/cloudcoil.models.kyverno) |\n| [prometheus-operator](https://github.com/prometheus-operator/prometheus-operator) | [models-kyverno](https://github.com/cloudcoil/models-prometheus-operator) | [cloudcoil.models.prometheus-operator](https://pypi.org/project/cloudcoil.models.prometheus_operator) |\n\n\nYou can install these integrations using\n\n```bash\nuv add cloudcoil[kyverno]\n# You can also install multiple dependencies at once\nuv add cloudcoil[cert-manager,fluxcd,kyverno]\n```\n\n> Missing an integration you need? [Open a model request](https://github.com/cloudcoil/cloudcoil/issues/new?template=%F0%9F%94%8C-model-request.md) to suggest a new integration!\n\n## \ud83d\udca1 Examples\n\n### Reading Resources\n\n```python\nfrom cloudcoil.client import Config\nimport cloudcoil.models.kubernetes as k8s\n\n# Get a resource - as simple as that!\nservice = k8s.core.v1.Service.get(\"kubernetes\")\n\n# List resources with elegant pagination\nfor pod in k8s.core.v1.Pod.list(namespace=\"default\"):\n print(f\"Found pod: {pod.metadata.name}\")\n\n# Async support out of the box\nasync for pod in await k8s.core.v1.Pod.async_list():\n print(f\"Found pod: {pod.metadata.name}\")\n```\n### Building resources\n\n#### Using Models\n\n```python\nfrom cloudcoil import apimachinery\nimport cloudcoil.models.kubernetes.core.v1 as k8score\nimport cloudcoil.models.kubernetes.apps.v1 as k8sapps\n\n# Create a Deployment\ndeployment = k8sapps.Deployment(\n metadata=apimachinery.ObjectMeta(name=\"nginx\"),\n spec=k8sapps.DeploymentSpec(\n replicas=3,\n selector=apimachinery.LabelSelector(\n match_labels={\"app\": \"nginx\"}\n ),\n template=k8score.PodTemplateSpec(\n metadata=apimachinery.ObjectMeta(\n labels={\"app\": \"nginx\"}\n ),\n spec=k8score.PodSpec(\n containers=[\n k8score.Container(\n name=\"nginx\",\n image=\"nginx:latest\",\n ports=[k8score.ContainerPort(container_port=80)]\n )\n ]\n )\n )\n )\n).create()\n\n# Create a Service\nservice = k8score.Service(\n metadata=apimachinery.ObjectMeta(name=\"nginx\"),\n spec=k8score.ServiceSpec(\n selector={\"app\": \"nginx\"},\n ports=[k8score.ServicePort(port=80, target_port=80)]\n )\n).create()\n\n# List Deployments\nfor deploy in k8sapps.Deployment.list():\n print(f\"Found deployment: {deploy.metadata.name}\")\n\n# Update a Deployment\ndeployment.spec.replicas = 5\ndeployment.save()\n\n# Delete resources\nk8score.Service.delete(\"nginx\")\nk8sapps.Deployment.delete(\"nginx\")\n```\n\n#### Using the Fluent Builder API\n\nCloudcoil provides a powerful fluent builder API for Kubernetes resources with full IDE support and rich autocomplete capabilities:\n\n```python\nfrom cloudcoil.models.kubernetes.apps.v1 import Deployment\nfrom cloudcoil.models.kubernetes.core.v1 import Service\n\n# Create a Deployment using the fluent builder\n# The fluent style is great for one-liners and simple configurations\nnginx_deployment = (\n Deployment.builder()\n # Metadata can be configured in a single chain for simple objects\n .metadata(lambda metadata: metadata\n .name(\"nginx\")\n .namespace(\"default\")\n )\n # Complex nested structures can be built using nested lambda functions\n .spec(lambda deployment_spec: deployment_spec\n .replicas(3)\n # Each level of nesting gets its own lambda for clarity\n .selector(lambda label_selector: label_selector\n .match_labels({\"app\": \"nginx\"})\n )\n .template(lambda pod_template: pod_template\n .metadata(lambda pod_metadata: pod_metadata\n .labels({\"app\": \"nginx\"})\n )\n .spec(lambda pod_spec: pod_spec\n # Lists can be built using array literals with lambda items\n .containers([\n lambda container: container\n .name(\"nginx\")\n .image(\"nginx:latest\")\n # Nested collections can use the add() helper\n .ports(lambda port_list: port_list.add(\n lambda port: port.container_port(80)\n ))\n ])\n )\n )\n )\n .build()\n)\n\n# Create a Service using the builder\nservice = (\n Service.builder()\n .metadata(lambda m: m\n .name(\"nginx\")\n .namespace(\"default\")\n )\n .spec(lambda s: s\n .selector({\"app\": \"nginx\"})\n .ports(lambda ports: ports.add(lambda p: p.container_port(80)))\n )\n .build()\n)\n```\n\nThe fluent builder provides:\n- \u2728 Full IDE support with detailed type information\n- \ud83d\udd0d Rich autocomplete for all fields and nested objects\n- \u26a1 Compile-time validation of your configuration\n- \ud83c\udfaf Clear and chainable API that guides you through resource creation\n\n#### Using the Context Manager Builder API\n\nFor complex nested resources, Cloudcoil also provides a context manager-based builder pattern that can make the structure more clear:\n\n```python\nfrom cloudcoil.models.kubernetes.apps.v1 import Deployment\nfrom cloudcoil.models.kubernetes.core.v1 import Service\n\n# Create a deployment using context managers\n# Context managers are ideal for deeply nested structures\nwith Deployment.new() as nginx_deployment:\n # Each context creates a clear visual scope\n with nginx_deployment.metadata() as deployment_metadata:\n deployment_metadata.name(\"nginx\")\n deployment_metadata.namespace(\"default\")\n \n with nginx_deployment.spec() as deployment_spec:\n # Simple fields can be set directly\n deployment_spec.replicas(3)\n \n # Each nested object gets its own context\n with deployment_spec.selector() as label_selector:\n label_selector.match_labels({\"app\": \"nginx\"})\n \n with deployment_spec.template() as pod_template:\n with pod_template.metadata() as pod_metadata:\n pod_metadata.labels({\"app\": \"nginx\"})\n \n with pod_template.spec() as pod_spec:\n # Collections use a parent context for the list\n with pod_spec.containers() as container_list:\n # And child contexts for each item\n with container_list.add() as nginx_container:\n nginx_container.name(\"nginx\")\n nginx_container.image(\"nginx:latest\")\n # Ports can be added one by one\n with nginx_container.add_port() as container_port:\n container_port.container_port(80)\n\nfinal_deployment = nginx_deployment.build()\n\n# Create a service using context managers\nwith Service.new() as nginx_service:\n # Context managers make the structure very clear\n with nginx_service.metadata() as service_metadata:\n service_metadata.name(\"nginx\")\n service_metadata.namespace(\"default\")\n \n with nginx_service.spec() as service_spec:\n # Simple fields can still be set directly\n service_spec.selector({\"app\": \"nginx\"})\n # Port configuration is more readable with contexts\n with service_spec.add_port() as service_port:\n service_port.port(80)\n service_port.target_port(80)\n\nfinal_service = nginx_service.build()\n```\n\nThe context manager builder provides:\n- \ud83c\udfad Clear visual nesting of resource structure\n- \ud83d\udd12 Automatic resource cleanup\n- \ud83c\udfaf Familiar Python context manager pattern\n- \u2728 Same great IDE support as the fluent builder\n\n#### Mixing Builder Styles\n\nCloudCoil's intelligent builder system automatically detects which style you're using and provides appropriate IDE support:\n\n```python\nfrom cloudcoil.models.kubernetes.apps.v1 import Deployment\nfrom cloudcoil import apimachinery\n\n# Mixing styles lets you choose the best approach for each part\n# The IDE automatically adapts to your chosen style at each level\nwith Deployment.new() as nginx_deployment:\n # Direct object initialization with full type checking\n nginx_deployment.metadata(apimachinery.ObjectMeta(\n name=\"nginx\",\n namespace=\"default\",\n labels={\"app\": \"nginx\"}\n ))\n \n with nginx_deployment.spec() as deployment_spec:\n # IDE shows all available fields with types\n deployment_spec.replicas(3)\n # Fluent style with rich autocomplete\n deployment_spec.selector(lambda sel: sel.match_labels({\"app\": \"nginx\"}))\n \n # Context manager style with full type hints\n with deployment_spec.template() as pod_template:\n # Mix and match freely - IDE adjusts automatically\n pod_template.metadata(apimachinery.ObjectMeta(labels={\"app\": \"nginx\"}))\n with pod_template.spec() as pod_spec:\n with pod_spec.containers() as container_list:\n with container_list.add() as nginx_container:\n # Complete IDE support regardless of style\n nginx_container.name(\"nginx\")\n nginx_container.image(\"nginx:latest\")\n # Switch styles any time\n nginx_container.ports(lambda ports: ports\n .add(lambda p: p.container_port(80))\n .add(lambda p: p.container_port(443))\n )\n\nfinal_deployment = nginx_deployment.build()\n```\n\nThis flexibility allows you to:\n- \ud83d\udd00 Choose the most appropriate style for each part of your configuration\n- \ud83d\udcd6 Maximize readability for both simple and complex structures\n- \ud83c\udfa8 Format your code according to your team's preferences\n- \ud83e\udde0 Get full IDE support with automatic style detection\n- \u2728 Enjoy rich autocomplete in all styles\n- \u26a1 Benefit from type checking across mixed styles\n- \ud83c\udfaf Receive immediate feedback on type errors\n- \ud83d\udd0d See documentation for all fields regardless of style\n\n\n### Creating Resources\n\n```python\n# Create with Pythonic syntax\nnamespace = k8s.core.v1.Namespace(\n metadata=dict(name=\"dev\")\n).create()\n\n# Generate names automatically\ntest_ns = k8s.core.v1.Namespace(\n metadata=dict(generate_name=\"test-\")\n).create()\n```\n\n### Modifying Resources\n\n```python\n# Update resources fluently\ndeployment = k8s.apps.v1.Deployment.get(\"web\")\ndeployment.spec.replicas = 3\ndeployment.update()\n\n# Or use the save method which handles both create and update\nconfigmap = k8s.core.v1.ConfigMap(\n metadata=dict(name=\"config\"),\n data={\"key\": \"value\"}\n)\nconfigmap.save() # Creates the ConfigMap\n\nconfigmap.data[\"key\"] = \"new-value\"\nconfigmap.save() # Updates the ConfigMap\n```\n\n### Deleting Resources\n\n```python\n# Delete by name\nk8s.core.v1.Pod.delete(\"nginx\", namespace=\"default\")\n\n# Or remove the resource instance\npod = k8s.core.v1.Pod.get(\"nginx\")\npod.remove()\n```\n\n### Watching Resources\n\n```python\nfor event_type, resource in k8s.core.v1.Pod.watch(field_selector=\"metadata.name=mypod\"):\n # Wait for the pod to be deleted\n if event_type == \"DELETED\":\n break\n\n# You can also use the async watch\nasync for event_type, resource in await k8s.core.v1.Pod.async_watch(field_selector=\"metadata.name=mypod\"):\n # Wait for the pod to be deleted\n if event_type == \"DELETED\":\n break\n```\n\n### Waiting for Resources\n\n```python\n# Wait for a resource to reach a desired state\npod = k8s.core.v1.Pod.get(\"nginx\")\npod.wait_for(lambda _, pod: pod.status.phase == \"Running\", timeout=300)\n\n# You can also check of the resource to be deleted\nawait pod.async_wait_for(lambda event, _: event == \"DELETED\", timeout=300)\n\n# You can also supply multiple conditions. The wait will end when the first condition is met.\n# It will also return the key of the condition that was met.\ntest_pod = k8s.core.v1.Pod.get(\"tests\")\nstatus = await test_pod.async_wait_for({\n \"succeeded\": lambda _, pod: pod.status.phase == \"Succeeded\",\n \"failed\": lambda _, pod: pod.status.phase == \"Failed\"\n }, timeout=300)\nassert status == \"succeeded\"\n```\n\n### Dynamic Resources\n\n```python\nfrom cloudcoil.resources import get_dynamic_resource\n\n# Get a dynamic resource class for any CRD or resource without a model\nDynamicJob = get_dynamic_resource(\"Job\", \"batch/v1\")\n\n# Create using dictionary syntax\njob = DynamicJob(\n metadata={\"name\": \"dynamic-job\"},\n spec={\n \"template\": {\n \"spec\": {\n \"containers\": [{\"name\": \"job\", \"image\": \"busybox\"}],\n \"restartPolicy\": \"Never\"\n }\n }\n }\n)\n\n# Create on the cluster\ncreated = job.create()\n\n# Access fields using dict-like syntax\nassert created[\"spec\"][\"template\"][\"spec\"][\"containers\"][0][\"image\"] == \"busybox\"\n\n# Update fields\ncreated[\"spec\"][\"template\"][\"spec\"][\"containers\"][0][\"image\"] = \"alpine\"\nupdated = created.update()\n\n# Get raw dictionary representation\nraw_dict = updated.raw\n```\n\n### Resource Parsing\n\n```python\nfrom cloudcoil import resources\n\n# Parse YAML files\ndeployment = resources.parse_file(\"deployment.yaml\")\n\n# Parse multiple resources\nresources = resources.parse_file(\"k8s-manifests.yaml\", load_all=True)\n\n# Get resource class by GVK if its an existing resource model class\nJob = resources.get_model(\"Job\", api_version=\"batch/v1\")\n```\n\n### Context Management\n\n```python\n# Temporarily switch namespace\nwith Config(namespace=\"kube-system\"):\n pods = k8s.core.v1.Pod.list()\n\n# Custom configs\nwith Config(kubeconfig=\"dev-cluster.yaml\"):\n services = k8s.core.v1.Service.list()\n```\n\n\n## \ud83e\uddea Testing Integration\n\nCloudcoil provides powerful pytest fixtures for Kubernetes integration testing:\n\n### Installation\n\n> uv add cloudcoil[test]\n\n### Basic Usage\n\n```python\nimport pytest\nfrom cloudcoil.models.kubernetes import core, apps\n\n@pytest.mark.configure_test_cluster\ndef test_deployment(test_config):\n with test_config:\n # Creates a fresh k3d cluster for testing\n deployment = apps.v1.Deployment.get(\"app\")\n assert deployment.spec.replicas == 3\n```\n\n### Advanced Configuration\n\n```python\n@pytest.mark.configure_test_cluster(\n cluster_name=\"my-test-cluster\", # Custom cluster name\n k3d_version=\"v5.7.5\", # Specific k3d version\n k8s_version=\"v1.31.4\", # Specific K8s version\n k8s_image=\"custom/k3s:latest\", # Custom K3s image\n remove=True # Auto-remove cluster after tests\n)\nasync def test_advanced(test_config):\n with test_config:\n # Async operations work too!\n service = await core.v1.Service.async_get(\"kubernetes\")\n assert service.spec.type == \"ClusterIP\"\n```\n\n### Shared Clusters\n\nReuse clusters across tests for better performance:\n\n```python\n@pytest.mark.configure_test_cluster(\n cluster_name=\"shared-cluster\",\n remove=False # Keep cluster after tests\n)\ndef test_first(test_config):\n with test_config:\n # Uses existing cluster if available\n namespace = core.v1.Namespace.get(\"default\")\n assert namespace.status.phase == \"Active\"\n\n@pytest.mark.configure_test_cluster(\n cluster_name=\"shared-cluster\", # Same cluster name\n remove=True # Last test removes the cluster\n)\ndef test_second(test_config):\n with test_config:\n # Uses same cluster as previous test\n pods = core.v1.Pod.list(namespace=\"kube-system\")\n assert len(pods) > 0\n```\n\n### Parallel Testing\n\nThe fixtures are compatible with pytest-xdist for parallel testing:\n\n```bash\n# Run tests in parallel\npytest -n auto tests/\n\n# Or specify number of workers\npytest -n 4 tests/\n```\n\n### Testing Fixtures API\n\nThe testing module provides two main fixtures:\n\n- `test_cluster`: Creates and manages k3d clusters\n - Returns path to kubeconfig file\n - Handles cluster lifecycle\n - Supports cluster reuse\n - Compatible with parallel testing\n\n- `test_config`: Provides configured `Config` instance\n - Uses test cluster kubeconfig\n - Manages client connections\n - Handles cleanup automatically\n - Context manager support\n\n## \ud83d\udee1\ufe0f MyPy Integration\n\ncloudcoil provides a mypy plugin that enables type checking for dynamically loaded kinds from the scheme. To enable the plugin, add this to your pyproject.toml:\n\n```toml\n# pyproject.toml\n[tool.mypy]\nplugins = ['cloudcoil.mypy']\n```\n\nThis plugin enables full type checking for scheme.get() calls when the kind name is a string literal:\n\n```py\nfrom cloudcoil import resources\n\n# This will be correctly typed as k8s.batch.v1.Job\njob_class = resources.get_model(\"Job\")\n\n# Type checking works on the returned class\njob = job_class(\n metadata={\"name\": \"test\"}, # type checked!\n spec={\n \"template\": {\n \"spec\": {\n \"containers\": [{\"name\": \"test\", \"image\": \"test\"}],\n \"restartPolicy\": \"Never\"\n }\n }\n }\n)\n```\n\n## \ud83c\udfd7\ufe0f Model Generation\n\nCloudcoil supports generating typed models from CustomResourceDefinitions (CRDs). You can either use the provided cookiecutter template or set up model generation manually.\n\n### Using the Cookiecutter Template\n\nThe fastest way to get started is using our cookiecutter template: [cloudcoil-models-cookiecutter](https://github.com/cloudcoil/cloudcoil/tree/main/cookiecutter)\n\n### Codegen Config\n\nCloudcoil includes a CLI tool, cloudcoil-model-codegen, which reads configuration from your pyproject.toml under [tool.cloudcoil.codegen.models]. It supports options such as:\n\n\u2022 namespace: The Python package name for generated models \n\u2022 input: Path or URL to CRD (YAML/JSON) or OpenAPI schema \n\u2022 output: Output directory for the generated code \n\u2022 mode: Either \"resource\" (default) or \"base\" for the generated class hierarchy \n\u2022 crd-namespace: Inject a namespace for CRD resources \n\u2022 transformations / updates: Modify the schema before generation \n\u2022 exclude-unknown: Exclude definitions that cannot be mapped \n\u2022 aliases: Aliases for properties\n\u2022 additional-datamodel-codegen-args: Pass extra flags to the underlying generator \n\nExample pyproject.toml config - \n\n```toml\n[[tool.cloudcoil.codegen.models]]\n# Unique name for the models\n# This will be used as the name for the setuptools entrypoints\nnamespace = \"cloudcoil.models.fluxcd\"\ninput = \"https://github.com/fluxcd/flux2/releases/download/v2.4.0/install.yaml\"\ncrd-namespace = \"io.fluxcd.toolkit\"\n```\n\nFor more examples, check out the [cloudcoil-models](https://github.com/topics/cloudcoil-models) topic on Github.\n\nIf you are building a models package to be used with cloudcoil, please make sure to tag it with this topic for discovery.\n\n## \ud83d\udcda Documentation\n\nFor complete documentation, visit [cloudcoil.github.io/cloudcoil](https://cloudcoil.github.io/cloudcoil)\n\n## \ud83d\udcdc License\n\nApache License, Version 2.0 - see [LICENSE](LICENSE)\n",
"bugtrack_url": null,
"license": "Apache-2.0",
"summary": "Cloud native made easy with Python",
"version": "0.4.1",
"project_urls": {
"Changelog": "https://github.com/cloudcoil/cloudcoil/releases",
"Documentation": "https://cloudcoil.github.io/cloudcoil",
"Homepage": "https://github.com/cloudcoil/cloudcoil",
"Issues": "https://github.com/cloudcoil/cloudcoil/issues",
"Repository": "https://github.com/cloudcoil/cloudcoil"
},
"split_keywords": [
"async",
" cloud-native",
" kubernetes",
" pydantic",
" python"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "bad83ba00cf89837eecfdc7769a67eb0a2cf49ec8dcb7bf723d84f848ccde793",
"md5": "b4f8b070a7710c833335e5df38ad8797",
"sha256": "b9df63bcc490d846b1011c5d3c66d8f691f0de52ae2b4c99478e1b9e967909d1"
},
"downloads": -1,
"filename": "cloudcoil-0.4.1-py3-none-any.whl",
"has_sig": false,
"md5_digest": "b4f8b070a7710c833335e5df38ad8797",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.10",
"size": 67453,
"upload_time": "2025-01-28T22:02:54",
"upload_time_iso_8601": "2025-01-28T22:02:54.357223Z",
"url": "https://files.pythonhosted.org/packages/ba/d8/3ba00cf89837eecfdc7769a67eb0a2cf49ec8dcb7bf723d84f848ccde793/cloudcoil-0.4.1-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "b136d84b26393b75712d498bcd1ec6963bffc10291a03109d716a052feb2e9be",
"md5": "5c9c0cc2cbb7bef9771b59eefdd7f804",
"sha256": "7d6202fc00d94a48522c8bc2e244d4e8af407bf282a0b396c8ff06e4e850a07e"
},
"downloads": -1,
"filename": "cloudcoil-0.4.1.tar.gz",
"has_sig": false,
"md5_digest": "5c9c0cc2cbb7bef9771b59eefdd7f804",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.10",
"size": 339139,
"upload_time": "2025-01-28T22:02:56",
"upload_time_iso_8601": "2025-01-28T22:02:56.381645Z",
"url": "https://files.pythonhosted.org/packages/b1/36/d84b26393b75712d498bcd1ec6963bffc10291a03109d716a052feb2e9be/cloudcoil-0.4.1.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-01-28 22:02:56",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "cloudcoil",
"github_project": "cloudcoil",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "cloudcoil"
}
JSON
