Operational Context

A Python automation engineer owns the nightly job that decides which digitized objects a collections management system may expose to the public. Each asset carries an embargo_end_date set by a donor agreement, an institutional review period, or a statutory copyright term, and the job must flip access_status from restricted to public the moment the current timestamp crosses that boundary — no sooner, no later, and without a curator touching a keyboard. This page resolves the exact failure that surfaces when that trigger is naive: assets that should have gone live stay locked behind legacy access controls, or — far worse — donor-restricted material surfaces in the public portal one timestamp early. The trigger runs inside the batched, asynchronous evaluator described in the parent Implementing Embargo Workflows stage, and its output feeds the same rights-verification contract the rest of the rights metadata and licensing automation pipeline depends on.

Root Cause Analysis

Date-based embargo triggers fail along three concrete vectors, and each one has a single root cause rather than a family of edge cases.

First, temporal misalignment. A naive datetime.now() comparison reads the server’s local wall clock, so during a daylight-saving transition — or simply on a host configured for a non-UTC zone — the comparison drifts by an hour. An asset whose embargo_end_date is stored in UTC is then compared against a local-time “now,” and the boundary is evaluated at the wrong instant. The fix is to normalize every timestamp to UTC at ingestion and compare only timezone-aware values.

Second, memory exhaustion. Loading an entire rights-metadata table into a list to iterate over it keeps every row alive for the whole run, so a dataset above roughly 500,000 records triggers an out-of-memory kill on a constrained runner. This is the same reference-retention leak worked through in handling large CSV batches without memory leaks; the remedy is lazy, fixed-size chunking so only one batch is ever reachable.

Third, silent coercion on malformed dates. When an export carries an ambiguous or empty ISO-8601 string and the parser falls back to None or the Unix epoch, the asset either locks permanently or publishes immediately. Enforcing the contract with a strict Pydantic v2 model rejects the bad row at the boundary instead of letting a corrupted rights_statement propagate downstream.

Canonical Solution

The solution composes three invariants into one constant-memory pass: validate-and-normalize each record to UTC at ingestion, evaluate the boundary with a >= comparison against a single now captured once per run, and stream records in fixed-size chunks so peak memory is bounded by batch size rather than table size. The annotated implementation below is runnable against any iterable of raw dicts.

python
from __future__ import annotations

from datetime import datetime
from enum import Enum
from itertools import islice
from typing import Iterable, Iterator
from zoneinfo import ZoneInfo

from pydantic import BaseModel, field_validator

UTC = ZoneInfo("UTC")

class AccessStatus(str, Enum):
    RESTRICTED = "restricted"   # embargo not yet started
    EMBARGOED = "embargoed"     # embargo active, awaiting end date
    PUBLIC = "public"           # embargo cleared, eligible for release

class RightsRecord(BaseModel):
    object_id: str
    rights_statement: str
    rights_start_date: datetime | None = None
    embargo_end_date: datetime | None = None
    access_status: AccessStatus = AccessStatus.RESTRICTED

    @field_validator("rights_start_date", "embargo_end_date", mode="before")
    @classmethod
    def normalize_utc(cls, v: str | datetime | None) -> datetime | None:
        # Reject nothing silently: None stays None, but any supplied value must
        # parse to an aware datetime. "Z" is not accepted by fromisoformat on
        # 3.9/3.10, so translate it to a numeric offset before parsing.
        if v is None:
            return None
        dt = v if isinstance(v, datetime) else datetime.fromisoformat(
            str(v).replace("Z", "+00:00")
        )
        # A naive value would compare unsafely against an aware "now"; force one.
        if dt.tzinfo is None:
            raise ValueError(f"timestamp {v!r} is timezone-naive; embargo dates must carry an offset")
        return dt.astimezone(UTC)

