Parsing MARC Records with pymarc

Operating within the broader Catalog Ingestion & ILS Sync Pipelines architecture, parsing MARC21 with pymarc is the first deterministic transformation every downstream stage depends on: nothing normalizes, validates, or synchronizes correctly until the raw ISO 2709 byte stream has been decoded into structured records. Library-tech staff hit this problem the moment a vendor SFTP drop, an OAI-PMH harvest, or a full ILS export lands in the staging area — the file is a wall of length-prefixed directory entries and delimiter-separated fields that no schema validator can read directly. This guide walks through instantiating the reader safely, interpreting leader bytes, extracting fields and subfields, handling encoding, and wiring the parse stage into the quarantine, compliance, and synchronization stages around it, with production-grade Python throughout.

Specification & Contract

A MARC21 record is an ISO 2709 exchange structure with three physical regions: a fixed 24-byte leader, a variable-length directory of 12-byte entries, and the variable fields themselves, terminated by a record terminator (0x1D). pymarc hides the directory arithmetic, but it does not interpret the leader for you — that is the pipeline’s responsibility. The leader positions below drive routing and extraction logic and must be read on every record before any field is trusted.

The variable fields split into two classes. Control fields (tags 001–009) carry a single data string with no indicators or subfields — 001 is the record control number, 003 its owning organization, 005 the transaction timestamp. Data fields (tags 010–999) carry two single-character indicators plus one or more subfields, each introduced by a delimiter (0x1F) and a subfield code. The table below is the minimum contract the parse stage guarantees to hand downstream; full tag-to-attribute translation lives in MARC21 field mapping for modern pipelines.

Prerequisites & Environment Setup

The examples target Python 3.11+ and pymarc 5.x, whose Subfield named-tuple API differs from the 4.x ['code', 'value'] list form — code written against the old API silently breaks after upgrade, so pin the major version. Credentials for the downstream ILS and message broker are injected through environment variables, never hard-coded, so the same image runs across staging and production.

• Python >= 3.11 available in the worker image (python --version)
• pymarc >= 5.1 pinned in requirements.txt (the 5.x Subfield API is assumed throughout)
• lxml >= 4.9 installed if any source delivers MARCXML rather than .mrc
• MARC_QUARANTINE_DIR set to a writable path for records that fail to parse
• PII_HASH_SALT exported as a secret (32+ random bytes, never committed)
• INGEST_CORRELATION_ID propagated from the batch orchestrator for audit tracing
• Read access confirmed on the source drop directory; write access denied on originals (parse never mutates the source file)

python -m venv .venv && source .venv/bin/activate
pip install "pymarc>=5.1,<6" "lxml>=4.9"
export MARC_QUARANTINE_DIR="/var/lib/ingest/quarantine"
export PII_HASH_SALT="$(openssl rand -hex 32)"

Core Implementation

The parse stage is built from small, typed functions that each do one job: open and stream, read the leader, extract control fields, extract data fields. Composing them keeps every failure attributable to a single step.

Step 1 — Stream records with an explicit reader configuration. Hand the open binary file object directly to pymarc.MARCReader; it is an incremental parser, so records are decoded one at a time rather than materialized as a list. Setting to_unicode=True converts MARC-8 to Python str, and force_utf8=True is appropriate only when leader position 09 promises Unicode — mixing the two on legacy MARC-8 corrupts diacritics, a case handled in handling UTF-8 encoding in legacy MARC records.

import logging
from pymarc import MARCReader, Record
from typing import Iterator

logger = logging.getLogger("marc_ingest.parse")


def stream_records(filepath: str) -> Iterator[Record]:
    """Yield MARC records one at a time from an ISO 2709 file.

    MARCReader is an incremental parser, so we pass the open file object
    directly instead of reading the whole file into memory. A None value
    signals an unrecoverable parse error for the current record; the reader
    can continue past it to the next record terminator.
    """
    with open(filepath, "rb") as fh:
        reader = MARCReader(fh, to_unicode=True, force_utf8=False, utf8_handling="replace")
        for record in reader:
            if record is None:
                logger.warning(
                    "unrecoverable parse error; record skipped",
                    extra={"reader_error": str(reader.current_exception)},
                )
                continue
            yield record

