Resolve external payload storage policies

durable-workflow-ops · durable-workflow-ops · commit 4c6ce2fbbbbc · 2026-04-22T10:43:31.000Z
diff --git a/README.md b/README.md
@@ -215,15 +215,24 @@ payload storage.
 Retention cleanup should delete by typed reference rather than by raw URI:
 `delete_external_payload(storage, reference, cache=cache)` calls the configured
 driver and evicts any verified replay-cache entry for the same reference.
+When the server or Cloud API returns an external payload storage policy, use
+`ExternalPayloadStoragePolicy.from_dict(...)` plus
+`external_storage_driver_from_policy(...)` to turn that control-plane payload
+into the matching SDK driver while keeping provider clients application-owned.
 
 ```python
-from durable_workflow import S3ExternalStorage, serializer
+from durable_workflow import (
+    ExternalPayloadStoragePolicy,
+    external_storage_driver_from_policy,
+    serializer,
+)
 
-storage = S3ExternalStorage(s3_client, bucket="workflow-payloads", prefix="prod")
+policy = ExternalPayloadStoragePolicy.from_dict(namespace_response)
+storage = external_storage_driver_from_policy(policy, s3_client=s3_client)
 payload = serializer.external_storage_envelope(
     {"large": "value"},
     external_storage=storage,
-    threshold_bytes=2 * 1024 * 1024,
+    threshold_bytes=policy.threshold_bytes or 2 * 1024 * 1024,
 )
 ```
 
diff --git a/src/durable_workflow/__init__.py b/src/durable_workflow/__init__.py
@@ -67,11 +67,13 @@
     ExternalPayloadCache,
     ExternalPayloadIntegrityError,
     ExternalPayloadReference,
+    ExternalPayloadStoragePolicy,
     ExternalStorageDriver,
     GCSExternalStorage,
     LocalFilesystemExternalStorage,
     S3ExternalStorage,
     delete_external_payload,
+    external_storage_driver_from_policy,
     fetch_external_payload,
     store_external_payload,
 )
@@ -197,6 +199,7 @@
     "ExternalPayloadCache",
     "ExternalPayloadIntegrityError",
     "ExternalPayloadReference",
+    "ExternalPayloadStoragePolicy",
     "ExternalStorageDriver",
     "ExternalTaskInput",
     "ExternalTaskInputError",
@@ -244,6 +247,7 @@
     "EXTERNAL_TASK_RESULT_VERSION",
     "delete_external_payload",
     "external_storage_envelope",
+    "external_storage_driver_from_policy",
     "fetch_external_payload",
     "parse_external_task_input",
     "parse_external_task_input_artifact",
diff --git a/src/durable_workflow/external_storage.py b/src/durable_workflow/external_storage.py
@@ -4,7 +4,8 @@
 
 import hashlib
 from collections import OrderedDict
-from dataclasses import dataclass
+from collections.abc import Mapping
+from dataclasses import dataclass, field
 from datetime import datetime
 from pathlib import Path
 from typing import Any, Protocol
@@ -30,6 +31,47 @@ def delete(self, uri: str) -> None:
         """Delete previously persisted payload bytes when retention removes a run."""
 
 
