# cloudcoil
Cloud native made easy with Python
### PyPI stats
[](https://pypi.python.org/pypi/cloudcoil)
[](https://github.com/cloudcoil/cloudcoil)
[](https://pepy.tech/project/cloudcoil)
[](https://pepy.tech/project/cloudcoil)
[](https://pepy.tech/project/cloudcoil)
### Repo information
[](https://opensource.org/license/apache-2-0/)
[](https://github.com/cloudcoil/cloudcoil/actions/workflows/ci.yml)
[](https://codecov.io/gh/cloudcoil/cloudcoil)
## Installation
```bash
# Minimal dependencies
# pydantic, httpx and pyyaml
uv add cloudcoil
```
## Quick Start
```python
from pathlib import Path
# Config is the core way to interact with your Kubernetes API Server
from cloudcoil.client import Config
from cloudcoil.client import errors
# All default kubernetes types are neatly arranged
# with appropriate apiversions as module paths
from cloudcoil.kinds.apps import v1 as apps_v1
from cloudcoil.kinds.core import v1 as core_v1
from cloudcoil.kinds.batch import v1 as batch_v1
# Uses the default config based on KUBECONFIG
# Feels just as natural as kubectl
# But comes with full pydantic validation
kubernetes_service = core_v1.Service.get("kubernetes")
# You can create temporary config contexts
# This is similar to doing kubens kube-system
with Config(namespace="kube-system"):
# This searches for deployments in the kube-system namespace
core_dns_deployment = apps_v1.Deployment.get("core-dns")
# Also comes with async client out of the box!
kube_dns_service = await core_v1.Service.async_get("kube-dns")
# Create new objects with generate name easily
test_namespace = core_v1.Namespace(metadata=dict(generate_name="test-")).create()
# You can also modify the object and update it
test_namespace.metadata.labels = {"test": "true"}
# Get the new value of the test namespace back from the server
# after updating it
test_namespace = test_namespace.update()
# You can also easily fetch namespace using
kube_system_namespace = core_v1.Namespace()
kube_system_namespace.name = "kube-system"
# Fetch the latest version of the namespace from the server
kube_system_namespace = kube_system_namespace.fetch()
# We can access the output from the APIServer from the create method
# Switch to the new namespace
with Config(namespace=test_namespace.metadata.name):
try:
core_dns_deployment = apps_v1.Deployment.get("core-dns")
except errors.ResourceNotFound:
pass
# Finally you can remove the namespace
# And also inspect the output to ensure it is terminating
test_namespace.remove().status.phase == "Terminating"
# You can also delete it using the name/namespace if you wish
core_v1.Namespace.delete(name=test_namespace.metadata.name)
# You can also parse kubernetes resource easily
# Let's start by create a default scheme which has all the default kubernetes kinds
# registered with it
from cloudcoil import scheme
# Let's assume we have a hello-world.yaml file that looks like so
# apiVersion: batch/v1
# kind: Job
# metadata:
# name: hello-world
# spec:
# template:
# spec:
# containers:
# - name: hello-world
# image: ubuntu
# command: ["echo", "Hello, World!"]
# restartPolicy: Never
job = scheme.parse_file("hello-world.yaml")
# It is serialized to the correc type
assert isinstance(job, batch_v1.Job)
# You can now create the job
job.create()
# You can also access different registered types from the scheme
Job = scheme.get("Job")
# Now you can parse the file using the from_file classmethod
job = Job.from_file("hello-world.yaml")
# the above is correctly typed with mypy if you use the cloudcoil mypy extension
# even though the class was dynamically loaded from the scheme via a string
# Listing resources
# cloudcoil provides two ways to list resources:
# 1. Iterator API (Recommended)
# This automatically handles pagination and provides a clean interface
for pod in core_v1.Pod.list(all_namespaces=True):
print(pod.metadata.name)
# For async code, you can use async iteration
async for pod in await core_v1.Pod.async_list(all_namespaces=True):
print(pod.metadata.name)
# Get total count for all items
total_pods = len(pods)
print(f"Total pods: {total_pods}")
# 2. Manual Pagination API
# If you need more control over pagination, you can use the raw API
# This returns a ResourceList object that contains the first page
pods = core_v1.Pod.list(namespace="kube-system", limit=10)
print(f"First page has {len(pods.items)} items")
# Access the items directly using the items attribute
for pod in pods.items:
print(pod.metadata.name)
# Check if there are more pages
if pods.has_next_page():
# Get the next page
next_page = pods.get_next_page()
# For async code
next_page = await pods.async_get_next_page()
# It's recommended to use the Iterator API as it handles pagination
# automatically and provides a more ergonomic interface
```
### Testing Integration
cloudcoil includes pytest fixtures to help you test your Kubernetes applications. Install with test dependencies:
```bash
uv add cloudcoil[test]
```
The testing integration provides two key fixtures:
- `test_cluster`: Creates and manages a k3d cluster for testing
- `test_config`: Provides a Config instance configured for the test cluster
Example usage:
```python
import pytest
from cloudcoil.kinds.core import v1 as corev1
@pytest.mark.configure_test_cluster(
cluster_name="my-test-cluster",
k3d_version="v5.7.5",
k8s_version="v1.31.4",
remove=True
)
def test_my_resources(test_config):
with test_config:
namespace = corev1.Namespace.get("default")
assert namespace.metadata.name == "default"
```
#### Test Cluster Configuration
The `configure_test_cluster` mark accepts these arguments:
- `cluster_name`: Name of the test cluster (default: auto-generated)
- `k3d_version`: Version of k3d to use (default: v5.7.5)
- `k8s_version`: Kubernetes version to use (default: v1.31.4)
- `k8s_image`: Custom k3s image (default: rancher/k3s:{k8s_version}-k3s1)
- `remove`: Whether to remove the cluster after tests (default: True)
## 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
[tool.mypy]
plugins = ['cloudcoil.mypy']
```
This plugin enables full type checking for scheme.get() calls when the kind name is a string literal:
```python
from cloudcoil import scheme
# This will be correctly typed as batch_v1.Job
job_class = scheme.get("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"
}
}
}
)
```
## Documentation
For full documentation, please visit [cloudcoil.github.io/cloudcoil](https://cloudcoil.github.io/cloudcoil)
## License
This project is licensed under the Apache License, Version 2.0 - see the [LICENSE](https://github.com/cloudcoil/cloudcoil/blob/main/LICENSE) file for details.
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/48/23/65b62146872ce439b80bfaa88404cfbae461f68589d6b301568c97885616/cloudcoil-0.0.12.tar.gz",
"platform": null,
"description": "# cloudcoil\n\nCloud native made easy with Python\n\n### PyPI stats\n\n[](https://pypi.python.org/pypi/cloudcoil)\n[](https://github.com/cloudcoil/cloudcoil)\n\n[](https://pepy.tech/project/cloudcoil)\n[](https://pepy.tech/project/cloudcoil)\n[](https://pepy.tech/project/cloudcoil)\n\n### Repo information\n\n[](https://opensource.org/license/apache-2-0/)\n[](https://github.com/cloudcoil/cloudcoil/actions/workflows/ci.yml)\n[](https://codecov.io/gh/cloudcoil/cloudcoil)\n\n## Installation\n\n```bash\n# Minimal dependencies\n# pydantic, httpx and pyyaml\nuv add cloudcoil\n```\n\n## Quick Start\n\n```python\nfrom pathlib import Path\n\n# Config is the core way to interact with your Kubernetes API Server\nfrom cloudcoil.client import Config\nfrom cloudcoil.client import errors\n# All default kubernetes types are neatly arranged\n# with appropriate apiversions as module paths\nfrom cloudcoil.kinds.apps import v1 as apps_v1\nfrom cloudcoil.kinds.core import v1 as core_v1\nfrom cloudcoil.kinds.batch import v1 as batch_v1\n\n\n# Uses the default config based on KUBECONFIG\n# Feels just as natural as kubectl\n# But comes with full pydantic validation\nkubernetes_service = core_v1.Service.get(\"kubernetes\")\n\n# You can create temporary config contexts\n# This is similar to doing kubens kube-system\nwith Config(namespace=\"kube-system\"):\n # This searches for deployments in the kube-system namespace\n core_dns_deployment = apps_v1.Deployment.get(\"core-dns\")\n # Also comes with async client out of the box!\n kube_dns_service = await core_v1.Service.async_get(\"kube-dns\")\n\n# Create new objects with generate name easily\ntest_namespace = core_v1.Namespace(metadata=dict(generate_name=\"test-\")).create()\n# You can also modify the object and update it\ntest_namespace.metadata.labels = {\"test\": \"true\"}\n# Get the new value of the test namespace back from the server\n# after updating it\ntest_namespace = test_namespace.update()\n\n# You can also easily fetch namespace using\nkube_system_namespace = core_v1.Namespace()\nkube_system_namespace.name = \"kube-system\"\n# Fetch the latest version of the namespace from the server\nkube_system_namespace = kube_system_namespace.fetch()\n\n# We can access the output from the APIServer from the create method\n# Switch to the new namespace\nwith Config(namespace=test_namespace.metadata.name):\n try:\n core_dns_deployment = apps_v1.Deployment.get(\"core-dns\")\n except errors.ResourceNotFound:\n pass\n\n# Finally you can remove the namespace\n# And also inspect the output to ensure it is terminating\ntest_namespace.remove().status.phase == \"Terminating\"\n# You can also delete it using the name/namespace if you wish\ncore_v1.Namespace.delete(name=test_namespace.metadata.name)\n\n# You can also parse kubernetes resource easily\n# Let's start by create a default scheme which has all the default kubernetes kinds\n# registered with it\nfrom cloudcoil import scheme\n# Let's assume we have a hello-world.yaml file that looks like so\n# apiVersion: batch/v1\n# kind: Job\n# metadata:\n# name: hello-world\n# spec:\n# template:\n# spec:\n# containers:\n# - name: hello-world\n# image: ubuntu\n# command: [\"echo\", \"Hello, World!\"]\n# restartPolicy: Never\njob = scheme.parse_file(\"hello-world.yaml\")\n# It is serialized to the correc type\nassert isinstance(job, batch_v1.Job)\n# You can now create the job\njob.create()\n# You can also access different registered types from the scheme\nJob = scheme.get(\"Job\")\n# Now you can parse the file using the from_file classmethod\njob = Job.from_file(\"hello-world.yaml\")\n# the above is correctly typed with mypy if you use the cloudcoil mypy extension\n# even though the class was dynamically loaded from the scheme via a string\n\n# Listing resources\n# cloudcoil provides two ways to list resources:\n\n# 1. Iterator API (Recommended)\n# This automatically handles pagination and provides a clean interface\nfor pod in core_v1.Pod.list(all_namespaces=True):\n print(pod.metadata.name)\n\n# For async code, you can use async iteration\nasync for pod in await core_v1.Pod.async_list(all_namespaces=True):\n print(pod.metadata.name)\n\n# Get total count for all items\ntotal_pods = len(pods)\nprint(f\"Total pods: {total_pods}\")\n\n\n# 2. Manual Pagination API\n# If you need more control over pagination, you can use the raw API\n# This returns a ResourceList object that contains the first page\npods = core_v1.Pod.list(namespace=\"kube-system\", limit=10)\nprint(f\"First page has {len(pods.items)} items\")\n\n# Access the items directly using the items attribute\nfor pod in pods.items:\n print(pod.metadata.name)\n\n# Check if there are more pages\nif pods.has_next_page():\n # Get the next page\n next_page = pods.get_next_page()\n # For async code\n next_page = await pods.async_get_next_page()\n\n# It's recommended to use the Iterator API as it handles pagination\n# automatically and provides a more ergonomic interface\n```\n\n### Testing Integration\n\ncloudcoil includes pytest fixtures to help you test your Kubernetes applications. Install with test dependencies:\n\n```bash\nuv add cloudcoil[test]\n```\n\nThe testing integration provides two key fixtures:\n\n- `test_cluster`: Creates and manages a k3d cluster for testing\n- `test_config`: Provides a Config instance configured for the test cluster\n\nExample usage:\n\n```python\nimport pytest\nfrom cloudcoil.kinds.core import v1 as corev1\n\n@pytest.mark.configure_test_cluster(\n cluster_name=\"my-test-cluster\",\n k3d_version=\"v5.7.5\",\n k8s_version=\"v1.31.4\",\n remove=True\n)\ndef test_my_resources(test_config):\n with test_config:\n namespace = corev1.Namespace.get(\"default\")\n assert namespace.metadata.name == \"default\"\n```\n\n#### Test Cluster Configuration\n\nThe `configure_test_cluster` mark accepts these arguments:\n\n- `cluster_name`: Name of the test cluster (default: auto-generated)\n- `k3d_version`: Version of k3d to use (default: v5.7.5)\n- `k8s_version`: Kubernetes version to use (default: v1.31.4)\n- `k8s_image`: Custom k3s image (default: rancher/k3s:{k8s_version}-k3s1)\n- `remove`: Whether to remove the cluster after tests (default: True)\n\n## 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[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```python\nfrom cloudcoil import scheme\n\n# This will be correctly typed as batch_v1.Job\njob_class = scheme.get(\"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## Documentation\n\nFor full documentation, please visit [cloudcoil.github.io/cloudcoil](https://cloudcoil.github.io/cloudcoil)\n\n## License\n\nThis project is licensed under the Apache License, Version 2.0 - see the [LICENSE](https://github.com/cloudcoil/cloudcoil/blob/main/LICENSE) file for details.\n",
"bugtrack_url": null,
"license": "Apache-2.0",
"summary": "Cloud native made easy with Python",
"version": "0.0.12",
"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": "",
"digests": {
"blake2b_256": "4a501e9388367559f3c7115907c60bd950568dd64da94ab45d07d058d3849ce2",
"md5": "8eb43c0f2ea811b4d750e689b9730886",
"sha256": "6cd9128cbd17806d5c6e78ede3b4d1936c5023bcf6a435469a99eba82c0ab505"
},
"downloads": -1,
"filename": "cloudcoil-0.0.12-py3-none-any.whl",
"has_sig": false,
"md5_digest": "8eb43c0f2ea811b4d750e689b9730886",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.10",
"size": 219609,
"upload_time": "2025-01-02T11:15:12",
"upload_time_iso_8601": "2025-01-02T11:15:12.466951Z",
"url": "https://files.pythonhosted.org/packages/4a/50/1e9388367559f3c7115907c60bd950568dd64da94ab45d07d058d3849ce2/cloudcoil-0.0.12-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "482365b62146872ce439b80bfaa88404cfbae461f68589d6b301568c97885616",
"md5": "7e5cb77ec1dac06d669f5be5970c22e4",
"sha256": "7da409ddd4ff4b478d358a2e9caa3388355e388a51589efa93e0042fa35820fb"
},
"downloads": -1,
"filename": "cloudcoil-0.0.12.tar.gz",
"has_sig": false,
"md5_digest": "7e5cb77ec1dac06d669f5be5970c22e4",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.10",
"size": 254548,
"upload_time": "2025-01-02T11:15:14",
"upload_time_iso_8601": "2025-01-02T11:15:14.336278Z",
"url": "https://files.pythonhosted.org/packages/48/23/65b62146872ce439b80bfaa88404cfbae461f68589d6b301568c97885616/cloudcoil-0.0.12.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-01-02 11:15:14",
"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