Step 2 — Read and validate the leader before trusting any field. Leader position 05 decides whether the record is an insert, update, or delete; a record whose status you cannot classify must never reach the synchronization stage. Full leader validation before a database write is covered in validating MARC leader fields before database insert.

from enum import Enum


class SyncAction(str, Enum):
    UPSERT = "upsert"
    DELETE = "delete"
    REENRICH = "reenrich"


def classify_by_leader(record: Record) -> SyncAction:
    """Map leader/05 (record status) onto a downstream sync action."""
    status = record.leader[5]
    if status == "d":
        return SyncAction.DELETE
    if status == "a":  # increase in encoding level -> re-enrich, don't overwrite
        return SyncAction.REENRICH
    if status in ("n", "c", "p"):
        return SyncAction.UPSERT
    raise ValueError(f"unrecognized leader/05 record status: {status!r}")

Step 3 — Extract control fields by tag. Control fields have no subfields; read their .data directly. Guard for absence — a record with no 001 has no stable identity and cannot be synchronized idempotently, so it is quarantined rather than assigned a synthetic key.

def record_identity(record: Record) -> tuple[str, str]:
    """Return (control_number, owning_org) from 001/003, or raise if 001 absent."""
    field_001 = record.get("001")
    if field_001 is None or not field_001.data.strip():
        raise ValueError("record has no 001 control number; cannot assign identity")
    org_003 = record.get("003")
    return field_001.data.strip(), (org_003.data.strip() if org_003 else "UNKNOWN")

Step 4 — Extract data fields and subfields into a normalized shape. Iterate record.get_fields(tag) for repeatable tags and read subfields through the 5.x Subfield API. Yield normalized tuples rather than building intermediate dictionaries, and never assume a subfield exists — field.get_subfields("a") returns a list that may be empty. A common pitfall is calling field["a"], which raises KeyError on absence; prefer get_subfields and handle the empty case explicitly.

from dataclasses import dataclass


@dataclass(frozen=True, slots=True)
class SubjectHeading:
    topic: str
    subdivisions: tuple[str, ...]


def extract_subjects(record: Record) -> list[SubjectHeading]:
    """Extract repeatable 650 topical subjects with their subdivisions."""
    headings: list[SubjectHeading] = []
    for field in record.get_fields("650"):
        topics = field.get_subfields("a")
        if not topics:
            continue  # malformed 650 with no $a; skip rather than fabricate
        subdivisions = tuple(field.get_subfields("x", "v", "z"))
        headings.append(SubjectHeading(topic=topics[0].strip(), subdivisions=subdivisions))
    return headings

Composing the four steps gives the full parse function: stream, classify, identify, extract, and emit a normalized delta for the broker. The vendor-specific field quirks that survive this stage are neutralized later by ILS schema translation patterns, so the rest of the pipeline only ever sees the canonical model.

PII & Compliance Checkpoints

Public sector libraries frequently receive records carrying embedded patron identifiers, circulation notes, or local system metadata that violate state privacy statutes if they reach a discovery layer or analytics warehouse. Run a deterministic sanitization pass immediately after extraction and before the record crosses into the broker — the same patron PII masking discipline the exports pipeline uses. Two rules keep this defensible: masking must be idempotent (reprocessing a record produces byte-identical output) and any retained audit identifier must be irreversibly hashed with an environment-scoped salt, never reversibly encoded.

import hashlib
import os
from pymarc import Field, Subfield

PII_SALT = os.environ["PII_HASH_SALT"].encode()
# Local/system field ranges that must never leave the parse boundary intact.
_LOCAL_TAG_PREFIXES = ("59", "9")
_REDACT_CODES = {"a", "b", "p"}


def hash_identifier(value: str) -> str:
    """Irreversible, salted digest for audit correlation of a masked value."""
    return hashlib.sha256(PII_SALT + value.encode("utf-8")).hexdigest()[:32]


