Within the broader rights metadata mapping and licensing automation pipeline, an embargo workflow is the machinery that lets an asset become public on a schedule rather than on a manual curator action. A donor agreement seals a bequest until 2030; a publication ban lifts the morning after an exhibition opens; a statutory copyright term lapses on a fixed January 1. In each case the institution has already decided when the asset may open — the workflow’s job is to make that decision fire on time, exactly once, and only after a final rights check confirms nothing else still blocks release. This page specifies the record contract for a time-bound asset, the evaluation engine that compares boundaries against the clock, the routing that carries an expired embargo through a secondary verification instead of straight to publication, and the deployment pattern that keeps a nightly batch idempotent.

The single design rule that governs everything below is that embargo expiration is not publication. When embargo_end passes, the asset does not become public; it becomes eligible for a rights re-check. Only after that check clears does it reach IIIF delivery. Collapsing the two steps is the most common and most damaging mistake in this domain, because a donor hold and a copyright term are independent constraints: an embargo can lapse while the work is still in copyright, and publishing it the instant the calendar ticks over is an unauthorized disclosure. The engine here always routes an expired embargo to a pending_review state that a downstream copyright verdict must clear before the asset resolves to anyone.

Workflow Context

Three institutional roles meet at this stage. The collections manager owns the donor agreements and needs an audit trail proving that no embargoed asset surfaced before its agreed date and that every release was gated on a rights check. The Python automation engineer owns the evaluator and needs an executable contract that fails loudly on a malformed date rather than defaulting a locked asset to open. The DAMS administrator owns the schedule and needs the run to be idempotent, so that a nightly cron that overlaps a previous slow run — or a retry after a crash — never emits a duplicate publish event or races an expiration trigger.

This stage is a sibling to the other three rights subsystems and depends on one of them directly. Records enter already carrying a resolved copyright verdict from automating copyright status checks; that engine sends any still-protected asset with a defined release date here, to be held until the term lapses. When an embargo lifts and the re-check clears, an open-licensed asset is handed to routing Creative Commons licenses for its final license tagging, while an asset released on the strength of an expired copyright term depends on the same rolling-cutoff arithmetic examined under threshold tuning for public domain. The boundary-calculation edge cases — daylight-saving drift, memory pressure on large tables, the strict-versus-inclusive comparison at the exact expiry moment — are covered in depth in the child guide on setting date-based embargo triggers.

The engine is a scheduled, asynchronous evaluator, deliberately decoupled from ingestion. It polls the metadata store for embargoed records, evaluates each one’s temporal boundary against a single captured now, and emits state-change events. Nothing about it is triggered by an asset arriving; it is triggered by the clock, which is why the whole workflow can be reasoned about as a batch that is safe to run again.

Prerequisites

Before wiring the embargo evaluator, confirm the following are in place:

  • Python 3.9+ for the PEP 604 | union syntax, zoneinfo from the standard library, and the asyncio primitives used in the batch loop.
  • pydantic (v2) for the record contract — this page uses the v2 API (field_validator, model_dump, ConfigDict); see the pydantic v2 documentation for validator signatures.
  • aiohttp (3.9+) as the async client for paginated fetches and event dispatch, configured with a bounded connection pool and per-request timeouts.
  • A metadata store queryable by embargo state, exposing an endpoint that returns only records currently in the active embargo state so the poller never rescans the whole collection.
  • A message broker or event sink (Kafka, SNS/SQS, or a database outbox) to carry pending_review and published state-change events to CMS and manifest consumers.
  • A UTC clock discipline. Every stored boundary must be timezone-aware; the evaluator captures one now = datetime.now(timezone.utc) per run and compares every record against that same instant.
  • Interchange standards for the surviving payload. Embargo state must serialize into the LIDO schema rights elements, and access itself is enforced through the IIIF Presentation API 3.0 rights property plus the separate IIIF Authorization Flow API, never by hiding the manifest.

Schema & State Reference

The record contract fixes exactly what the evaluator may reason about, and the state vocabulary is closed so a downstream consumer can switch on a known set of values. A strict Pydantic v2 model normalizes every boundary to UTC at construction, so no naive datetime ever reaches the comparison.

Field Type Constraint Purpose
asset_id str required Idempotency and audit key
embargo_start datetime required, UTC-normalized Start of the restricted window
embargo_end datetime | None UTC-normalized; None = indefinite Expiry boundary the clock is compared against
rights_statement str | None RightsStatements.org URI Rights assertion carried into the re-check
correlation_id str required Deduplicates emitted state-change events
state EmbargoState one of the four below Current position in the lifecycle

The state vocabulary is closed and each value has exactly one legal successor set:

