OAIS Reference Model Implementation: Compiling ISO 14721 Functional Entities into Python Services

The Open Archival Information System reference model (ISO 14721) remains the foundational specification for institutional digital preservation, yet theoretical compliance rarely survives contact with production environments without rigorous engineering. This is the subsystem within the parent OAIS-Compliant Digital Preservation Architecture that owns package promotion: it defines what a valid Submission Information Package (SIP) is, what deterministic transformations turn it into an Archival Information Package (AIP), and how a Dissemination Information Package (DIP) is projected without ever mutating the sealed original. Implementing the model means treating each functional entity as a typed service contract rather than a diagram on a wall — a boundary that validates content, emits exactly one preservation event, and refuses to advance a package whose invariants do not hold. The promotion logic here feeds provenance records into PREMIS metadata mapping, depends on signature-based preservation format identification to characterise every bitstream, and hands sealed AIPs to the long-term storage architecture that guarantees their survival. For archivists, digital preservation specialists, and Python automation engineers, the payoff is a machine-verifiable pipeline that an ISO 16363 auditor can trace end to end.

# Functional Entities as Typed Service Contracts

The six OAIS functional entities must be translated into discrete, interoperable services with explicit input and output types. Ingest, Archival Storage, Data Management, Administration, Preservation Planning, and Access cannot function as a monolithic application; each is a function with a typed signature and a single documented side effect on the preservation record. Modelling them this way is what makes the system testable — you can assert that a malformed SIP never yields an AIP, that a DIP is byte-reducible to its source AIP, and that every state transition emitted exactly one auditable event.

Each data-path entity is a service with a typed input and output contract; Data Management, Administration, and Preservation Planning form a management plane governing all three.

The engineering discipline is that no entity may be implicit. Archival Storage is not a filesystem path; it is an interface with put, verify, and retrieve contracts. Access is a projection that derives a DIP on demand and holds a read lock on the AIP for the duration of the copy. When each boundary is expressed as a type, schema validation against METS, PREMIS, and EAD profiles happens at exactly one place — the moment a package crosses from one entity to the next — and never has to be re-checked defensively downstream.

# Package Data Structures and Promotion Invariants

Before any code runs, the SIP, AIP, and DIP must exist as versioned data structures. Modelling them with pydantic gives you validation-on-construction: an object that cannot be built from malformed input can never enter the pipeline in the first place. The following models capture the minimum invariants each package must hold.

from __future__ import annotations

from datetime import datetime, timezone
from enum import Enum
from pathlib import Path

from pydantic import BaseModel, Field, field_validator


class PackageState(str, Enum):
    """Lifecycle states a package may occupy during promotion."""

    RECEIVED = "received"
    VALIDATED = "validated"
    QUARANTINED = "quarantined"
    SEALED = "sealed"          # an AIP whose fixity is committed
    DISSEMINATED = "disseminated"


class FileEntry(BaseModel):
    path: str = Field(..., description="Path relative to the package root")
    size_bytes: int = Field(..., ge=0)
    sha256: str = Field(..., pattern=r"^[0-9a-f]{64}$")


class SIPManifest(BaseModel):
    """A Submission Information Package as a strictly validated boundary object."""

    sip_id: str
    producer: str
    received_at: datetime
    files: list[FileEntry]

    @field_validator("files")
    @classmethod
    def _reject_empty(cls, files: list[FileEntry]) -> list[FileEntry]:
        if not files:
            raise ValueError("A SIP must contain at least one content file")
        return files


class AIP(BaseModel):
    """An Archival Information Package: the object preserved for its full retention life."""

    aip_id: str
    derived_from: str                 # originating sip_id
    state: PackageState = PackageState.SEALED
    created_at: datetime
    files: list[FileEntry]
    preservation_events: list[dict] = Field(default_factory=list)

The field-level constraints are load-bearing. The sha256 pattern rejects truncated or upper-case digests before they can poison a manifest; the non-empty files validator prevents an empty submission from being silently promoted; and PackageState makes the legal transitions explicit — a package moves RECEIVED → VALIDATED → SEALED, or diverts to QUARANTINED, but can never jump straight from RECEIVED to DISSEMINATED. These invariants are the same properties an ISO 16363 control mapping later has to demonstrate, so encoding them in the type system means the audit evidence is generated for free.

# Information Package State Machine

The following diagram traces the deterministic control flow of a single package: a producer’s SIP is validated for schema conformance and fixity, promoted to a sealed AIP on success or diverted to quarantine on failure, stored, and only then made projectable as a DIP through the Access entity.

