PR feedback

Author: andystaples

docs/features.md

@@ -529,8 +529,6 @@ inheritance required — it's a `typing.Protocol`) to send exports to
 any destination (S3, GCS, SFTP, local filesystem, a database, etc.):
 
 ```python
-from typing import Optional
-
 from durabletask.extensions.history_export import HistoryWriter
 
 
@@ -542,13 +540,18 @@ class LocalFileSystemHistoryWriter:
         self,
         *,
         instance_id: str,
+        container: str,
         blob_name: str,
         payload: bytes,
         content_type: str,
-        content_encoding: Optional[str],
+        content_encoding: str | None,
     ) -> None:
         import os
-        path = os.path.join(self._root, blob_name)
+        # ``container`` is the destination's logical container name
+        # (an ExportDestination.container).  Per-job routing writers
+        # combine it with ``blob_name``; writers that pin to a fixed
+        # location at construction time may ignore it.
+        path = os.path.join(self._root, container, blob_name)
         os.makedirs(os.path.dirname(path), exist_ok=True)
         with open(path, "wb") as fp:
             fp.write(payload)

durabletask/extensions/history_export/activities.py

@@ -149,6 +149,10 @@ def export_instance_history(
         raise TypeError("format must be a mapping")
     fmt = ExportFormat.from_dict(cast("Mapping[str, Any]", fmt_input))
     destination_raw: Mapping[str, Any] = input.get("destination") or {}
+    container_raw: Any = destination_raw.get("container")
+    if not container_raw:
+        raise ValueError("destination.container is required")
+    container: str = str(container_raw)
     prefix_raw: Any = destination_raw.get("prefix")
     prefix: str | None = str(prefix_raw) if prefix_raw is not None else None
 
@@ -169,6 +173,7 @@ def export_instance_history(
         blob_name = _blob_name_for(instance_id=instance_id, prefix=prefix, fmt=fmt)
         ctx.writer.write(
             instance_id=instance_id,
+            container=container,
             blob_name=blob_name,
             payload=payload,
             content_type=content_type_for(fmt),

durabletask/extensions/history_export/azure_blob.py

@@ -65,11 +65,19 @@ class AzureBlobHistoryExportWriterOptions:
     def __post_init__(self) -> None:
         if not self.container_name:
             raise ValueError("container_name is required")
+        if self.connection_string and self.account_url:
+            raise ValueError(
+                "'connection_string' and 'account_url' are mutually exclusive"
+            )
         if not self.connection_string and not self.account_url:
             raise ValueError(
                 "Either 'connection_string' or 'account_url' (with 'credential') "
                 "must be provided"
             )
+        if self.account_url and self.credential is None:
+            raise ValueError(
+                "'credential' is required when 'account_url' is provided"
+            )
 
 
 class AzureBlobHistoryExportWriter:
@@ -116,19 +124,33 @@ def write(
         self,
         *,
         instance_id: str,
+        container: str,
         blob_name: str,
         payload: bytes,
         content_type: str,
         content_encoding: str | None,
     ) -> None:
         del instance_id  # included by the protocol but not needed here
+        # This writer pins to the container configured at construction
+        # time and ignores the per-call ``container`` argument; the
+        # configured value is authoritative for any given writer
+        # instance.  Run a separate writer per destination container
+        # if you need per-job routing.
+        del container
         self._ensure_container()
         container_client = self._service.get_container_client(
             self._options.container_name
         )
-        content_settings = ContentSettings(
-            content_type=content_type,
-            content_encoding=content_encoding or "",
+        # Only set Content-Encoding if the format actually compresses
+        # the payload; an empty header value would be persisted on
+        # the blob and confuse downstream clients.
+        content_settings = (
+            ContentSettings(
+                content_type=content_type,
+                content_encoding=content_encoding,
+            )
+            if content_encoding
+            else ContentSettings(content_type=content_type)
         )
         container_client.upload_blob(
             name=blob_name,

durabletask/extensions/history_export/client.py

@@ -219,7 +219,10 @@ def list_jobs(
             instance_id_starts_with=_ENTITY_ID_PREFIX,
             last_modified_from=query.last_modified_from,
             last_modified_to=query.last_modified_to,
-            include_state=query.include_state,
+            # list_jobs always needs the persisted state to populate
+            # ExportJobDescription; an entity-only view doesn't carry
+            # status or progress and would always be filtered out.
+            include_state=True,
             page_size=query.page_size,
         )
         status_filter = set(query.status) if query.status else None
@@ -230,7 +233,7 @@ def list_jobs(
             # explicit entity-name check.
             if meta.id.entity != ENTITY_NAME.lower():
                 continue
-            raw = meta.get_state(str) if meta.includes_state else None
+            raw = meta.get_state(str)
             if not raw:
                 continue
             try:

durabletask/extensions/history_export/models.py

@@ -64,14 +64,12 @@ class ExportJobStatus(Enum):
     Status meanings
     ---------------
     ``PENDING``
-        The job has been created but the entity has not yet processed
-        the ``create`` signal, *or* the entity has accepted the
-        configuration but has not yet kicked off its driving
-        orchestrator.  The value is reserved for the forthcoming
-        ``run`` operation (see the .NET ``ExportJob.Run`` pattern);
-        the current implementation transitions directly from creation
-        to :attr:`ACTIVE`, so jobs are not persisted in ``Pending``
-        today.
+        The job has been created and persisted but the entity has not
+        yet kicked off its driving orchestrator.  Jobs sit in this
+        state briefly between the ``create`` and ``run`` signals
+        (the public client sends both in immediate succession), or
+        for longer if ``run`` is never invoked or if a caller revives
+        a previously terminal job via ``create``.
     ``ACTIVE``
         The job is running and the driving orchestrator is making
         progress through pages of terminal instances.
@@ -305,15 +303,12 @@ class ExportJobQuery:
             this timestamp are returned.
         page_size: Backend page size used to enumerate the underlying
             entities.
-        include_state: Whether to fetch full job state (set ``False``
-            to retrieve job IDs and metadata only).
     """
 
     status: list[ExportJobStatus] | None = None
     last_modified_from: datetime | None = None
     last_modified_to: datetime | None = None
     page_size: int | None = None
-    include_state: bool = True
 
 
 @dataclass

durabletask/extensions/history_export/writer.py

@@ -26,13 +26,18 @@ def write(
             self,
             *,
             instance_id: str,
+            container: str,
             blob_name: str,
             payload: bytes,
             content_type: str,
             content_encoding: str | None,
         ) -> None:
             import os
-            path = os.path.join(self._root, blob_name)
+            # The ``container`` value comes from the export job's
+            # ExportDestination.container and is the logical
+            # bucket / subdirectory the caller asked the job to
+            # write into.
+            path = os.path.join(self._root, container, blob_name)
             os.makedirs(os.path.dirname(path), exist_ok=True)
             with open(path, "wb") as fp:
                 fp.write(payload)
@@ -63,6 +68,7 @@ def write(
         self,
         *,
         instance_id: str,
+        container: str,
         blob_name: str,
         payload: bytes,
         content_type: str,
@@ -74,8 +80,17 @@ def write(
             instance_id: The orchestration instance whose history this
                 payload represents.  Provided so destinations may use
                 it as a key, metadata, or sharding hint.
+            container: The destination container / bucket name the
+                job's :class:`ExportDestination` declared.  Writers
+                that want to honour per-job container routing should
+                use this value; writers that pin to a fixed container
+                at construction time (such as the bundled Azure Blob
+                writer) may ignore it.
             blob_name: Destination-relative path / key, including any
                 configured destination prefix and file extension.
+                Does NOT include the ``container`` component — a
+                writer that routes per-container is expected to
+                combine the two.
             payload: The serialized history bytes.  Already compressed
                 if the configured format calls for it.
             content_type: The HTTP-style content type appropriate for

examples/history_export/app.py

@@ -59,7 +59,12 @@ def sample_orchestrator(ctx: task.OrchestrationContext, n: int):
 
 def main() -> None:
     print(f"Using container: {CONTAINER_NAME}")
-    print(f"Using storage connection: {AZURITE_CONN_STR}")
+    # Avoid printing the raw connection string — a real Azure Storage
+    # connection string contains the account key.
+    if AZURITE_CONN_STR == "UseDevelopmentStorage=true":
+        print("Using storage connection: Azurite (UseDevelopmentStorage=true)")
+    else:
+        print("Using storage connection: (redacted)")
 
     backend = create_test_backend(port=50300)
     try:

tests/durabletask/extensions/history_export/test_activities.py

@@ -46,7 +46,7 @@ def __init__(self) -> None:
         self._lock = threading.Lock()
         self.blobs: dict[str, dict] = {}
 
-    def write(self, *, instance_id, blob_name, payload, content_type, content_encoding):
+    def write(self, *, instance_id, container, blob_name, payload, content_type, content_encoding):
         with self._lock:
             self.blobs[blob_name] = {
                 "instance_id": instance_id,

tests/durabletask/extensions/history_export/test_azure_blob_writer_e2e.py

@@ -93,6 +93,7 @@ def test_write_json_blob(writer):
     blob_name = f"json/{uuid.uuid4().hex}.json"
     writer.write(
         instance_id="inst-json",
+        container=TEST_CONTAINER,
         blob_name=blob_name,
         payload=payload,
         content_type=content_type_for(fmt),
@@ -118,6 +119,7 @@ def test_write_jsonl_gzip_blob(writer):
     blob_name = f"gz/{uuid.uuid4().hex}.jsonl.gz"
     writer.write(
         instance_id="inst-gz",
+        container=TEST_CONTAINER,
         blob_name=blob_name,
         payload=payload,
         content_type=content_type_for(fmt),

tests/durabletask/extensions/history_export/test_client.py

@@ -46,7 +46,7 @@ def __init__(self) -> None:
         self._lock = threading.Lock()
         self.blobs: dict[str, dict] = {}
 
-    def write(self, *, instance_id, blob_name, payload, content_type, content_encoding):
+    def write(self, *, instance_id, container, blob_name, payload, content_type, content_encoding):
         with self._lock:
             self.blobs[blob_name] = {
                 "instance_id": instance_id,