State Meaning Next legal states
active Embargo in force; asset non-resolvable expired
expired Boundary passed; awaiting rights re-check pending_review
pending_review Secondary copyright check in progress published, active (re-embargo)
published Cleared and delivered to public endpoints terminal

An indefinite embargo (embargo_end is None) never leaves active on its own — it can only be lifted by a curator supplying an end date, which is why the evaluator treats a null boundary as “remain active” rather than as an error.

Step-by-Step Implementation

The evaluator is split into three responsibilities — the record contract, the pure state evaluation, and the async batch loop — each presented in turn and assembled in main(). The stack uses pydantic for schema enforcement, aiohttp with a semaphore for bounded concurrency, and explicit routing rather than a scoring heuristic.

1. Define and normalize the record contract

The ingestion boundary coerces raw store rows into a strict model. Boundaries arrive as ISO-8601 strings or naive datetimes and are forced to UTC before anything compares them, so a stored 2030-01-01T00:00:00 is never silently interpreted in server-local time.

python
import asyncio
import logging
from datetime import datetime, timezone
from enum import Enum
from typing import Optional, Union, Any
from pydantic import BaseModel, Field, field_validator
import aiohttp

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(message)s"
)
logger = logging.getLogger("embargo_pipeline")

class EmbargoState(str, Enum):
    ACTIVE = "active"
    EXPIRED = "expired"
    PENDING_REVIEW = "pending_review"
    PUBLISHED = "published"

class AssetMetadata(BaseModel):
    asset_id: str
    embargo_start: datetime
    embargo_end: Optional[datetime] = None
    rights_statement: Optional[str] = None
    correlation_id: str
    state: EmbargoState = EmbargoState.ACTIVE

    @field_validator("embargo_start", "embargo_end", mode="before")
    @classmethod
    def enforce_utc(cls, v: Optional[Union[str, datetime]]) -> Optional[datetime]:
        if v is None:
            return None
        if isinstance(v, str):
            # Accept trailing 'Z' as UTC, which fromisoformat rejects before 3.11.
            v = datetime.fromisoformat(v.replace("Z", "+00:00"))
        # A naive datetime is assumed UTC; an aware one is converted to UTC.
        return v.replace(tzinfo=timezone.utc) if v.tzinfo is None else v.astimezone(timezone.utc)

The correlation_id is required rather than optional because every downstream event is deduplicated on it — a record with no correlation key cannot be published safely, so the contract refuses to construct one.

2. Evaluate the boundary against a single captured clock

Evaluation is a pure function of a record and one now. It never calls datetime.now() internally, because a batch that reads the clock per-record can straddle a midnight boundary and evaluate two otherwise-identical records inconsistently. Expiry uses >= so an asset expires at its boundary, not one instant after.

python
def evaluate_embargo_state(asset: AssetMetadata, now: datetime) -> EmbargoState:
    """Determine the embargo state for one asset against a fixed UTC instant."""
    if asset.state in (EmbargoState.PENDING_REVIEW, EmbargoState.PUBLISHED):
        # Already past the temporal gate; the rights re-check owns it now.
        return asset.state
    if asset.embargo_end is None:
        return EmbargoState.ACTIVE          # indefinite hold, never auto-lifts
    if now >= asset.embargo_end:
        return EmbargoState.EXPIRED         # eligible for re-check, NOT published
    return EmbargoState.ACTIVE

Note what the function deliberately does not do: it never returns PUBLISHED. Reaching the boundary can only move an asset to EXPIRED. Publication is a separate decision made after the rights re-check, encoded in the routing layer below.

3. Fetch batches and route through the re-check

The batch loop pages the store for active records, evaluates each against the run’s single now, and routes the expired ones into a secondary copyright verification before any publish event is emitted. A semaphore caps concurrency so a large collection never overruns the downstream API.

python
async def fetch_active_batch(session: aiohttp.ClientSession, offset: int, limit: int) -> list[dict[str, Any]]:
    """Page only currently-active embargoes from the metadata store."""
    async with session.get(
        "/api/v1/assets/embargoed",
        params={"state": "active", "offset": offset, "limit": limit},
        timeout=aiohttp.ClientTimeout(total=15),
    ) as resp:
        resp.raise_for_status()
        return await resp.json()

async def rights_recheck(session: aiohttp.ClientSession, asset: AssetMetadata) -> EmbargoState:
    """Secondary gate: an expired embargo only publishes if the rights check clears."""
    async with session.post(
        "/api/v1/rights/verify",
        json={"asset_id": asset.asset_id, "rights_statement": asset.rights_statement},
        timeout=aiohttp.ClientTimeout(total=15),
    ) as resp:
        resp.raise_for_status()
        verdict = (await resp.json()).get("verdict")
    # Cleared -> publish; still protected -> re-embargo; anything else -> hold for a human.
    if verdict == "cleared":
        return EmbargoState.PUBLISHED
    if verdict == "still_protected":
        return EmbargoState.ACTIVE
    return EmbargoState.PENDING_REVIEW