+@dataclass(frozen=True)
+class ExternalPayloadStoragePolicy:
+    """Normalized external payload storage policy from server or Cloud APIs."""
+
+    enabled: bool
+    driver: str | None = None
+    threshold_bytes: int | None = None
+    config: Mapping[str, Any] = field(default_factory=dict)
+    reference: str | None = None
+    prefix: str = ""
+    mode: str | None = None
+    status: str | None = None
+    integrity_required: bool = True
+
+    @classmethod
+    def from_dict(cls, data: object) -> ExternalPayloadStoragePolicy:
+        """Parse a server namespace or Cloud organization storage policy."""
+        policy = _extract_policy(data)
+        driver = _optional_string(policy.get("driver"), "external payload storage driver")
+        enabled = bool(policy.get("enabled", driver is not None))
+        threshold_bytes = _optional_positive_int(policy.get("threshold_bytes"), "threshold_bytes")
+        config = _optional_mapping(policy.get("config"), "config")
+        prefix = _optional_string(policy.get("prefix"), "prefix") or _optional_string(
+            config.get("prefix"),
+            "config.prefix",
+        )
+        integrity_required = bool(policy.get("integrity_required", True))
+
+        return cls(
+            enabled=enabled,
+            driver=driver,
+            threshold_bytes=threshold_bytes,
+            config=config,
+            reference=_optional_string(policy.get("reference"), "reference"),
+            prefix=prefix or "",
+            mode=_optional_string(policy.get("mode"), "mode"),
+            status=_optional_string(policy.get("status"), "status"),
+            integrity_required=integrity_required,
+        )
+
+
 @dataclass(frozen=True)
 class ExternalPayloadReference:
     """Stable wire envelope for a payload stored outside workflow history."""
@@ -321,6 +363,70 @@ def delete(self, uri: str) -> None:
         self.container_client.delete_blob(key)
 
 