OAIS package promotion: SIP validation gates entry to Archival Storage, and a DIP is projected read-only from the sealed AIP.

Every edge in this graph is a function call that either succeeds and emits a preservation event or fails and emits a quarantine event — there is no third outcome and no silent pass-through. The invalid branch is not an error path bolted on afterward; it is a first-class output of the Ingest contract, which is why a malformed SIP produces a durable, auditable record rather than an exception that disappears into a log.

# Core Implementation: Deterministic AIP Assembly

Production-grade ingest must be idempotent and fully instrumented. Every file entering the system requires immediate cryptographic hashing, and every hash must generate a PREMIS-aligned event with a real timestamp and agent identifier. The implementation below assembles an AIP manifest from a SIP directory with structured logging and explicit error handling, so a failed run leaves a diagnosable trail rather than a partial artefact.

import hashlib
import json
import logging
from datetime import datetime, timezone
from pathlib import Path

logger = logging.getLogger("oais.ingest")

_CHUNK = 8192


def _sha256_of(path: Path) -> str:
    """Stream a file through SHA-256 without loading it entirely into memory."""
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for chunk in iter(lambda: handle.read(_CHUNK), b""):
            digest.update(chunk)
    return digest.hexdigest()


def _premis_event(event_type: str, detail: str, outcome: str) -> dict:
    """Build a PREMIS-style event record with a UTC timestamp and agent."""
    return {
        "event_type": event_type,
        "event_detail": detail,
        "event_outcome": outcome,
        "event_date_time": datetime.now(timezone.utc).isoformat(),
        "linking_agent": "oais_ingest_service/v1.4",
    }


def generate_aip_manifest(sip_dir: str, output_dir: str) -> dict:
    """Ingest a SIP directory, compute SHA-256 fixity, and write an AIP manifest.

    Raises FileNotFoundError if the SIP directory is absent and ValueError if the
    SIP contains no content files, so an empty submission can never be sealed.
    """
    sip_path = Path(sip_dir)
    if not sip_path.is_dir():
        logger.error("SIP directory not found: %s", sip_dir)
        raise FileNotFoundError(f"SIP directory not found: {sip_dir}")

    aip_id = f"AIP-{datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%SZ')}"
    manifest: dict = {
        "aip_id": aip_id,
        "created": datetime.now(timezone.utc).isoformat(),
        "state": "sealed",
        "files": [],
        "preservation_events": [],
    }

    for file_path in sorted(sip_path.rglob("*")):
        if not file_path.is_file():
            continue
        try:
            checksum = _sha256_of(file_path)
        except OSError as exc:
            logger.exception("Unreadable file during ingest: %s", file_path)
            manifest["preservation_events"].append(
                _premis_event("fixity_check", str(file_path), outcome="failure")
            )
            raise RuntimeError(f"Fixity generation failed for {file_path}") from exc

        relative = file_path.relative_to(sip_path)
        manifest["files"].append(
            {
                "path": str(relative),
                "size_bytes": file_path.stat().st_size,
                "sha256": checksum,
            }
        )
        manifest["preservation_events"].append(
            _premis_event("fixity_check", f"SHA-256 for {relative}", outcome="success")
        )
        logger.info("Sealed %s (%s)", relative, checksum[:12])

    if not manifest["files"]:
        logger.error("SIP %s contained no content files; refusing to seal", sip_dir)
        raise ValueError("Cannot seal an AIP from an empty SIP")

    out = Path(output_dir)
    out.mkdir(parents=True, exist_ok=True)
    manifest_path = out / "aip_manifest.json"
    manifest_path.write_text(json.dumps(manifest, indent=2), encoding="utf-8")
    logger.info("Wrote AIP manifest %s (%d files)", aip_id, len(manifest["files"]))
    return manifest


# Usage: generate_aip_manifest("/path/to/sip", "/path/to/aip/output")

This pattern guarantees that every bitstream crossing the ingest boundary is cryptographically sealed, logged with a linking_agent, and either fully committed or explicitly failed. For a complete, worked pipeline that carries a package from a producer drop through fixity validation to a sealed archival object, follow the guide to setting up OAIS SIP/AIP/DIP workflows in Python. In production, wrap the manifest write with xmlschema or lxml validation of the METS wrapper against your institutional profile before the object is handed to storage.