class EmbargoEvaluator:
    def __init__(self, session: aiohttp.ClientSession, max_concurrency: int = 10):
        self.session = session
        self.semaphore = asyncio.Semaphore(max_concurrency)

    async def process(self, asset: AssetMetadata, now: datetime) -> Optional[dict[str, Any]]:
        async with self.semaphore:
            new_state = evaluate_embargo_state(asset, now)
            if new_state != EmbargoState.EXPIRED:
                return None  # still active or already downstream; no event
            # Expired: run the mandatory rights re-check before publishing.
            resolved = await rights_recheck(self.session, asset)
            event = {
                "asset_id": asset.asset_id,
                "correlation_id": asset.correlation_id,
                "from_state": asset.state.value,
                "to_state": resolved.value,
            }
            logger.info("%s: %s -> %s", asset.asset_id, asset.state.value, resolved.value)
            await self._emit(event)
            return event

    async def _emit(self, event: dict[str, Any]) -> None:
        # Idempotency key = correlation_id; a duplicate delivery is a no-op downstream.
        async with self.session.post(
            "/api/v1/events/embargo",
            json=event,
            headers={"Idempotency-Key": event["correlation_id"]},
            timeout=aiohttp.ClientTimeout(total=15),
        ) as resp:
            if resp.status not in (200, 202, 409):  # 409 = already applied, fine
                logger.warning("emit failed for %s: %s", event["asset_id"], resp.status)

async def main() -> None:
    now = datetime.now(timezone.utc)  # one clock for the whole run
    conn = aiohttp.TCPConnector(limit=20)
    async with aiohttp.ClientSession(base_url="https://dam.internal", connector=conn) as session:
        evaluator = EmbargoEvaluator(session, max_concurrency=8)
        offset, limit = 0, 500
        while True:
            rows = await fetch_active_batch(session, offset, limit)
            if not rows:
                break
            assets = [AssetMetadata.model_validate(r) for r in rows]
            results = await asyncio.gather(
                *(evaluator.process(a, now) for a in assets),
                return_exceptions=True,
            )
            for r in results:
                if isinstance(r, Exception):
                    logger.error("record failed: %s", r)
            offset += limit

if __name__ == "__main__":
    asyncio.run(main())

The input variants converge on the same path: a CSV export whose date columns are strings is coerced by the enforce_utc validator; an XML payload carrying a Z-suffixed timestamp is normalized before comparison; and an API record with a null embargo_end flows deterministically back to active instead of raising. asyncio.gather(..., return_exceptions=True) guarantees one poisoned record cannot cancel the rest of the batch.

Embargo State Machine

The lifecycle is easiest to read as a state diagram: an active embargo can only reach expired, an expired one must pass the rights re-check before it can publish, and the re-check may send it back into active as a re-embargo rather than forward.

Embargo lifecycle — reaching the boundary routes to a re-check, never straight to publication Left to right: a start marker feeds the active state; active advances to expired on "end date reached"; expired advances to pending_review on "rights re-check"; pending_review advances to published (terminal) on "cleared". A dashed feedback arc carries a still-protected verdict from pending_review back to active as a re-embargo, so an expired embargo can only publish after the secondary rights check clears. Embargo lifecycle — expiry opens a re-check gate, it never publishes directly active held, non-resolvable expired eligible for re-check pending_review secondary rights gate published delivered, terminal end date reached rights re-check cleared re-embargo — still protected, copyright term governs

Rights and Access Routing

The evaluator’s output is a routing signal, not a publish command. An expired verdict opens the secondary gate; the gate’s own verdict — cleared, still protected, or indeterminate — decides the access tier. A cleared asset routes to publication and receives a permissive rights property on its IIIF manifest; where that clearance rests on an open license, the final tagging is handed to routing Creative Commons licenses. A still-protected asset is re-embargoed back to active — the calendar lapsed but the copyright term did not, so the more restrictive constraint governs, exactly as it does in automating copyright status checks. An indeterminate verdict parks the asset in pending_review for a registrar rather than guessing.