def evaluate(record: RightsRecord, now: datetime) -> AccessStatus:
    # `now` is captured ONCE per run and passed in, so every record in a batch
    # is judged against the same instant — no per-row clock skew.
    if record.rights_start_date is not None and now < record.rights_start_date:
        return AccessStatus.RESTRICTED
    if record.embargo_end_date is None:
        return AccessStatus.EMBARGOED            # open-ended embargo never auto-publishes
    # ">=" makes the asset public AT the boundary, not one tick after it.
    return AccessStatus.PUBLIC if now >= record.embargo_end_date else AccessStatus.EMBARGOED

def chunked(rows: Iterable[dict], size: int = 5000) -> Iterator[list[dict]]:
    # islice pulls at most `size` items from the lazy source, so the full table
    # is never materialized — only one batch is reachable at a time.
    it = iter(rows)
    while batch := list(islice(it, size)):
        yield batch

def run_triggers(rows: Iterable[dict], size: int = 5000) -> Iterator[RightsRecord]:
    now = datetime.now(UTC)                      # single source of truth for the run
    for batch in chunked(rows, size):
        for raw in batch:
            record = RightsRecord(**raw)         # raises on malformed / naive dates
            new_status = evaluate(record, now)
            if new_status is not record.access_status:
                yield record.model_copy(update={"access_status": new_status})

Only records whose status actually changes are yielded, so the caller commits a minimal, idempotent set of updates to the DAMS rather than rewriting every row on every pass.

Embargo lifecycle timeline with an inclusive end boundary A single asset moves left to right along a time axis. The RESTRICTED zone runs until rights_start_date, where now is still less than the start. The EMBARGOED zone runs from rights_start_date until embargo_end_date, where start is less than or equal to now and now is less than end. The PUBLIC zone begins at embargo_end_date and continues onward, where now is greater than or equal to end. A now cursor is placed precisely on the embargo_end_date boundary, demonstrating that the greater-than-or-equal comparison flips access_status to public at the boundary instant rather than one clock tick later. One asset's lifecycle — the end boundary is inclusive (now ≥ end publishes) RESTRICTED now < rights_start stays locked EMBARGOED start ≤ now < end awaiting boundary PUBLIC now ≥ end eligible for release time rights_start_date embargo_end_date now now lands ON the boundary → PUBLIC

The branching that decides each status — start-date check, open-ended embargo, and the boundary comparison — is shown below.

Embargo status decision flow Top to bottom: a record's asset metadata is normalized so every timestamp is timezone-aware UTC. The first decision compares now against embargo_end; when now is greater than or equal to embargo_end the record's access_status becomes public. Otherwise a second decision compares now against rights_start; when now is less than rights_start the asset remains restricted, and when it is not the embargo is still active and the asset stays embargoed. The two comparisons use different inequalities — inclusive on the end boundary, strict on the start boundary. Status decision — inclusive on the end, strict on the start Asset metadata Normalize timestamps to UTC now ≥ embargo_end? inclusive access_status = public now < rights_start? strict Remain restricted Active embargo yes no yes no

Edge Cases and Variants

  • CSV vs. XML vs. API input. The run_triggers tail is agnostic to source: feed it a csv.DictReader, an iterparse generator that calls elem.clear() per record, or a paged API cursor as covered in polling museum APIs with Python Requests. Only the row source changes; the batch/validate/evaluate loop is unchanged.
  • Strict vs. lenient validation. The model above is strict — a naive or unparsable date raises. In a backfill, wrap RightsRecord(**raw) in a try/except ValidationError and route the offender to a quarantine table; in an interactive re-sync, surface the error to the operator instead of skipping it.
  • Boundary equality. Use >= for the expiration test so an asset publishes at its embargo_end_date, and strict < for the start test so it stays restricted until rights_start_date is reached. Mixing these inequalities is the most common source of one-tick-off publication.
  • Open-ended embargoes. A None embargo_end_date means “restricted until manual review,” so evaluate returns EMBARGOED and never auto-publishes — the release path stays with a curator.
  • Local-time source data. If an upstream system stores wall-clock times without an offset, attach the originating zone (e.g. ZoneInfo("America/New_York")) during extraction before the record reaches this validator, which only accepts aware values.
  • Inverted ranges. Flag any record where embargo_end_date precedes rights_start_date as a data-entry error before evaluation rather than letting the state machine resolve an impossible window.