The probability that two distinct bitstreams collide under SHA-256 is negligible for archival scale. For a collection of (n) objects the birthday-bound approximation is:

$$P_{\text{collision}} \approx \frac{n^2}{2 \cdot 2^{256}}$$

Even at (n = 10^{12}) objects this remains far below any realistic bit-rot rate, which is why SHA-256 fixity — not mere file-size comparison — is the accepted evidence of integrity for the AIP’s full retention life.

# Integration Points Across the Pipeline

The reference model implementation is the hub that other subsystems attach to, and each attachment is a typed handoff rather than a shared mutable state. Ingest does not guess file types: it delegates characterisation to the signature-based preservation format identification service, which resolves a PRONOM PUID and a risk profile for every file before the AIP is sealed. Those PUIDs stay meaningful over time only because the format registry integration subsystem keeps a checksum-verified local mirror of PRONOM in sync, so a format flagged obsolete triggers a Preservation Planning directive rather than a silent gap.

Every event this subsystem emits — ingestion, fixity_check, migration — is serialised through PREMIS metadata mapping, which owns the canonical event vocabulary and the mapping from Dublin Core descriptive fields onto PREMIS semantic units. Sealed AIPs are then persisted by the long-term storage architecture, whose replication and integrity-scrubbing guarantees are what let the Access entity trust that the bytes it projects into a DIP still match the manifest. Upstream, the packages this model consumes originate in the automated ingestion and batch scanning workflows that produce the raw digitisation output — the SIP boundary is precisely where that batch pipeline hands control to the preservation system.

# Validation and Compliance Rules

The reference model is only satisfied when each transition produces the correct PREMIS event and each package satisfies its schema constraints. The event vocabulary below is the minimum an ISO 16363 audit will expect to reconstruct a package’s history.

Three compliance rules are non-negotiable. First, fixity is computed once at ingest and re-verified on every copy and every scheduled scrub — never re-derived from a downstream artefact. Second, migration is additive: the original bitstream is retained and linked to its migrated representation through a PREMIS relationship, so provenance is never overwritten. Third, dissemination is read-only by construction; a DIP is a projection, and the Access contract holds a lock that guarantees the AIP it derived from is byte-identical to its manifest at the moment of copy. Aligning these against the CCSDS OAIS reference model and Python’s standard-library hashlib keeps the implementation traceable to the published specification.

# Troubleshooting Reference

# Frequently Asked Questions

# Does implementing OAIS require a specific database or storage backend?

No. The reference model specifies functional entities and their information flows, not technology. Archival Storage is an interface with put, verify, and retrieve contracts, so it can be backed by object storage, a tape library, or a distributed filesystem — provided the backend supports post-write verification and the replication guarantees defined in the long-term storage architecture.

# How is a SIP different from an AIP in practice?

A SIP is an untrusted boundary object supplied by a producer; it is validated on construction but not yet fixity-committed. An AIP is the sealed, checksum-committed object preserved for its full retention life, carrying the complete PREMIS event history that a SIP does not yet have. Promotion from one to the other is the single most tightly gated transition in the pipeline.

# Where does format identification fit in the promotion flow?

Format characterisation runs during Ingest, before the AIP is sealed, so every file carries a resolved PRONOM PUID and risk profile in its manifest. That lets Preservation Planning detect obsolescence later and issue migration directives without ever re-opening the original bitstream.

# What makes an implementation ISO 16363-certifiable rather than merely OAIS-shaped?

Auditability. ISO 16363 probes whether every preservation action produced a durable, machine-readable record: a malformed SIP that never yielded an AIP, a DIP that is byte-reducible to its source, and exactly one preservation event per transition. Encoding those invariants in typed contracts is what turns a working pipeline into certifiable evidence.

# Related

• OAIS-Compliant Digital Preservation Architecture — the parent architecture this subsystem belongs to, mapping all six preservation subsystems end to end.
• Setting up OAIS SIP/AIP/DIP workflows in Python — a runnable, worked pipeline that promotes a package from producer drop to sealed AIP.
• PREMIS Metadata Mapping — the canonical event vocabulary and Dublin Core-to-PREMIS mapping every transition here serialises through.
• Preservation Format Identification — signature-based PUID resolution that characterises each bitstream before an AIP is sealed.
• Long-Term Storage Architecture — replication, erasure coding, and integrity scrubbing that keep sealed AIPs verifiable over decades.
• Digital Preservation Security Policies — the policy-as-code layer that enforces admission control and ISO 16363 control mapping over these transitions.