Access control itself is never implemented by hiding the manifest. The IIIF manifest’s rights property records what rights apply — a RightsStatements.org or Creative Commons URI — while the IIIF Authorization Flow API, referenced through the manifest’s service property, enforces who may view the asset at serving time. During an active embargo the manifest carries an in-copyright statement such as https://rightsstatements.org/vocab/InC/1.0/ and the Auth API withholds the image; when the embargo lifts and the re-check clears, the pipeline rewrites the rights URI and the Auth layer opens access in one atomic update. Keeping the two concerns separate is what lets an asset be publicly described while still being access-restricted, which is precisely the state an embargo represents.

Verification and Testing

The engine’s correctness rests on two boundary details — the inclusive >= at the expiry instant and the guarantee that reaching the boundary never publishes directly — so the tests pin both rather than trust them. The cases below are pure and run under pytest -q or as a plain script; none of them touches the network, because evaluate_embargo_state is deliberately side-effect free.

python
from datetime import datetime, timedelta, timezone

def _asset(**kw) -> AssetMetadata:
    base = dict(asset_id="A", embargo_start=datetime(2020, 1, 1, tzinfo=timezone.utc),
                correlation_id="c-A")
    return AssetMetadata(**{**base, **kw})

def test_expires_at_the_boundary_not_after():
    boundary = datetime(2030, 1, 1, tzinfo=timezone.utc)
    a = _asset(embargo_end=boundary)
    assert evaluate_embargo_state(a, boundary) == EmbargoState.EXPIRED       # exactly at
    assert evaluate_embargo_state(a, boundary - timedelta(seconds=1)) == EmbargoState.ACTIVE

def test_reaching_boundary_never_publishes_directly():
    a = _asset(embargo_end=datetime(2020, 1, 1, tzinfo=timezone.utc))
    # A long-past boundary is EXPIRED, never PUBLISHED — publication is a later gate.
    assert evaluate_embargo_state(a, datetime.now(timezone.utc)) == EmbargoState.EXPIRED

def test_indefinite_embargo_stays_active():
    a = _asset(embargo_end=None)
    assert evaluate_embargo_state(a, datetime.now(timezone.utc)) == EmbargoState.ACTIVE

def test_naive_boundary_is_normalized_to_utc():
    # A naive string is coerced by the validator, so comparison never mixes tz-aware and naive.
    a = _asset(embargo_end="2030-01-01T00:00:00")
    assert a.embargo_end.tzinfo is not None

All four cases should pass. For a live dry run, point main() at a read replica and stub _emit to tally intended transitions without dispatching them — a pre-flight that surfaces a drifted export or a clock-skew problem before a single publish event leaves the pipeline.

In This Section

One focused guide goes deeper on the boundary arithmetic this engine depends on:

  • Setting Date-Based Embargo Triggers — resolving the daylight-saving drift, memory-exhaustion, and silent-fallback failures that lock assets permanently or publish them one timestamp too early, so the boundary the evaluator reads is always correct.

FAQ

Why does embargo expiration route to a review state instead of publishing directly?

Because a lapsed embargo and an active copyright term are independent. A donor hold can expire while the work is still protected, so publishing the instant embargo_end passes risks an unauthorized disclosure. The evaluator only ever moves an asset to expired, and a mandatory rights re-check decides between publication, re-embargo, and human review. Collapsing the two steps is the failure this whole design exists to prevent.

Why capture one now for the entire batch rather than reading the clock per record?

A large batch takes seconds to minutes to process. If each record calls datetime.now() independently, two otherwise-identical assets evaluated on either side of a midnight boundary get different verdicts, which makes the run non-reproducible and breaks idempotency. Capturing a single UTC instant at the start of the run means the whole batch is evaluated against one consistent clock, so a retry produces exactly the same transitions.

How is a duplicate publish event prevented when a cron run overlaps the previous one?

Every emitted event carries the record’s correlation_id as an Idempotency-Key, and the event sink treats a repeated key as a no-op (returning 409, which the dispatcher accepts as success). Because evaluation is a pure function of the record and the captured clock, two overlapping runs compute the same transition and the same key, so the second delivery changes nothing downstream.

Should an embargoed asset’s IIIF manifest be hidden until release?

No. Hiding the manifest conflates description with access. The manifest’s rights property should carry an in-copyright statement URI while the asset is embargoed, and the IIIF Authorization Flow API — referenced through the manifest’s service property — withholds the actual image. This lets the asset be publicly discoverable and described while remaining access-restricted, and it means release is a single URI-and-auth update rather than a manifest that suddenly appears from nowhere.

What happens to an embargo with no end date?

It is treated as an indefinite hold and never auto-lifts. evaluate_embargo_state returns active for any record whose embargo_end is None, so an indefinite bequest condition can only be released when a curator supplies an explicit end date. Treating a missing boundary as “publish now” would be the mirror image of the disclosure risk the whole engine is built to avoid.