Validation

Prove the two invariants that matter — the boundary is inclusive, and a timezone-naive date is rejected rather than silently coerced — with an assert-based test that needs no database:

python
# test_embargo_triggers.py  ->  run with:  pytest -q test_embargo_triggers.py
from datetime import datetime, timedelta
from zoneinfo import ZoneInfo
import pytest
from pydantic import ValidationError

UTC = ZoneInfo("UTC")

def _row(**over):
    base = {"object_id": "OBJ.1", "rights_statement": "InC",
            "embargo_end_date": "2026-01-01T00:00:00+00:00"}
    return {**base, **over}

def test_publishes_exactly_at_boundary():
    rec = RightsRecord(**_row())
    boundary = datetime(2026, 1, 1, tzinfo=UTC)
    assert evaluate(rec, boundary) is AccessStatus.PUBLIC          # inclusive >=
    assert evaluate(rec, boundary - timedelta(seconds=1)) is AccessStatus.EMBARGOED

def test_naive_dates_are_rejected():
    with pytest.raises(ValidationError):
        RightsRecord(**_row(embargo_end_date="2026-01-01T00:00:00"))  # no offset

def test_chunking_is_lazy():
    consumed = []
    def source():
        for i in range(12000):
            consumed.append(i)                                     # records what was pulled
            yield _row(object_id=f"OBJ.{i}")
    first = next(chunked(source(), size=5000))
    assert len(first) == 5000 and len(consumed) == 5000            # only one batch materialized

A green run confirms the asset flips public at the boundary (not after), that a naive timestamp raises instead of publishing at the epoch, and that pulling the first batch never materializes the whole source.

Standards Alignment

The normalized states must serialize predictably into the institution’s metadata exports. In LIDO, the embargo classification belongs in <lido:rightsWork> via <lido:termRightsType>, with the boundary carried as a normalized <lido:rightsDate> value — the UTC datetimes produced here map directly. For delivery, an IIIF Presentation API 3.0 manifest carries the static rights assertion in its rights property using a RightsStatements.org URI such as https://rightsstatements.org/vocab/InC/1.0/ while the asset is embargoed; when the trigger flips an asset to public, the pipeline rewrites that URI and regenerates the manifest. Runtime access control itself is not a manifest property — it is enforced by the separate IIIF Authorization Flow API. Assigning the correct URI at each state is covered in mapping RightsStatements.org to collection fields, and post-embargo statutory clearance runs through automating copyright status checks before any asset reaches the public portal. See the Python zoneinfo documentation for the standard-library timezone handling this solution relies on.

Frequently Asked Questions

Why compare against a single now instead of calling datetime.now() per record?

Capturing now once per run guarantees every record in the batch is judged against the same instant. Calling the clock per row lets a long-running batch straddle a boundary, so two otherwise-identical assets can resolve to different states within one pass — a non-deterministic result that is nearly impossible to reproduce in a bug report.

Should the boundary use >= or >?

Use >= for the expiration test. An embargo that ends at 2026-01-01T00:00:00Z is over at that instant, so the asset becomes eligible for release exactly then. Using strict > delays publication by one clock tick, which is invisible in testing but shows up as assets that “should be live” lagging their stated end date.

What happens to a record with no embargo_end_date?

It is treated as an open-ended embargo and stays EMBARGOED — the trigger never auto-publishes it. Releasing an open-ended embargo is a deliberate curatorial action, not something a date comparison should ever decide.

How do I keep this from exhausting memory on a multi-million-row table?

Feed run_triggers a lazy source (a database cursor, csv.DictReader, or a paged API generator) rather than a materialized list. The chunked helper uses itertools.islice so at most size rows are reachable at once, and only status changes are yielded, keeping both the read and write footprints flat regardless of table size.