# digital-ocean-cluster
A well tested library for managing a fleet of droplets.
[](../../actions/workflows/lint.yml)
[](../../actions/workflows/push_macos.yml)
[](../../actions/workflows/push_ubuntu.yml)
[](../../actions/workflows/push_win.yml)
# About
This library concurrent creates and runs digital ocean droplets through the doctl command line interface. This api allows massive concurrency running each action on a seperate thread.
The amount of implemented features for doctl is very few, but just enough to bring up a Droplet cloud, install dependencies, and execute commands on the cluster.
To develop software, run `. ./activate`
# Windows
This environment requires you to use `git-bash`.
# Linting
Run `./lint.sh` to find linting errors using `pylint`, `flake8` and `mypy`.
# Pre-requesits
* You will need to have an ssh key registered with digital ocean. This key must also be in your ~/.ssh folder.
* You will need to have the doctl binary installed in your path.
TODO: Make a more minimal example
# Example
```python
"""
Unit test file.
"""
import os
import subprocess
import unittest
from pathlib import Path
from digital_ocean_cluster import (
DigitalOceanCluster,
Droplet,
DropletCluster,
DropletCreationArgs,
)
# os.environ["home"] = "/home/niteris"
IS_GITHUB = os.environ.get("GITHUB_ACTIONS", False)
TAGS = ["test", "cluster"]
CLUSTER_SIZE = 4
def install(droplet: Droplet) -> None:
"""Install a package."""
# droplet.run_cmd("apt-get update")
#droplet.run_cmd("apt-get install -y vim")
droplet.copy_text_to("echo 'Install Done!'", Path("/root/test.sh"))
class DigitalOceanClusterTester(unittest.TestCase):
"""Main tester class."""
@unittest.skipIf(IS_GITHUB, "Skipping test for GitHub Actions")
def test_create_droplets(self) -> None:
"""Test command line interface (CLI)."""
# first delete the previous cluster
# create a cluster of 4 machines
# Deleting the cluster
deleted: list[Droplet] = DigitalOceanCluster.delete_cluster(TAGS)
print(f"Deleted: {[d.name for d in deleted]}")
creation_args: list[DropletCreationArgs] = [
DropletCreationArgs(name=f"test-droplet-creation-{i}", tags=TAGS, install=install)
for i in range(CLUSTER_SIZE)
]
print(f"Creating droplets: {creation_args}")
cluster: DropletCluster = DigitalOceanCluster.create_droplets(creation_args)
self.assertEqual(len(cluster.droplets), CLUSTER_SIZE)
self.assertEqual(len(cluster.failed_droplets), 0)
# now run ls on all of them
cmd = "pwd"
result: dict[Droplet, CompletedProcess] = cluster.run_cmd(cmd)
for _, cp in result.items():
self.assertIn(
"/root",
cp.stdout,
f"Error: {cp.returncode}\n\nstderr:\n{cp.stderr}\n\nstdout:\n{cp.stdout}",
)
content: str = "the quick brown fox jumps over the lazy dog"
remote_path = Path("/root/test.txt")
# now copy a file to all of them
cluster.copy_text_to(content, remote_path)
# now get the text back
results: dict[Droplet, str | Exception] = cluster.copy_text_from(remote_path)
for droplet, text in results.items():
if isinstance(text, Exception):
print(f"Error: {text}")
self.fail(f"Droplet {droplet.name} failed\nError: {text}")
else:
print(f"Text: {text}")
print("Deleting cluster")
# now delete the cluster
DigitalOceanCluster.delete_cluster(cluster)
print("Deleted cluster")
if __name__ == "__main__":
unittest.main()
```
Raw data
{
"_id": null,
"home_page": null,
"name": "digital-ocean-cluster",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.10",
"maintainer_email": null,
"keywords": "template-python-cmd",
"author": null,
"author_email": null,
"download_url": "https://files.pythonhosted.org/packages/72/07/9e68fb76c0e8a8fa9d87a67445b61f5a25cde082492071c7c10fe31d250b/digital_ocean_cluster-1.1.22.tar.gz",
"platform": null,
"description": "# digital-ocean-cluster\r\n\r\nA well tested library for managing a fleet of droplets.\r\n\r\n[](../../actions/workflows/lint.yml)\r\n\r\n[](../../actions/workflows/push_macos.yml)\r\n[](../../actions/workflows/push_ubuntu.yml)\r\n[](../../actions/workflows/push_win.yml)\r\n\r\n\r\n# About\r\n\r\nThis library concurrent creates and runs digital ocean droplets through the doctl command line interface. This api allows massive concurrency running each action on a seperate thread.\r\n\r\nThe amount of implemented features for doctl is very few, but just enough to bring up a Droplet cloud, install dependencies, and execute commands on the cluster.\r\n\r\nTo develop software, run `. ./activate`\r\n\r\n# Windows\r\n\r\nThis environment requires you to use `git-bash`.\r\n\r\n# Linting\r\n\r\nRun `./lint.sh` to find linting errors using `pylint`, `flake8` and `mypy`.\r\n\r\n# Pre-requesits\r\n\r\n * You will need to have an ssh key registered with digital ocean. This key must also be in your ~/.ssh folder.\r\n * You will need to have the doctl binary installed in your path.\r\n\r\nTODO: Make a more minimal example\r\n\r\n# Example\r\n\r\n```python\r\n\"\"\"\r\nUnit test file.\r\n\"\"\"\r\n\r\nimport os\r\nimport subprocess\r\nimport unittest\r\nfrom pathlib import Path\r\n\r\nfrom digital_ocean_cluster import (\r\n DigitalOceanCluster,\r\n Droplet,\r\n DropletCluster,\r\n DropletCreationArgs,\r\n)\r\n\r\n# os.environ[\"home\"] = \"/home/niteris\"\r\n\r\nIS_GITHUB = os.environ.get(\"GITHUB_ACTIONS\", False)\r\n\r\nTAGS = [\"test\", \"cluster\"]\r\n\r\nCLUSTER_SIZE = 4\r\n\r\n\r\ndef install(droplet: Droplet) -> None:\r\n \"\"\"Install a package.\"\"\"\r\n # droplet.run_cmd(\"apt-get update\")\r\n #droplet.run_cmd(\"apt-get install -y vim\")\r\n droplet.copy_text_to(\"echo 'Install Done!'\", Path(\"/root/test.sh\"))\r\n\r\nclass DigitalOceanClusterTester(unittest.TestCase):\r\n \"\"\"Main tester class.\"\"\"\r\n\r\n @unittest.skipIf(IS_GITHUB, \"Skipping test for GitHub Actions\")\r\n def test_create_droplets(self) -> None:\r\n \"\"\"Test command line interface (CLI).\"\"\"\r\n # first delete the previous cluster\r\n # create a cluster of 4 machines\r\n # Deleting the cluster\r\n deleted: list[Droplet] = DigitalOceanCluster.delete_cluster(TAGS)\r\n print(f\"Deleted: {[d.name for d in deleted]}\")\r\n\r\n creation_args: list[DropletCreationArgs] = [\r\n DropletCreationArgs(name=f\"test-droplet-creation-{i}\", tags=TAGS, install=install)\r\n for i in range(CLUSTER_SIZE)\r\n ]\r\n\r\n print(f\"Creating droplets: {creation_args}\")\r\n cluster: DropletCluster = DigitalOceanCluster.create_droplets(creation_args)\r\n self.assertEqual(len(cluster.droplets), CLUSTER_SIZE)\r\n self.assertEqual(len(cluster.failed_droplets), 0)\r\n\r\n # now run ls on all of them\r\n cmd = \"pwd\"\r\n result: dict[Droplet, CompletedProcess] = cluster.run_cmd(cmd)\r\n for _, cp in result.items():\r\n self.assertIn(\r\n \"/root\",\r\n cp.stdout,\r\n f\"Error: {cp.returncode}\\n\\nstderr:\\n{cp.stderr}\\n\\nstdout:\\n{cp.stdout}\",\r\n )\r\n\r\n content: str = \"the quick brown fox jumps over the lazy dog\"\r\n remote_path = Path(\"/root/test.txt\")\r\n\r\n # now copy a file to all of them\r\n cluster.copy_text_to(content, remote_path)\r\n\r\n # now get the text back\r\n results: dict[Droplet, str | Exception] = cluster.copy_text_from(remote_path)\r\n for droplet, text in results.items():\r\n if isinstance(text, Exception):\r\n print(f\"Error: {text}\")\r\n self.fail(f\"Droplet {droplet.name} failed\\nError: {text}\")\r\n else:\r\n print(f\"Text: {text}\")\r\n\r\n print(\"Deleting cluster\")\r\n # now delete the cluster\r\n DigitalOceanCluster.delete_cluster(cluster)\r\n print(\"Deleted cluster\")\r\n\r\n\r\nif __name__ == \"__main__\":\r\n unittest.main()\r\n\r\n```\r\n",
"bugtrack_url": null,
"license": "BSD 3-Clause License",
"summary": "digital ocean cluster management through droplets",
"version": "1.1.22",
"project_urls": null,
"split_keywords": [
"template-python-cmd"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "72079e68fb76c0e8a8fa9d87a67445b61f5a25cde082492071c7c10fe31d250b",
"md5": "01ea26255f22070f5cfd8543f01dafa1",
"sha256": "877611de8314419a93ba731050e4d5f0be2268743c7be05b5a8ca95a8bd3f050"
},
"downloads": -1,
"filename": "digital_ocean_cluster-1.1.22.tar.gz",
"has_sig": false,
"md5_digest": "01ea26255f22070f5cfd8543f01dafa1",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.10",
"size": 22362,
"upload_time": "2025-02-10T23:07:32",
"upload_time_iso_8601": "2025-02-10T23:07:32.541737Z",
"url": "https://files.pythonhosted.org/packages/72/07/9e68fb76c0e8a8fa9d87a67445b61f5a25cde082492071c7c10fe31d250b/digital_ocean_cluster-1.1.22.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-02-10 23:07:32",
"github": false,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"lcname": "digital-ocean-cluster"
}
JSON