+def external_storage_driver_from_policy(
+    policy: ExternalPayloadStoragePolicy | Mapping[str, Any],
+    *,
+    s3_client: Any | None = None,
+    gcs_client: Any | None = None,
+    azure_container_client: Any | None = None,
+    local_root: str | Path | None = None,
+) -> ExternalStorageDriver:
+    """Build an SDK storage driver from a server or Cloud policy payload.
+
+    Provider SDK clients remain application-owned. Pass the already-configured
+    S3/GCS/Azure client that matches the policy returned by the control plane.
+    """
+    normalized = (
+        policy
+        if isinstance(policy, ExternalPayloadStoragePolicy)
+        else ExternalPayloadStoragePolicy.from_dict(policy)
+    )
+    if not normalized.enabled:
+        raise ValueError("external payload storage policy is disabled")
+    if normalized.driver is None:
+        raise ValueError("external payload storage policy driver is required")
+
+    driver = normalized.driver.lower()
+    if driver == "local":
+        root = local_root or _policy_string(normalized, "uri") or normalized.reference
+        if root is None:
+            raise ValueError("local external payload storage policy requires config.uri or local_root")
+        return LocalFilesystemExternalStorage(_local_path(root))
+
+    if driver == "s3":
+        if s3_client is None:
+            raise ValueError("s3 external payload storage policy requires s3_client")
+        bucket = _policy_string(normalized, "bucket") or _s3_bucket_from_reference(normalized.reference)
+        if bucket is None:
+            raise ValueError("s3 external payload storage policy requires config.bucket or an S3 bucket reference")
+        return S3ExternalStorage(s3_client, bucket=bucket, prefix=_policy_prefix(normalized))
+
+    if driver == "gcs":
+        if gcs_client is None:
+            raise ValueError("gcs external payload storage policy requires gcs_client")
+        bucket = _policy_string(normalized, "bucket") or _gcs_bucket_from_reference(normalized.reference)
+        if bucket is None:
+            raise ValueError("gcs external payload storage policy requires config.bucket or a GCS bucket reference")
+        return GCSExternalStorage(gcs_client, bucket=bucket, prefix=_policy_prefix(normalized))
+
+    if driver in {"azure", "azure-blob"}:
+        if azure_container_client is None:
+            raise ValueError("azure external payload storage policy requires azure_container_client")
+        container = (
+            _policy_string(normalized, "container")
+            or _policy_string(normalized, "bucket")
+            or _azure_container_from_reference(normalized.reference)
+        )
+        if container is None:
+            raise ValueError(
+                "azure external payload storage policy requires config.container, "
+                "config.bucket, or a container reference"
+            )
+        return AzureBlobExternalStorage(azure_container_client, container=container, prefix=_policy_prefix(normalized))
+
+    raise ValueError(f"unsupported external payload storage driver {normalized.driver!r}")
+
+
 def store_external_payload(
     driver: ExternalStorageDriver,
     data: bytes,
@@ -387,6 +493,96 @@ def _validate_sha256(sha256: str) -> None:
         raise ValueError("sha256 must be a hex digest") from exc
 
 
+def _extract_policy(data: object) -> Mapping[str, Any]:
+    if not isinstance(data, Mapping):
+        raise ValueError("external payload storage policy must be an object")
+    nested = data.get("external_payload_storage")
+    if nested is not None:
+        if not isinstance(nested, Mapping):
+            raise ValueError("external_payload_storage must be an object")
+        return nested
+    return data
+
+
+def _optional_mapping(data: object, field_name: str) -> Mapping[str, Any]:
+    if data is None:
+        return {}
+    if not isinstance(data, Mapping):
+        raise ValueError(f"external payload storage policy {field_name} must be an object")
+    return data
+
+
+def _optional_string(data: object, field_name: str) -> str | None:
+    if data is None:
+        return None
+    if not isinstance(data, str):
+        raise ValueError(f"external payload storage policy {field_name} must be a string")
+    return data
+
+
+def _optional_positive_int(data: object, field_name: str) -> int | None:
+    if data is None:
+        return None
+    if not isinstance(data, int) or data < 1:
+        raise ValueError(f"external payload storage policy {field_name} must be a positive integer")
+    return data
+
+
+def _policy_string(policy: ExternalPayloadStoragePolicy, key: str) -> str | None:
+    return _optional_string(policy.config.get(key), f"config.{key}")
+
+
+def _policy_prefix(policy: ExternalPayloadStoragePolicy) -> str:
+    return _policy_string(policy, "prefix") or policy.prefix
+
+
+def _local_path(root: str | Path) -> str | Path:
+    if isinstance(root, Path):
+        return root
+    parsed = urlparse(root)
+    if parsed.scheme == "file":
+        if parsed.netloc not in {"", "localhost"}:
+            raise ValueError("local external payload storage file URI must be local")
+        return unquote(parsed.path)
+    if parsed.scheme:
+        raise ValueError("local external payload storage policy must use a file:// URI or filesystem path")
+    return root
+
+
+def _s3_bucket_from_reference(reference: str | None) -> str | None:
+    if reference is None:
+        return None
+    if reference.startswith("arn:aws:s3:::"):
+        bucket = reference.removeprefix("arn:aws:s3:::").split("/", 1)[0]
+        return bucket or None
+    parsed = urlparse(reference)
+    if parsed.scheme == "s3" and parsed.netloc:
+        return parsed.netloc
+    return reference or None
+
+
+def _gcs_bucket_from_reference(reference: str | None) -> str | None:
+    if reference is None:
+        return None
+    marker = "/buckets/"
+    if marker in reference:
+        bucket = reference.rsplit(marker, 1)[1].split("/", 1)[0]
+        return bucket or None
+    parsed = urlparse(reference)
+    if parsed.scheme == "gs" and parsed.netloc:
+        return parsed.netloc
+    return reference or None
+
+
+def _azure_container_from_reference(reference: str | None) -> str | None:
+    if reference is None:
+        return None
+    parsed = urlparse(reference)
+    if parsed.scheme in {"azure", "azure-blob"} and parsed.netloc:
+        return parsed.netloc
+    return reference or None
+
+
 def _validate_rfc3339(value: str) -> None:
     normalized = value[:-1] + "+00:00" if value.endswith("Z") else value
     try:
diff --git a/tests/test_external_storage.py b/tests/test_external_storage.py
@@ -10,10 +10,12 @@
     ExternalPayloadCache,
     ExternalPayloadIntegrityError,
     ExternalPayloadReference,
+    ExternalPayloadStoragePolicy,
     GCSExternalStorage,
     LocalFilesystemExternalStorage,
     S3ExternalStorage,
     delete_external_payload,
+    external_storage_driver_from_policy,
     fetch_external_payload,
     store_external_payload,
 )
