Fetching the gazette: a clean-room INlabs ingestor

In update #1 I chose where RegWatch’s data comes from: INlabs, the government’s own service that hands you each day’s Diário Oficial da União as a ZIP of per-article XML — and proved the whole path by hand: login, list a date, download, parse. The next slice would be the real ingestor, turning that validated path into a tested fetch_editions(date). I just did it.

One function to rule them all

The ingestor’s entire public surface is a single function:

def fetch_editions(date: datetime.date, *, client=None) -> list[RawEdition]:

Back in #1 I built and tested the matcher and enricher against a frozen contract before any network code existed:

@dataclass(frozen=True)
class RawItem:
    identifier: str
    title: str
    agency: str
    raw_text: str
    source_anchor: str

@dataclass(frozen=True)
class RawEdition:
    date: datetime.date
    section: str
    source_url: str
    items: tuple[RawItem, ...]

fetch_editions goes under that pipeline and produces exactly this shape. Everything above it (matching, enrichment, digests) was already done, with the network deferred behind the contract. The ingestor’s job was just to fill the boundary in — and doing it this way is what let me ship the core first.

Three were given to the gazette/inlabs module

I split the work along the obvious fault lines and kept each piece ignorant of the others:

• parser.py — pure XML → RawItem/RawEdition. No HTTP, no idea where bytes come from.
• client.py — httpx: login, per-section download, cookies, retries. No idea what a gazette is.
• fetch.py — the orchestration that wires them together, plus a content_hash for later.

fetch.py is small enough to read whole. It owns the section list, pairs each downloaded ZIP with its source URL, and is careful about who closes the client:

SECTIONS = ("DO1", "DO2", "DO3", "DO1E", "DO2E", "DO3E")

def fetch_editions(date, *, client=None) -> list[RawEdition]:
    owns_client = client is None
    client = client or InlabsClient.from_env()
    try:
        client.login()
        editions: list[RawEdition] = []
        for section in SECTIONS:
            zip_bytes = client.download_section(date, section)
            if zip_bytes is None:
                continue                       # section not published — skip
            editions.append(
                parse_section_zip(
                    zip_bytes, source_url=client.section_url(date, section)
                )
            )
        return editions
    finally:
        if owns_client:
            client.close()

Deep in the data, he poured his cruelty, his malice and his will to dominate

Here I want to show how important it is to inspect the data in detail before writing any scraper: by measuring field presence across three editions and 1,297 real articles, I found that several “obvious” fields are traps: Titulo is dead (0% populated), Ementa is unreliable (0% in the Seção 2 personnel section), and idMateria looks like a key but isn’t unique.

The title comes from Identifica, but ~7% of Seção 1 articles and over half of some extra editions ship without one — so there’s a fallback to "{artType} {id}":

art_type = article.get("artType", "").strip()

identifica_el = article.find("body/Identifica")
identifica = (identifica_el.text or "").strip() if identifica_el is not None else ""
title = identifica or f"{art_type} {article_id}".strip()

The body lives in Texto as escaped HTML inside a CDATA block. lxml’s itertext() flattens it to plain text; I wrap it in a synthetic root so multiple <p> siblings have one parent, and collapse whitespace:

def _plain_text(texto_html: str) -> str:
    # Texto holds escaped HTML; wrap so multiple <p> siblings have one root,
    # join pieces with spaces so adjacent paragraphs don't glue together.
    fragment = html.fragment_fromstring(texto_html, create_parent="div")
    pieces = [t for t in fragment.itertext() if t.strip()]
    return _collapse(" ".join(pieces))

The rest is a near-mechanical mapping: identifier ← @id (the stable key, not idMateria), agency ← @artCategory (already a clean Ministério/.../Órgão string, no parsing needed), source_anchor ← @pdfPage, and the edition’s date/section from @pubDate/@pubName. The pdfPage URLs arrive with & entities; I get the un-escaping for free because they’re parsed as XML attributes, not handled as strings.

The dead and unreliable fields I simply never read.

Testing became legend, legend became design

The discipline from #1 holds now and should hold till the end of this project: no production code without a failing test first — a.k.a. Test-Driven Development (TDD).

The catch is that the thing I’m building is a network client. So how do you test it over and over without hitting a live government endpoint on every unit test? In other words: how do you TDD a scraper?

Two techniques solve it:

• The parser is pure, so it’s tested directly against the real XML fixtures the spike captured — including the awkward retificação with an empty Identifica and self-closed Ementa.

• The HTTP client is tested with httpx’s MockTransport: a fake transport you hand a request→response function. The client’s real login, cookie, 404, and re-login logic runs; only the wire is faked. No recorded cassettes, no network, no flakiness. Here’s the test that the client re-authenticates exactly once when the session cookie expires mid-download:

def handler(request: httpx.Request) -> httpx.Response:
    if request.url.path == "/logar.php":
        calls["login"] += 1
        return httpx.Response(200, headers={"set-cookie": LOGIN_COOKIE})
    calls["download"] += 1
    if calls["download"] == 1:
        return httpx.Response(403, text="session expired")
    return httpx.Response(200, content=b"PK\x03\x04ok")

That’s my answer to “how to TDD a scraper”: make the parser pure and fake the transport, not the server. Nineteen tests cover the three modules this way; the whole suite is 38 tests, green on Postgres. There’s also a small runnable notebook that drives login→download→parse end to end against the live service when I want to eyeball real output — but nothing in the test path touches the network.

The data came to me, my own, my love, my own, my precious!

The spike’s operational notes are now code. download_section is where the gazette’s quirks live:

def download_section(self, date, section) -> bytes | None:
    resp = self._get_zip(date, section)
    if resp.status_code in (401, 403):
        self.login()                       # one capped, non-recursive re-login
        resp = self._get_zip(date, section)
    if resp.status_code == 404:
        return None                        # not published → skip and continue
    resp.raise_for_status()
    body = resp.content
    return body if body.startswith(b"PK") else None

A 404 or a body that doesn’t start with the PK ZIP magic both mean “this section wasn’t published today” (return None), and the orchestrator skips it. Each day has up to six sections: DO1–DO3 plus the extras DO1E–DO3E. And fetch.py exposes a content_hash:

def content_hash(data: bytes) -> str:
    return hashlib.sha256(data).hexdigest()

It’s unused for now, but it’s there on purpose. INlabs stamps a fresh Last-Modified on every ZIP roughly every 30 minutes whether the content changed or not, so timestamps are useless for “did this actually change.” The scheduler will hash bytes instead.

But something happened that the developer did intend

For now, fetch_editions is all-or-nothing. One malformed article, or one non-404 HTTP error on any section, aborts the entire day’s run for every section. I’ve already written about how much resilience matters for web scrapers in Scraping a fragile legacy site into a clean time series.

And there’s no per-section try/except in that loop — I left it that way on purpose, though.

The reasoning: the fields I actually depend on are present in 100% of the 1,297 articles I measured, so a parse failure is unlikely; the spike notes XML completeness isn’t guaranteed, but it’s been reliable. And the pipeline is idempotent by construction (those unique constraints from #1), so a failed run is safe to simply re-run once fixed — fail-fast trades a one-article gap for a one-day gap, and the day is recoverable. Per-section isolation and skip-and-count for bad members are real improvements, but they belong with the scheduler, where observability and retry policy live. So I wrote it down as a tracked deferral rather than quietly shipping a half-measure.

Not all those who overflow are lost

After all 38 tests returned green I did the one thing the tests can’t: wrote a sample code and pointed it to a real edition and ran it all the way into the database. Login, download, parse. Result: 400 articles from a single day’s Seção 1, exactly as expected. Then ingest_edition reached Postgres and Postgres said no:

DataError: value too long for type character varying(300)

The agency field was to blame. Back in #1 I sized the Act columns from the spike’s tidy fixtures, and CharField(max_length=300) looked generous for a ministry name. But real DOU artCategory values are whole hierarchies, and they run long:

Ministério da Gestão e da Inovação em Serviços Públicos/Secretaria de Gestão de Pessoas/Diretoria de Serviços de Aposentados e de Pensionistas e Órgãos Extintos/Coordenação-Geral de Benefícios

The fix is boring, which is the point: these fields are free text straight from the source, so they shouldn’t carry a length cap at all. CharField → TextField (in Postgres both compile to the same text type, just without the ceiling):

# before — sized from the spike's tidy fixtures
title  = models.CharField(max_length=500)
agency = models.CharField(max_length=300, blank=True, default="")
# after — real ministries don't fit in 300 characters
title  = models.TextField()
agency = models.TextField(blank=True, default="")

One migration, the suite still 38 green, and the run that had just crashed now ingests all 400 acts and answers a watch for “ministério” with 109 matches — real titles, real agencies, real in.gov.br links. The live run was the test I hadn’t written: a fixture is clean by definition, and clean is exactly what real gazette data is not. That’s the whole argument for build-in-public in one stack trace.

What I learned

A frozen-dataclass contract is what makes “test the core before the network exists” go from a slogan to something you actually do — the boundary is a real object you can build both sides against independently. MockTransport makes an HTTP client genuinely unit-testable without recording a single fixture from the live server; I’d reach for it on any client from now on. And mostly: choosing where to put the seam — which is the same as choosing what to defer — shaped this slice more than any line of the code did.

What’s next

The scheduler: a management command wiring fetch_editions → run_pipeline on a daily cadence, headed for Cloud Run + Cloud Scheduler. That’s also where the all-or-nothing resilience gets fixed for real and where content_hash finally gets persisted to skip unchanged ZIPs.

RegWatch is now open source: the ingestor above, clean-room and TDD’d, lives at github.com/bessavagner/regwatch. Follow the repo if you want to watch the scheduler land in the next update.