def sanitize_local_fields(record: Record) -> Record:
    """Mask local/system fields in place; idempotent and structure-preserving."""
    for field in record.fields:
        if field.is_control_field():
            continue
        if field.tag.startswith(_LOCAL_TAG_PREFIXES):
            field.subfields = [
                Subfield(code=sf.code, value="[REDACTED]")
                if sf.code in _REDACT_CODES else sf
                for sf in field.subfields
            ]
    return record

Records flagged with a retention marker (for example a local note indicating a hold on circulation history) must carry that flag into the audit log, not be silently dropped; the retention rules that govern how long each record class may persist are defined in the Patron Validation & Privacy Data Routing domain and the broader data privacy boundaries in library systems contract.

Error Handling & Quarantine Patterns

pymarc surfaces parse failures two ways: it yields None from the reader for a record it cannot decode (leaving reader.current_exception populated), and it raises exceptions such as RecordLeaderInvalid, RecordDirectoryInvalid, BaseAddressInvalid, and UnicodeDecodeError during field access. A malformed record must never halt the batch — route it to a quarantine queue with enough context to reproduce the failure, then continue. This mirrors the schema validation quarantine queue that catches records which parse cleanly but violate downstream schema contracts.

import json
from pathlib import Path
from pymarc.exceptions import (
    RecordLeaderInvalid,
    RecordDirectoryInvalid,
    BaseAddressInvalid,
)

QUARANTINE_DIR = Path(os.environ["MARC_QUARANTINE_DIR"])
_RETRYABLE = ()  # parse errors are deterministic; a retry re-fails. Do not retry.


def quarantine(raw_bytes: bytes, reason: str, correlation_id: str) -> None:
    """Persist an unparseable record with its failure reason for later triage."""
    QUARANTINE_DIR.mkdir(parents=True, exist_ok=True)
    digest = hashlib.sha256(raw_bytes).hexdigest()[:16]
    (QUARANTINE_DIR / f"{digest}.mrc").write_bytes(raw_bytes)
    (QUARANTINE_DIR / f"{digest}.json").write_text(json.dumps({
        "reason": reason,
        "correlation_id": correlation_id,
        "byte_length": len(raw_bytes),
    }))
    logger.error("record quarantined", extra={"reason": reason, "digest": digest})


def parse_or_quarantine(record: Record, raw: bytes, correlation_id: str):
    """Attempt full extraction; quarantine on any structural or encoding fault."""
    try:
        action = classify_by_leader(record)
        control_number, org = record_identity(record)
        record = sanitize_local_fields(record)
        return action, control_number, org, record
    except (RecordLeaderInvalid, RecordDirectoryInvalid,
            BaseAddressInvalid, UnicodeDecodeError, ValueError) as exc:
        quarantine(raw, reason=f"{type(exc).__name__}: {exc}", correlation_id=correlation_id)
        return None

Because parse failures are deterministic, retrying the same bytes always re-fails — retry logic belongs on the transport (SFTP fetch, HTTP harvest) and synchronization stages, not here. Once records are parsed and queued, the retry-and-backoff behaviour is owned by async batch processing for catalog updates and, for live ILS reads, ILS REST API polling & rate limiting.

Performance Considerations

The single biggest memory failure is accumulating parsed Record objects in a list — each record retains its full field graph, and a multi-gigabyte vendor dump will exhaust a containerized worker’s heap and trigger an OOM kill. Streaming through the generator in Step 1 keeps resident set size flat regardless of file size. For MARCXML, avoid pymarc.parse_xml_to_array, which materializes the entire DOM; stream <record> elements with lxml.etree.iterparse and clear each element after use. Pre-compile any regex used in subfield inspection at module scope, and cache tag-profile lookup tables at import rather than per record. The full profiling methodology — memory snapshots, del/gc.collect() cadence, and file-descriptor backpressure — is covered in Optimizing pymarc Performance for Large Record Sets.

As a rule of thumb: parse-and-emit throughput is bound by field extraction, not I/O, so the highest-leverage optimization is doing less per record — extract only the tags the downstream contract requires, and defer full-record retention to the stages that actually need it.

Verification & Testing