@@ -419,3 +421,99 @@ def test_azure_external_storage_round_trips_and_deletes_payload() -> None:
 def test_object_storage_rejects_unsafe_prefix() -> None:
     with pytest.raises(ValueError, match="unsafe"):
         S3ExternalStorage(FakeS3Client(), bucket="payloads", prefix="../tenant-a")
+
+
+def test_external_storage_policy_builds_s3_driver_from_server_namespace_policy() -> None:
+    client = FakeS3Client()
+    policy = ExternalPayloadStoragePolicy.from_dict(
+        {
+            "external_payload_storage": {
+                "driver": "s3",
+                "enabled": True,
+                "threshold_bytes": 2_097_152,
+                "config": {
+                    "bucket": "dw-payloads",
+                    "prefix": "billing/",
+                },
+            }
+        }
+    )
+
+    storage = external_storage_driver_from_policy(policy, s3_client=client)
+    assert isinstance(storage, S3ExternalStorage)
+    assert policy.threshold_bytes == 2_097_152
+
+    reference = store_external_payload(storage, b'{"from":"server-policy"}', codec="json")
+
+    assert reference.uri.startswith("s3://dw-payloads/billing/json/")
+    assert fetch_external_payload(storage, reference) == b'{"from":"server-policy"}'
+
+
+def test_external_storage_policy_builds_gcs_driver_from_cloud_reference() -> None:
+    client = FakeGCSClient()
+    policy = ExternalPayloadStoragePolicy.from_dict(
+        {
+            "enabled": True,
+            "mode": "byob",
+            "driver": "gcs",
+            "reference": "projects/acme/buckets/workflow-payloads",
+            "prefix": "prod",
+            "threshold_bytes": 1_500_000,
+            "status": "pending_validation",
+        }
+    )
+
+    storage = external_storage_driver_from_policy(policy, gcs_client=client)
+
+    reference = store_external_payload(storage, b'{"from":"cloud-policy"}', codec="json")
+
+    assert reference.uri.startswith("gs://workflow-payloads/prod/json/")
+    assert fetch_external_payload(storage, reference) == b'{"from":"cloud-policy"}'
+
+
+def test_external_storage_policy_builds_local_driver_from_file_uri(tmp_path: Path) -> None:
+    root = tmp_path / "payloads"
+    storage = external_storage_driver_from_policy(
+        {
+            "driver": "local",
+            "enabled": True,
+            "config": {"uri": root.as_uri()},
+        }
+    )
+
+    reference = store_external_payload(storage, b"local-policy", codec="json")
+
+    assert fetch_external_payload(storage, reference) == b"local-policy"
+
+
+def test_external_storage_policy_builds_azure_driver_from_container_config() -> None:
+    client = FakeAzureContainerClient()
+    storage = external_storage_driver_from_policy(
+        {
+            "driver": "azure",
+            "enabled": True,
+            "config": {"container": "payloads", "prefix": "tenant-a"},
+        },
+        azure_container_client=client,
+    )
+
+    reference = store_external_payload(storage, b'{"from":"azure-policy"}', codec="json")
+
+    assert reference.uri.startswith("azure-blob://payloads/tenant-a/json/")
+    assert fetch_external_payload(storage, reference) == b'{"from":"azure-policy"}'
+
+
+def test_external_storage_policy_rejects_disabled_or_missing_provider_client() -> None:
+    with pytest.raises(ValueError, match="disabled"):
+        external_storage_driver_from_policy({"driver": "s3", "enabled": False, "config": {"bucket": "payloads"}})
+
+    with pytest.raises(ValueError, match="s3_client"):
+        external_storage_driver_from_policy({"driver": "s3", "enabled": True, "config": {"bucket": "payloads"}})
+
+
+def test_external_storage_policy_validates_threshold_and_config_shape() -> None:
+    with pytest.raises(ValueError, match="threshold_bytes"):
+        ExternalPayloadStoragePolicy.from_dict({"driver": "local", "threshold_bytes": 0})
+
+    with pytest.raises(ValueError, match="config"):
+        ExternalPayloadStoragePolicy.from_dict({"driver": "local", "config": "file:///tmp/payloads"})
