diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
index 4cec833..fc434ac 100644
--- a/.pre-commit-config.yaml
+++ b/.pre-commit-config.yaml
@@ -9,7 +9,7 @@ repos:
hooks:
- id: sync-with-uv
- repo: https://github.com/charliermarsh/ruff-pre-commit
- rev: v0.15.17
+ rev: v0.15.20
hooks:
- id: ruff-check
args: [--fix, --exit-non-zero-on-fix]
diff --git a/CHANGELOG.md b/CHANGELOG.md
index e5a0336..935907e 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -7,6 +7,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
## [Unreleased]
+## [0.55.0] - 2026-07-01
+
+### Added
+
+- `tilebox-workflows`: Added workflow management client methods for creating, listing, finding, updating, and deleting
+ workflows.
+- `tilebox-workflows`: Added workflow release deployment client methods for unpublishing, deploying, and undeploying
+ releases.
+- `tilebox-workflows`: Added `ClusterClient.update()` and optional cluster slugs for `ClusterClient.create()`
+- `tilebox-workflows`: Cluster slug filters for `JobClient.query()`.
+
## [0.54.0] - 2026-06-17
### Added
@@ -383,7 +394,8 @@ the first client that does not cache data (since it's already on the local file
- Released under the [MIT](https://opensource.org/license/mit) license.
- Released packages: `tilebox-datasets`, `tilebox-workflows`, `tilebox-storage`, `tilebox-grpc`
-[Unreleased]: https://github.com/tilebox/tilebox-python/compare/v0.54.0...HEAD
+[Unreleased]: https://github.com/tilebox/tilebox-python/compare/v0.55.0...HEAD
+[0.55.0]: https://github.com/tilebox/tilebox-python/compare/v0.54.0...v0.55.0
[0.54.0]: https://github.com/tilebox/tilebox-python/compare/v0.53.0...v0.54.0
[0.53.0]: https://github.com/tilebox/tilebox-python/compare/v0.52.0...v0.53.0
[0.52.0]: https://github.com/tilebox/tilebox-python/compare/v0.51.0...v0.52.0
diff --git a/tilebox-datasets/tilebox/datasets/protobuf_conversion/field_types.py b/tilebox-datasets/tilebox/datasets/protobuf_conversion/field_types.py
index 6954bd9..22d92fd 100644
--- a/tilebox-datasets/tilebox/datasets/protobuf_conversion/field_types.py
+++ b/tilebox-datasets/tilebox/datasets/protobuf_conversion/field_types.py
@@ -39,8 +39,8 @@
npdtypes.BoolDType: False,
npdtypes.StrDType: "",
npdtypes.ObjectDType: None,
- npdtypes.DateTime64DType: np.datetime64("NaT"),
- npdtypes.TimeDelta64DType: np.timedelta64("NaT"),
+ npdtypes.DateTime64DType: np.datetime64("NaT", "ns"),
+ npdtypes.TimeDelta64DType: np.timedelta64("NaT", "ns"),
}
diff --git a/tilebox-storage/tests/test_providers.py b/tilebox-storage/tests/test_providers.py
index 8de40f4..cb1b245 100644
--- a/tilebox-storage/tests/test_providers.py
+++ b/tilebox-storage/tests/test_providers.py
@@ -1,10 +1,8 @@
import re
-from typing import cast
import pytest
import responses
from niquests import AsyncSession
-from niquests.cookies import RequestsCookieJar
from tilebox.storage.providers import _asf_login
@@ -18,12 +16,11 @@ async def test_asf_login() -> None:
responses.add(responses.GET, ASF_LOGIN_URL, headers={"Set-Cookie": "logged_in=yes"})
client = await _asf_login(("username", "password"))
- cookies = cast(RequestsCookieJar, client.cookies)
assert isinstance(client, AsyncSession)
assert "asf_search" in str(client.headers["Client-Id"])
assert client.auth == ("username", "password")
- assert cookies["logged_in"] == "yes"
+ assert client.cookies["logged_in"] == "yes"
await client.close()
diff --git a/tilebox-workflows/tests/clusters/test_client.py b/tilebox-workflows/tests/clusters/test_client.py
index 2a2f8b1..10c2be9 100644
--- a/tilebox-workflows/tests/clusters/test_client.py
+++ b/tilebox-workflows/tests/clusters/test_client.py
@@ -19,6 +19,7 @@
GetClusterRequest,
ListClustersRequest,
ListClustersResponse,
+ UpdateClusterRequest,
)
from tilebox.workflows.workflows.v1.workflows_pb2_grpc import WorkflowsServiceStub
@@ -30,7 +31,19 @@ def __init__(self) -> None:
self.clusters: dict[str, ClusterMessage] = {}
def CreateCluster(self, req: CreateClusterRequest) -> ClusterMessage: # noqa: N802
- cluster = ClusterMessage(slug=str(uuid4()), display_name=req.name)
+ cluster = ClusterMessage(slug=req.slug or str(uuid4()), display_name=req.name, description=req.description)
+ self.clusters[cluster.slug] = cluster
+ return cluster
+
+ def UpdateCluster(self, req: UpdateClusterRequest) -> ClusterMessage: # noqa: N802
+ if req.cluster_slug not in self.clusters:
+ raise NotFoundError(f"Cluster {req.cluster_slug} not found")
+
+ cluster = ClusterMessage.FromString(self.clusters[req.cluster_slug].SerializeToString())
+ if req.HasField("name"):
+ cluster.display_name = req.name
+ if req.HasField("description"):
+ cluster.description = req.description
self.clusters[cluster.slug] = cluster
return cluster
@@ -50,6 +63,18 @@ def ListClusters(self, req: ListClustersRequest) -> ListClustersResponse: # noq
return ListClustersResponse(clusters=list(self.clusters.values()))
+def test_create_cluster_can_use_custom_slug() -> None:
+ service = ClusterService(MagicMock())
+ service.service = MockClusterService()
+ cluster_client = ClusterClient(service)
+
+ cluster = cluster_client.create("Test Cluster", description="Test description", slug="test-cluster")
+
+ assert cluster.slug == "test-cluster"
+ assert cluster.display_name == "Test Cluster"
+ assert cluster.description == "Test description"
+
+
class ClusterCRUDOperations(RuleBasedStateMachine):
"""
A state machine that tests the CRUD operations of the Clusters client.
@@ -75,7 +100,17 @@ def __init__(self) -> None:
@rule(target=inserted_clusters, cluster=clusters())
def create_cluster(self, cluster: Cluster) -> Cluster:
self.count_clusters += 1
- return self.cluster_client.create(cluster.display_name)
+ created = self.cluster_client.create(cluster.display_name, cluster.description)
+ assert created.description == cluster.description
+ return created
+
+ @rule(target=inserted_clusters, cluster=consumes(inserted_clusters), updated=clusters())
+ def update_cluster(self, cluster: Cluster, updated: Cluster) -> Cluster:
+ got = self.cluster_client.update(cluster.slug, name=updated.display_name, description=updated.description)
+ assert got.slug == cluster.slug
+ assert got.display_name == updated.display_name
+ assert got.description == (updated.description if updated.description is not None else cluster.description)
+ return got
@rule(cluster=inserted_clusters)
def get_cluster(self, cluster: Cluster) -> None:
diff --git a/tilebox-workflows/tests/jobs/test_client.py b/tilebox-workflows/tests/jobs/test_client.py
index 639a47e..0a55c22 100644
--- a/tilebox-workflows/tests/jobs/test_client.py
+++ b/tilebox-workflows/tests/jobs/test_client.py
@@ -2,6 +2,7 @@
from unittest.mock import MagicMock
from uuid import UUID, uuid4
+import pytest
from hypothesis.stateful import Bundle, RuleBasedStateMachine, consumes, rule
from tests.tasks_data import jobs
@@ -141,6 +142,7 @@ class MockJobService(JobServiceStub):
def __init__(self) -> None:
self.jobs: dict[UUID, JobMessage] = {}
+ self.query_requests: list[QueryJobsRequest] = []
def SubmitJob(self, req: SubmitJobRequest) -> JobMessage: # noqa: N802
job_id = uuid4()
@@ -193,10 +195,41 @@ def VisualizeJob(self, req: VisualizeJobRequest) -> Diagram: # noqa: N802
return Diagram(svg=b"")
def QueryJobs(self, req: QueryJobsRequest) -> QueryJobsResponse: # noqa: N802
- _ = req
+ self.query_requests.append(req)
return QueryJobsResponse(jobs=list(self.jobs.values()))
+def test_query_filters_by_clusters() -> None:
+ service = JobService(MagicMock())
+ mock_service = MockJobService()
+ service.service = mock_service
+ job_client = JobClient(service, MagicMock(), NoopWorkflowTracer())
+
+ job_client.query((uuid4(), uuid4()), clusters=["cluster-a", "cluster-b"])
+
+ assert list(mock_service.query_requests[-1].filters.cluster_slugs) == ["cluster-a", "cluster-b"]
+
+
+def test_query_empty_cluster_list_applies_no_cluster_filter() -> None:
+ service = JobService(MagicMock())
+ mock_service = MockJobService()
+ service.service = mock_service
+ job_client = JobClient(service, MagicMock(), NoopWorkflowTracer())
+
+ job_client.query((uuid4(), uuid4()), clusters=[])
+
+ assert list(mock_service.query_requests[-1].filters.cluster_slugs) == []
+
+
+def test_query_rejects_empty_cluster_slug() -> None:
+ service = JobService(MagicMock())
+ service.service = MockJobService()
+ job_client = JobClient(service, MagicMock(), NoopWorkflowTracer())
+
+ with pytest.raises(ValueError, match="explicit cluster slugs"):
+ job_client.query((uuid4(), uuid4()), clusters="")
+
+
class JobOperations(RuleBasedStateMachine):
"""
A state machine that tests the various job operations of the Jobs client.
diff --git a/tilebox-workflows/tests/tasks_data.py b/tilebox-workflows/tests/tasks_data.py
index c08ffef..0e2c936 100644
--- a/tilebox-workflows/tests/tasks_data.py
+++ b/tilebox-workflows/tests/tasks_data.py
@@ -28,16 +28,19 @@
from tilebox.datasets.query.id_interval import IDInterval
from tilebox.datasets.query.time_interval import TimeInterval
from tilebox.workflows.data import (
+ Artifact,
AutomationPrototype,
Cluster,
ComputedTask,
CronTrigger,
ExecutionStats,
+ FilesystemNode,
Idling,
Job,
JobState,
ProgressIndicator,
QueryFilters,
+ ReleaseContent,
SingleTaskSubmission,
StorageEventTrigger,
StorageLocation,
@@ -48,6 +51,8 @@
TaskState,
TaskSubmissionGroup,
TaskSubmissions,
+ Workflow,
+ WorkflowRelease,
)
@@ -63,7 +68,65 @@ def clusters(draw: DrawFn) -> Cluster:
slug = draw(alphanumerical_text(min_size=4, max_size=20))
display_name = draw(alphanumerical_text())
deletable = draw(booleans())
- return Cluster(slug, display_name, deletable)
+ description = draw(alphanumerical_text() | none())
+ return Cluster(slug, display_name, deletable, description)
+
+
+@composite
+def artifacts(draw: DrawFn) -> Artifact:
+ """A hypothesis strategy for generating random workflow release artifacts"""
+ artifact_id = draw(uuids(version=4))
+ digest = draw(text(alphabet="abcdef0123456789", min_size=64, max_size=64))
+ return Artifact(artifact_id, digest)
+
+
+@composite
+def release_filesystem_nodes(draw: DrawFn) -> FilesystemNode:
+ """A hypothesis strategy for generating random workflow release filesystem nodes"""
+ path = draw(alphanumerical_text())
+ directory = draw(booleans())
+ children = []
+ if directory:
+ children = [
+ FilesystemNode(draw(alphanumerical_text())) for _ in range(draw(integers(min_value=0, max_value=3)))
+ ]
+ return FilesystemNode(path, directory, children)
+
+
+@composite
+def release_contents(draw: DrawFn) -> ReleaseContent:
+ """A hypothesis strategy for generating random workflow release contents"""
+ fingerprint = draw(text(alphabet="abcdef0123456789", min_size=64, max_size=64))
+ return ReleaseContent(
+ fingerprint=fingerprint,
+ tasks=draw(lists(task_identifiers(), min_size=1, max_size=3)),
+ files=draw(lists(release_filesystem_nodes(), min_size=0, max_size=3)),
+ runner_object_path=draw(alphanumerical_text()),
+ command_override=draw(lists(alphanumerical_text(), min_size=0, max_size=3)),
+ )
+
+
+@composite
+def workflow_releases(draw: DrawFn) -> WorkflowRelease:
+ """A hypothesis strategy for generating random workflow releases"""
+ return WorkflowRelease(
+ id=draw(uuids(version=4)),
+ artifact=draw(artifacts() | none()),
+ content=draw(release_contents() | none()),
+ created_at=draw(datetimes(timezones=just(timezone.utc)) | none()),
+ clusters=draw(lists(clusters(), min_size=0, max_size=3)),
+ )
+
+
+@composite
+def workflows(draw: DrawFn) -> Workflow:
+ """A hypothesis strategy for generating random workflows"""
+ return Workflow(
+ slug=draw(alphanumerical_text(min_size=4, max_size=20)),
+ name=draw(alphanumerical_text()),
+ description=draw(alphanumerical_text() | none()) or "",
+ releases=draw(lists(workflow_releases(), min_size=0, max_size=3)),
+ )
@composite
@@ -354,4 +417,6 @@ def query_filters(draw: DrawFn) -> QueryFilters:
if task_states is not None:
task_states = list(set(task_states)) # de-duplicate
- return QueryFilters(time_interval, id_interval, automation_ids, job_states, name, task_states)
+ cluster_slugs = draw(lists(alphanumerical_text(min_size=4, max_size=20), min_size=0, max_size=3, unique=True))
+
+ return QueryFilters(time_interval, id_interval, automation_ids, job_states, name, task_states, cluster_slugs)
diff --git a/tilebox-workflows/tests/test_data.py b/tilebox-workflows/tests/test_data.py
index a7fdd1a..a6c5c76 100644
--- a/tilebox-workflows/tests/test_data.py
+++ b/tilebox-workflows/tests/test_data.py
@@ -1,6 +1,7 @@
from hypothesis import given
from tests.tasks_data import (
+ artifacts,
automations,
clusters,
computed_tasks,
@@ -9,6 +10,8 @@
jobs,
progress_indicators,
query_filters,
+ release_contents,
+ release_filesystem_nodes,
single_task_submissions,
storage_locations,
task_identifiers,
@@ -16,16 +19,21 @@
task_submission_groups,
task_submissions,
tasks,
+ workflow_releases,
+ workflows,
)
from tilebox.workflows.data import (
+ Artifact,
AutomationPrototype,
Cluster,
ComputedTask,
ExecutionStats,
+ FilesystemNode,
Idling,
Job,
ProgressIndicator,
QueryFilters,
+ ReleaseContent,
SingleTaskSubmission,
StorageLocation,
Task,
@@ -33,6 +41,8 @@
TaskLease,
TaskSubmissionGroup,
TaskSubmissions,
+ Workflow,
+ WorkflowRelease,
)
@@ -77,6 +87,31 @@ def test_clusters_to_message_and_back(cluster: Cluster) -> None:
assert Cluster.from_message(cluster.to_message()) == cluster
+@given(artifacts())
+def test_artifacts_to_message_and_back(artifact: Artifact) -> None:
+ assert Artifact.from_message(artifact.to_message()) == artifact
+
+
+@given(release_filesystem_nodes())
+def test_release_filesystem_nodes_to_message_and_back(node: FilesystemNode) -> None:
+ assert FilesystemNode.from_message(node.to_message()) == node
+
+
+@given(release_contents())
+def test_release_contents_to_message_and_back(content: ReleaseContent) -> None:
+ assert ReleaseContent.from_message(content.to_message()) == content
+
+
+@given(workflow_releases())
+def test_workflow_releases_to_message_and_back(release: WorkflowRelease) -> None:
+ assert WorkflowRelease.from_message(release.to_message()) == release
+
+
+@given(workflows())
+def test_workflows_to_message_and_back(workflow: Workflow) -> None:
+ assert Workflow.from_message(workflow.to_message()) == workflow
+
+
@given(task_submissions())
def test_task_submissions_to_message_and_back(sub_task: TaskSubmissions) -> None:
assert TaskSubmissions.from_message(sub_task.to_message()) == sub_task
diff --git a/tilebox-workflows/tests/workflows/test_client.py b/tilebox-workflows/tests/workflows/test_client.py
new file mode 100644
index 0000000..1d020ff
--- /dev/null
+++ b/tilebox-workflows/tests/workflows/test_client.py
@@ -0,0 +1,257 @@
+from datetime import datetime, timezone
+from unittest.mock import MagicMock
+from uuid import UUID, uuid4
+
+from _tilebox.grpc.error import NotFoundError
+from tilebox.datasets.query.time_interval import datetime_to_timestamp
+from tilebox.datasets.uuid import uuid_message_to_uuid, uuid_to_uuid_message
+from tilebox.workflows.data import (
+ Artifact,
+ Cluster,
+ FilesystemNode,
+ ReleaseContent,
+ TaskIdentifier,
+ Workflow,
+ WorkflowRelease,
+)
+from tilebox.workflows.workflows.client import WorkflowClient
+from tilebox.workflows.workflows.service import WorkflowService
+from tilebox.workflows.workflows.v1.workflows_pb2 import (
+ Cluster as ClusterMessage,
+)
+from tilebox.workflows.workflows.v1.workflows_pb2 import (
+ CreateWorkflowRequest,
+ DeleteWorkflowRequest,
+ DeployWorkflowReleaseRequest,
+ DeployWorkflowReleaseResponse,
+ GetWorkflowRequest,
+ ListWorkflowsRequest,
+ ListWorkflowsResponse,
+ UndeployWorkflowReleaseRequest,
+ UndeployWorkflowReleaseResponse,
+ UnpublishWorkflowReleaseRequest,
+ UpdateWorkflowRequest,
+)
+from tilebox.workflows.workflows.v1.workflows_pb2 import (
+ Workflow as WorkflowMessage,
+)
+from tilebox.workflows.workflows.v1.workflows_pb2 import (
+ WorkflowRelease as WorkflowReleaseMessage,
+)
+from tilebox.workflows.workflows.v1.workflows_pb2_grpc import WorkflowsServiceStub
+
+
+class MockWorkflowService(WorkflowsServiceStub):
+ """A mock implementation of the gRPC workflow service, that stores workflows in memory as a dict."""
+
+ def __init__(self) -> None:
+ self.workflows: dict[str, WorkflowMessage] = {}
+ self.unpublished_releases: list[tuple[str, UUID]] = []
+ self.deploy_requests: list[DeployWorkflowReleaseRequest] = []
+ self.undeploy_requests: list[UndeployWorkflowReleaseRequest] = []
+ self.update_requests: list[UpdateWorkflowRequest] = []
+
+ def CreateWorkflow(self, req: CreateWorkflowRequest) -> WorkflowMessage: # noqa: N802
+ slug = req.name.lower().replace(" ", "-")
+ workflow = WorkflowMessage(slug=slug, name=req.name, description=req.description)
+ self.workflows[workflow.slug] = workflow
+ return workflow
+
+ def ListWorkflows(self, req: ListWorkflowsRequest) -> ListWorkflowsResponse: # noqa: N802
+ _ = req
+ return ListWorkflowsResponse(workflows=list(self.workflows.values()))
+
+ def GetWorkflow(self, req: GetWorkflowRequest) -> WorkflowMessage: # noqa: N802
+ if req.workflow_slug in self.workflows:
+ return self.workflows[req.workflow_slug]
+ raise NotFoundError(f"Workflow {req.workflow_slug} not found")
+
+ def UpdateWorkflow(self, req: UpdateWorkflowRequest) -> WorkflowMessage: # noqa: N802
+ self.update_requests.append(req)
+ if req.workflow_slug not in self.workflows:
+ raise NotFoundError(f"Workflow {req.workflow_slug} not found")
+
+ workflow = WorkflowMessage.FromString(self.workflows[req.workflow_slug].SerializeToString())
+ if req.HasField("name"):
+ workflow.name = req.name
+ if req.HasField("description"):
+ workflow.description = req.description
+ self.workflows[workflow.slug] = workflow
+ return workflow
+
+ def DeleteWorkflow(self, req: DeleteWorkflowRequest) -> None: # noqa: N802
+ if req.workflow_slug in self.workflows:
+ del self.workflows[req.workflow_slug]
+ else:
+ raise NotFoundError(f"Workflow {req.workflow_slug} not found")
+
+ def UnpublishWorkflowRelease(self, req: UnpublishWorkflowReleaseRequest) -> None: # noqa: N802
+ self.unpublished_releases.append((req.workflow_slug, uuid_message_to_uuid(req.release_id)))
+
+ def DeployWorkflowRelease(self, req: DeployWorkflowReleaseRequest) -> DeployWorkflowReleaseResponse: # noqa: N802
+ self.deploy_requests.append(req)
+ return DeployWorkflowReleaseResponse(
+ release=WorkflowReleaseMessage(id=req.release_id),
+ clusters=[ClusterMessage(slug=slug, display_name=slug) for slug in req.cluster_slugs],
+ )
+
+ def UndeployWorkflowRelease(self, req: UndeployWorkflowReleaseRequest) -> UndeployWorkflowReleaseResponse: # noqa: N802
+ self.undeploy_requests.append(req)
+ return UndeployWorkflowReleaseResponse(
+ release=WorkflowReleaseMessage(id=req.release_id),
+ clusters=[ClusterMessage(slug=slug, display_name=slug) for slug in req.cluster_slugs],
+ )
+
+
+def test_create_workflow() -> None:
+ service = WorkflowService(MagicMock())
+ service.service = MockWorkflowService()
+ workflow_client = WorkflowClient(service)
+
+ workflow = workflow_client.create("Agentic Workflow", description="Description")
+
+ assert workflow.slug == "agentic-workflow"
+ assert workflow.name == "Agentic Workflow"
+ assert workflow.description == "Description"
+
+
+def test_list_find_and_delete_workflows() -> None:
+ service = WorkflowService(MagicMock())
+ mock_service = MockWorkflowService()
+ service.service = mock_service
+ workflow_client = WorkflowClient(service)
+ workflow = workflow_client.create("Agentic Workflow")
+
+ assert workflow_client.all() == [workflow]
+ assert workflow_client.find(workflow.slug) == workflow
+
+ workflow_client.delete(workflow)
+
+ assert workflow_client.all() == []
+
+
+def test_update_workflow() -> None:
+ service = WorkflowService(MagicMock())
+ mock_service = MockWorkflowService()
+ service.service = mock_service
+ workflow_client = WorkflowClient(service)
+ workflow = workflow_client.create("Agentic Workflow", description="Description")
+
+ updated_workflow = workflow_client.update(workflow, name="Updated Workflow", description="Updated description")
+
+ request = mock_service.update_requests[-1]
+ assert request.workflow_slug == workflow.slug
+ assert request.HasField("name")
+ assert request.HasField("description")
+ assert request.name == "Updated Workflow"
+ assert request.description == "Updated description"
+ assert updated_workflow.name == "Updated Workflow"
+ assert updated_workflow.description == "Updated description"
+
+
+def test_update_workflow_preserves_optional_presence() -> None:
+ service = WorkflowService(MagicMock())
+ mock_service = MockWorkflowService()
+ service.service = mock_service
+ workflow_client = WorkflowClient(service)
+ workflow = workflow_client.create("Agentic Workflow", description="Description")
+
+ workflow_client.update(workflow.slug)
+ request = mock_service.update_requests[-1]
+ assert not request.HasField("name")
+ assert not request.HasField("description")
+
+ updated_workflow = workflow_client.update(workflow.slug, description="")
+ request = mock_service.update_requests[-1]
+ assert not request.HasField("name")
+ assert request.HasField("description")
+ assert request.description == ""
+ assert updated_workflow.description == ""
+
+
+def test_unpublish_release() -> None:
+ service = WorkflowService(MagicMock())
+ mock_service = MockWorkflowService()
+ service.service = mock_service
+ workflow_client = WorkflowClient(service)
+ release_id = uuid4()
+
+ workflow_client.unpublish_release("agentic-workflow", str(release_id))
+
+ assert mock_service.unpublished_releases == [("agentic-workflow", release_id)]
+
+
+def test_deploy_release() -> None:
+ service = WorkflowService(MagicMock())
+ mock_service = MockWorkflowService()
+ service.service = mock_service
+ workflow_client = WorkflowClient(service)
+ release_id = uuid4()
+
+ deployment = workflow_client.deploy_release("agentic-workflow", release_id, clusters=["dev", "prod"])
+
+ request = mock_service.deploy_requests[-1]
+ assert request.workflow_slug == "agentic-workflow"
+ assert uuid_message_to_uuid(request.release_id) == release_id
+ assert list(request.cluster_slugs) == ["dev", "prod"]
+ assert deployment.release.id == release_id
+ assert [cluster.slug for cluster in deployment.clusters] == ["dev", "prod"]
+
+
+def test_undeploy_release() -> None:
+ service = WorkflowService(MagicMock())
+ mock_service = MockWorkflowService()
+ service.service = mock_service
+ workflow_client = WorkflowClient(service)
+ release_id = uuid4()
+
+ deployment = workflow_client.undeploy_release("agentic-workflow", release_id, clusters="dev")
+
+ request = mock_service.undeploy_requests[-1]
+ assert request.workflow_slug == "agentic-workflow"
+ assert uuid_message_to_uuid(request.release_id) == release_id
+ assert list(request.cluster_slugs) == ["dev"]
+ assert deployment.release.id == release_id
+ assert [cluster.slug for cluster in deployment.clusters] == ["dev"]
+
+
+def test_workflow_release_data_includes_release_content_and_clusters() -> None:
+ release_id = uuid4()
+ artifact_id = uuid4()
+ created_at = datetime.now(tz=timezone.utc).replace(microsecond=0)
+ release = WorkflowRelease(
+ id=release_id,
+ artifact=Artifact(artifact_id, "a" * 64),
+ content=ReleaseContent(
+ fingerprint="b" * 64,
+ tasks=[TaskIdentifier("tilebox.com/task/Review", "v1.0")],
+ files=[FilesystemNode(".", directory=True, children=[FilesystemNode("main.py")])],
+ runner_object_path="my_module.my_runner:runner",
+ command_override=["python", "main.py"],
+ ),
+ created_at=created_at,
+ clusters=[],
+ )
+ workflow = Workflow("agentic-workflow", "Agentic Workflow", "Description", releases=[release])
+ cluster = ClusterMessage(
+ slug="dev",
+ display_name="Dev",
+ deployed_releases=[workflow.to_message()],
+ )
+
+ restored_release = WorkflowRelease.from_message(
+ WorkflowReleaseMessage(
+ id=uuid_to_uuid_message(release_id),
+ artifact=release.artifact.to_message() if release.artifact else None,
+ content=release.content.to_message() if release.content else None,
+ created_at=datetime_to_timestamp(created_at),
+ clusters=[cluster],
+ )
+ )
+
+ assert restored_release.id == release_id
+ assert restored_release.artifact == release.artifact
+ assert restored_release.content == release.content
+ assert restored_release.created_at == created_at
+ assert restored_release.clusters[0].deployed_workflows == []
+ assert Cluster.from_message(cluster).deployed_workflows == [workflow]
diff --git a/tilebox-workflows/tilebox/workflows/client.py b/tilebox-workflows/tilebox/workflows/client.py
index 16db0b1..61275f5 100644
--- a/tilebox-workflows/tilebox/workflows/client.py
+++ b/tilebox-workflows/tilebox/workflows/client.py
@@ -25,6 +25,8 @@
from tilebox.workflows.runner.task_runner import TaskRunner, _LeaseRenewer
from tilebox.workflows.runner.task_service import TaskService
from tilebox.workflows.task import Task
+from tilebox.workflows.workflows.client import WorkflowClient
+from tilebox.workflows.workflows.service import WorkflowService
class Client:
@@ -57,6 +59,7 @@ def __init__(
self._job_service = open_channel(url, token)
self._telemetry_service = self._job_service
self._cluster_service = self._job_service
+ self._workflow_service = self._job_service
self._automation_service = self._job_service
self._task_service = self._job_service
case "http1":
@@ -82,6 +85,7 @@ def __init__(
self._cluster_service = ConnectStubAdapter(
WorkflowsServiceClientSync(address, http_client=http_client), headers
)
+ self._workflow_service = self._cluster_service
self._automation_service = ConnectStubAdapter(
AutomationServiceClientSync(address, http_client=http_client), headers
)
@@ -203,6 +207,15 @@ def clusters(self) -> ClusterClient:
"""
return ClusterClient(ClusterService(self._cluster_service))
+ def workflows(self) -> WorkflowClient:
+ """
+ Get a client for the workflows service.
+
+ Returns:
+ A client for the workflows service.
+ """
+ return WorkflowClient(WorkflowService(self._workflow_service))
+
def automations(self) -> AutomationClient:
"""
Get a client for the automations service.
diff --git a/tilebox-workflows/tilebox/workflows/clusters/client.py b/tilebox-workflows/tilebox/workflows/clusters/client.py
index 908913c..88c9ffb 100644
--- a/tilebox-workflows/tilebox/workflows/clusters/client.py
+++ b/tilebox-workflows/tilebox/workflows/clusters/client.py
@@ -15,18 +15,38 @@ def __init__(self, service: ClusterService) -> None:
"""
self._service = service
- def create(self, name: str) -> Cluster:
+ def create(self, name: str, description: str | None = None, slug: str | None = None) -> Cluster:
"""Create a new cluster with the given name.
- A unique cluster slug will be generated for the cluster.
+ If no slug is provided, a unique cluster slug will be generated for the cluster.
Args:
name: The name of the cluster to create.
+ description: The description of the cluster to create.
+ slug: The slug to use for the cluster.
Returns:
Cluster: The created cluster.
"""
- return self._service.create(name)
+ return self._service.create(name, description, slug)
+
+ def update(
+ self,
+ cluster_or_slug: ClusterSlugLike,
+ name: str | None = None,
+ description: str | None = None,
+ ) -> Cluster:
+ """Update a cluster by slug.
+
+ Args:
+ cluster_or_slug: The cluster or slug of the cluster to update.
+ name: The new display name for the cluster. If not provided, the name is left unchanged.
+ description: The new description for the cluster. If not provided, the description is left unchanged.
+
+ Returns:
+ The updated cluster.
+ """
+ return self._service.update(to_cluster_slug(cluster_or_slug), name, description)
def all(self) -> list[Cluster]:
"""List all available clusters.
diff --git a/tilebox-workflows/tilebox/workflows/clusters/service.py b/tilebox-workflows/tilebox/workflows/clusters/service.py
index e094089..e9851cf 100644
--- a/tilebox-workflows/tilebox/workflows/clusters/service.py
+++ b/tilebox-workflows/tilebox/workflows/clusters/service.py
@@ -12,6 +12,7 @@
GetClusterRequest,
ListClustersRequest,
ListClustersResponse,
+ UpdateClusterRequest,
)
from tilebox.workflows.workflows.v1.workflows_pb2_grpc import WorkflowsServiceStub
@@ -29,10 +30,14 @@ def __init__(self, channel: Channel | Any) -> None:
with_pythonic_errors(WorkflowsServiceStub(channel)) if hasattr(channel, "unary_unary") else channel
)
- def create(self, cluster_name: str) -> Cluster:
- request = CreateClusterRequest(name=cluster_name)
+ def create(self, cluster_name: str, description: str | None = None, slug: str | None = None) -> Cluster:
+ request = CreateClusterRequest(name=cluster_name, description=description, slug=slug or "")
return Cluster.from_message(self.service.CreateCluster(request))
+ def update(self, cluster_slug: str, name: str | None = None, description: str | None = None) -> Cluster:
+ request = UpdateClusterRequest(cluster_slug=cluster_slug, name=name, description=description)
+ return Cluster.from_message(self.service.UpdateCluster(request))
+
def get_by_slug(self, cluster_slug: str) -> Cluster:
request = GetClusterRequest(cluster_slug=cluster_slug)
return Cluster.from_message(self.service.GetCluster(request))
diff --git a/tilebox-workflows/tilebox/workflows/data.py b/tilebox-workflows/tilebox/workflows/data.py
index 482ba8a..0f89eda 100644
--- a/tilebox-workflows/tilebox/workflows/data.py
+++ b/tilebox-workflows/tilebox/workflows/data.py
@@ -320,15 +320,100 @@ class Cluster:
slug: str
display_name: str
deletable: bool
+ description: str | None = None
+ deployed_workflows: list["Workflow"] = field(default_factory=list)
@classmethod # lets use typing.Self once we require python >= 3.11
- def from_message(cls, cluster: workflows_pb2.Cluster) -> "Cluster":
+ def from_message(cls, cluster: workflows_pb2.Cluster, *, include_deployed_workflows: bool = True) -> "Cluster":
"""Convert a Cluster protobuf message to a Cluster object."""
- return cls(slug=cluster.slug, display_name=cluster.display_name, deletable=cluster.deletable)
+ return cls(
+ slug=cluster.slug,
+ display_name=cluster.display_name,
+ deletable=cluster.deletable,
+ description=cluster.description or None,
+ deployed_workflows=[Workflow.from_message(workflow) for workflow in cluster.deployed_releases]
+ if include_deployed_workflows
+ else [],
+ )
def to_message(self) -> workflows_pb2.Cluster:
"""Convert a Cluster object to a Cluster protobuf message."""
- return workflows_pb2.Cluster(slug=self.slug, display_name=self.display_name, deletable=self.deletable)
+ return workflows_pb2.Cluster(
+ slug=self.slug,
+ display_name=self.display_name,
+ deletable=self.deletable,
+ description=self.description,
+ deployed_releases=[workflow.to_message() for workflow in self.deployed_workflows],
+ )
+
+
+@dataclass(order=True, frozen=True)
+class Artifact:
+ id: UUID
+ digest: str
+
+ @classmethod
+ def from_message(cls, artifact: workflows_pb2.Artifact) -> "Artifact":
+ """Convert an Artifact protobuf message to an Artifact object."""
+ return cls(id=uuid_message_to_uuid(artifact.id), digest=artifact.digest)
+
+ def to_message(self) -> workflows_pb2.Artifact:
+ """Convert an Artifact object to an Artifact protobuf message."""
+ return workflows_pb2.Artifact(id=uuid_to_uuid_message(self.id), digest=self.digest)
+
+
+@dataclass(order=True, frozen=True)
+class FilesystemNode:
+ path: str
+ directory: bool = False
+ children: list["FilesystemNode"] = field(default_factory=list)
+
+ @classmethod
+ def from_message(cls, path: workflows_pb2.Path) -> "FilesystemNode":
+ """Convert a Path protobuf message to a FilesystemNode object."""
+ return cls(
+ path=path.path,
+ directory=path.directory,
+ children=[cls.from_message(child) for child in path.children],
+ )
+
+ def to_message(self) -> workflows_pb2.Path:
+ """Convert a FilesystemNode object to a Path protobuf message."""
+ return workflows_pb2.Path(
+ path=self.path,
+ directory=self.directory,
+ children=[child.to_message() for child in self.children],
+ )
+
+
+@dataclass(order=True, frozen=True)
+class ReleaseContent:
+ fingerprint: str
+ tasks: list[TaskIdentifier]
+ files: list[FilesystemNode] = field(default_factory=list)
+ runner_object_path: str = ""
+ command_override: list[str] = field(default_factory=list)
+
+ @classmethod
+ def from_message(cls, content: workflows_pb2.ReleaseContent) -> "ReleaseContent":
+ """Convert a ReleaseContent protobuf message to a ReleaseContent object."""
+ return cls(
+ fingerprint=content.fingerprint,
+ tasks=[TaskIdentifier.from_message(task) for task in content.tasks],
+ files=[FilesystemNode.from_message(path) for path in content.files],
+ runner_object_path=content.runner_object_path,
+ command_override=list(content.command_override),
+ )
+
+ def to_message(self) -> workflows_pb2.ReleaseContent:
+ """Convert a ReleaseContent object to a ReleaseContent protobuf message."""
+ return workflows_pb2.ReleaseContent(
+ fingerprint=self.fingerprint,
+ tasks=[task.to_message() for task in self.tasks],
+ files=[path.to_message() for path in self.files],
+ runner_object_path=self.runner_object_path,
+ command_override=self.command_override,
+ )
@dataclass(order=True, frozen=True)
@@ -336,15 +421,82 @@ class Workflow:
slug: str
name: str
description: str
+ releases: list["WorkflowRelease"] = field(default_factory=list)
@classmethod
def from_message(cls, workflow: workflows_pb2.Workflow) -> "Workflow":
"""Convert a Workflow protobuf message to a Workflow object."""
- return cls(slug=workflow.slug, name=workflow.name, description=workflow.description)
+ return cls(
+ slug=workflow.slug,
+ name=workflow.name,
+ description=workflow.description,
+ releases=[WorkflowRelease.from_message(release) for release in workflow.releases],
+ )
def to_message(self) -> workflows_pb2.Workflow:
"""Convert a Workflow object to a Workflow protobuf message."""
- return workflows_pb2.Workflow(slug=self.slug, name=self.name, description=self.description)
+ return workflows_pb2.Workflow(
+ slug=self.slug,
+ name=self.name,
+ description=self.description,
+ releases=[release.to_message() for release in self.releases],
+ )
+
+
+@dataclass(order=True, frozen=True)
+class WorkflowRelease:
+ id: UUID
+ artifact: Artifact | None
+ content: ReleaseContent | None
+ created_at: datetime | None
+ clusters: list[Cluster] = field(default_factory=list)
+
+ @classmethod
+ def from_message(cls, release: workflows_pb2.WorkflowRelease) -> "WorkflowRelease":
+ """Convert a WorkflowRelease protobuf message to a WorkflowRelease object."""
+ return cls(
+ id=uuid_message_to_uuid(release.id),
+ artifact=Artifact.from_message(release.artifact) if release.HasField("artifact") else None,
+ content=ReleaseContent.from_message(release.content) if release.HasField("content") else None,
+ created_at=timestamp_to_datetime(release.created_at) if release.HasField("created_at") else None,
+ clusters=[Cluster.from_message(cluster, include_deployed_workflows=False) for cluster in release.clusters],
+ )
+
+ def to_message(self) -> workflows_pb2.WorkflowRelease:
+ """Convert a WorkflowRelease object to a WorkflowRelease protobuf message."""
+ return workflows_pb2.WorkflowRelease(
+ id=uuid_to_uuid_message(self.id),
+ artifact=self.artifact.to_message() if self.artifact is not None else None,
+ content=self.content.to_message() if self.content is not None else None,
+ created_at=datetime_to_timestamp(self.created_at) if self.created_at is not None else None,
+ clusters=[cluster.to_message() for cluster in self.clusters],
+ )
+
+
+@dataclass(order=True, frozen=True)
+class WorkflowReleaseDeployment:
+ release: WorkflowRelease
+ clusters: list[Cluster]
+
+ @classmethod
+ def from_deploy_message(
+ cls, deployment: workflows_pb2.DeployWorkflowReleaseResponse
+ ) -> "WorkflowReleaseDeployment":
+ """Convert a DeployWorkflowReleaseResponse protobuf message to a WorkflowReleaseDeployment object."""
+ return cls(
+ release=WorkflowRelease.from_message(deployment.release),
+ clusters=[Cluster.from_message(cluster) for cluster in deployment.clusters],
+ )
+
+ @classmethod
+ def from_undeploy_message(
+ cls, deployment: workflows_pb2.UndeployWorkflowReleaseResponse
+ ) -> "WorkflowReleaseDeployment":
+ """Convert an UndeployWorkflowReleaseResponse protobuf message to a WorkflowReleaseDeployment object."""
+ return cls(
+ release=WorkflowRelease.from_message(deployment.release),
+ clusters=[Cluster.from_message(cluster) for cluster in deployment.clusters],
+ )
@dataclass
@@ -1045,6 +1197,7 @@ class QueryFilters:
job_states: list[JobState]
name: str | None
task_states: list[TaskState]
+ cluster_slugs: list[str] = field(default_factory=list)
@classmethod
def from_message(cls, filters: job_pb2.QueryFilters) -> "QueryFilters":
@@ -1057,6 +1210,7 @@ def from_message(cls, filters: job_pb2.QueryFilters) -> "QueryFilters":
job_states=[_JOB_STATES[state] for state in filters.states],
name=filters.name or None,
task_states=[_TASK_STATES[state] for state in filters.task_states],
+ cluster_slugs=list(filters.cluster_slugs),
)
def to_message(self) -> job_pb2.QueryFilters:
@@ -1071,4 +1225,5 @@ def to_message(self) -> job_pb2.QueryFilters:
task_states=[cast(core_pb2.TaskState, state.value) for state in self.task_states]
if self.task_states
else None,
+ cluster_slugs=self.cluster_slugs or None,
)
diff --git a/tilebox-workflows/tilebox/workflows/jobs/client.py b/tilebox-workflows/tilebox/workflows/jobs/client.py
index 60bab63..3d0c02b 100644
--- a/tilebox-workflows/tilebox/workflows/jobs/client.py
+++ b/tilebox-workflows/tilebox/workflows/jobs/client.py
@@ -209,13 +209,14 @@ def visualize(self, job: JobIDLike, direction: str = "down", layout: str = "dagr
"""
return self._service.visualize(_to_uuid(job), direction, layout, sketchy)
- def query(
+ def query( # noqa: PLR0913
self,
temporal_extent: "TimeIntervalLike | IDIntervalLike",
automation_ids: UUID | list[UUID] | None = None,
job_states: JobState | list[JobState] | None = None,
name: str | None = None,
task_states: TaskState | list[TaskState] | None = None,
+ clusters: ClusterSlugLike | list[ClusterSlugLike] | None = None,
) -> list[Job]:
"""List jobs in the given temporal extent.
@@ -240,6 +241,8 @@ def query(
case-insensitive and uses a fuzzy matching scheme.
task_states: A task state or list of task states to filter jobs by. If specified, only jobs that have at
least one task in any of the selected states are returned.
+ clusters: A cluster or list of clusters to filter jobs by. If specified, only jobs submitted to one of the
+ selected clusters are returned.
Returns:
A list of jobs matching the given filters.
@@ -287,6 +290,8 @@ def query(
if not isinstance(task_states, list):
task_states = [task_states]
+ cluster_slugs = _to_cluster_filter_slugs(clusters)
+
filters = QueryFilters(
time_interval=time_interval,
id_interval=id_interval,
@@ -294,6 +299,7 @@ def query(
job_states=job_states,
name=name,
task_states=task_states,
+ cluster_slugs=cluster_slugs,
)
def request(page: PaginationProtocol) -> QueryJobsResponse:
@@ -315,3 +321,15 @@ def _to_uuid(job_or_id: Job | UUID | str) -> UUID:
if isinstance(job_or_id, str):
return UUID(job_or_id)
return job_or_id
+
+
+def _to_cluster_filter_slugs(clusters: ClusterSlugLike | list[ClusterSlugLike] | None) -> list[str]:
+ if clusters is None:
+ return []
+ if isinstance(clusters, ClusterSlugLike):
+ cluster_slugs = [to_cluster_slug(clusters)]
+ else:
+ cluster_slugs = [to_cluster_slug(cluster) for cluster in clusters]
+ if any(cluster_slug == "" for cluster_slug in cluster_slugs):
+ raise ValueError("Cluster filters must use explicit cluster slugs")
+ return cluster_slugs
diff --git a/tilebox-workflows/tilebox/workflows/runner/__main__.py b/tilebox-workflows/tilebox/workflows/runner/__main__.py
index b81a609..7f4910c 100644
--- a/tilebox-workflows/tilebox/workflows/runner/__main__.py
+++ b/tilebox-workflows/tilebox/workflows/runner/__main__.py
@@ -1,13 +1,20 @@
import argparse
import importlib
+import os
+import sys
from collections.abc import Sequence
+from time import perf_counter
from typing import Any
+from loguru import logger
+
from tilebox.workflows.runner.runner import Runner
from tilebox.workflows.runner.worker_server import serve_runner
def main(argv: Sequence[str] | None = None) -> int:
+ _configure_logging()
+
parser = argparse.ArgumentParser(
prog="python -m tilebox.workflows.runner",
description="Start a Tilebox workflow worker runtime.",
@@ -15,10 +22,27 @@ def main(argv: Sequence[str] | None = None) -> int:
parser.add_argument("runner", help="Runner object import path, for example 'my_workflow.runner:runner'.")
args = parser.parse_args(argv)
- serve_runner(_import_runner(args.runner))
+ logger.debug(f"Starting Tilebox workflow runtime for runner {args.runner!r}")
+ runner = _import_runner(args.runner)
+ logger.debug(f"Imported runner {args.runner!r}; starting worker server")
+ serve_runner(runner)
+ logger.debug("Worker server stopped")
return 0
+def _configure_logging() -> None:
+ level = "DEBUG" if _is_debug_enabled() else "INFO"
+ logger.remove()
+ logger.add(sys.stderr, level=level, format="{process}: {level}: {message}", catch=True)
+
+
+def _is_debug_enabled() -> bool:
+ value = os.environ.get("TILEBOX_DEBUG")
+ if value is None:
+ return False
+ return value.strip().lower() in {"", "1", "true", "yes", "on"}
+
+
def _import_runner(import_path: str) -> Runner:
module_name, separator, object_path = import_path.partition(":")
if not module_name or not separator or not object_path:
@@ -26,15 +50,24 @@ def _import_runner(import_path: str) -> Runner:
"Expected runner import path in the format ':