Validate the parser against known-good and known-bad fixtures rather than live feeds. Build small .mrc fixtures with pymarc.MARCWriter, assert on the normalized output, and confirm that a deliberately corrupted leader routes to quarantine instead of raising past the batch loop. Assert idempotency of the PII pass by masking twice and comparing bytes.

import io
from pymarc import Record, Field, Subfield, MARCWriter


def _fixture_record(control_number: str = "ocm12345") -> bytes:
    record = Record()
    record.leader = "00000nam a2200000 a 4500"
    record.add_field(Field(tag="001", data=control_number))
    record.add_field(Field(
        tag="245", indicators=["1", "0"],
        subfields=[Subfield("a", "Test title :"), Subfield("b", "a subtitle")],
    ))
    record.add_field(Field(
        tag="650", indicators=[" ", "0"],
        subfields=[Subfield("a", "Cataloging"), Subfield("x", "Data processing")],
    ))
    buf = io.BytesIO()
    MARCWriter(buf).write(record)
    return buf.getvalue()


def test_identity_and_subjects():
    raw = _fixture_record()
    record = next(iter(MARCReader(io.BytesIO(raw), to_unicode=True)))
    assert record_identity(record)[0] == "ocm12345"
    subjects = extract_subjects(record)
    assert subjects[0].topic == "Cataloging"
    assert subjects[0].subdivisions == ("Data processing",)


def test_pii_pass_is_idempotent():
    raw = _fixture_record()
    record = next(iter(MARCReader(io.BytesIO(raw), to_unicode=True)))
    once = sanitize_local_fields(record).as_marc()
    twice = sanitize_local_fields(
        next(iter(MARCReader(io.BytesIO(once), to_unicode=True)))
    ).as_marc()
    assert once == twice

Run the suite in CI against the pinned pymarc version so an accidental 4.x/6.x resolution fails the build before it reaches production.

Troubleshooting & FAQ

Why does MARCReader yield None instead of raising an exception?

pymarc returns None when it cannot decode the current record but can still resynchronize on the next record terminator, so the batch is not aborted by one bad record. The underlying error is stored on reader.current_exception. Log that value and route the raw bytes to quarantine; do not treat None as end-of-file, which would silently truncate the batch.

Diacritics arrive as mojibake (e.g. Müller instead of Müller). What is wrong?

The record is MARC-8 (leader/09 is #) but was forced through UTF-8 decoding. Set to_unicode=True and leave force_utf8=False so pymarc performs the MARC-8→Unicode conversion; only set force_utf8=True when leader/09 is a. Mixed batches must branch on leader/09 per record, as detailed in the encoding page linked above.

KeyError is raised when reading a subfield that is usually present. How do I make extraction robust?

field["a"] raises KeyError when the subfield is absent. Use field.get_subfields("a"), which returns a possibly-empty list, and handle the empty case explicitly. Never fabricate a placeholder value for a missing identifying subfield — quarantine the record instead so the gap is visible in triage.

Memory climbs steadily until the worker is OOM-killed mid-batch. Why?

Records are being accumulated in a list (or a comprehension over the whole reader) instead of streamed. Consume the generator from Step 1 and emit each delta immediately; retain nothing across iterations. See the optimization page for the full RSS profiling procedure.

After upgrading pymarc, subfield access breaks with attribute or index errors. What changed?

pymarc 5.x replaced the 4.x ['code', 'value'] subfield lists with a Subfield(code=..., value=...) named tuple. Code that indexed subfields positionally must move to attribute access. Pin pymarc>=5.1,<6 and update construction sites to Subfield("a", "value").

Related

• Optimizing pymarc Performance for Large Record Sets — streaming, memory boundaries, and OOM recovery for multi-gigabyte batches
• Validating MARC Leader Fields Before Database Insert — leader-byte validation gate downstream of parsing
• Schema Validation for Ingested Records — the quarantine queue for records that parse but fail schema contracts
• MARC21 Field Mapping for Modern Pipelines — translating extracted tags into the canonical internal model
• PII Masking in Patron Data Exports — the masking discipline applied at the parse boundary