Walking California BizFile’s Paginated Entity Search Without Dropping or Double-Counting Rows #

This guide sits inside the Pagination Handling for Bulk Records area of the Secretary of State Portal & API Ingestion framework, and it narrows the general traversal problem to one portal that behaves badly enough to deserve its own treatment: California’s BizFile Online search. BizFile does not hand you an opaque cursor. It gives you a page number over a server-rendered, freshly-sorted result set that keeps mutating while you read it — which means a naive for page in range(...) will, on a long California sweep, quietly skip some entities and return others twice.

Scope of This Page #

This page covers exactly one thing: safely walking BizFile’s paginated search results for a single bulk query, end to end. That means imposing a stable sort so the page boundaries mean something, detecting the genuine last page instead of trusting a page count, deduplicating entities across page boundaries when the result window shifts under you, and — because a California sweep can run for many minutes and get interrupted — resuming a partial walk from a durable checkpoint rather than restarting it. It deliberately excludes the surrounding machinery: the per-jurisdiction error categorization & retry logic that decides whether a failed page is retryable, the async polling & rate limiting token bucket that keeps BizFile from throttling you in the first place, and the completeness gate that the parent area applies to the finished result set. Here we assume the request is paced and classified, and focus only on the walk itself. For the general model that abstracts BizFile alongside cursor-based portals, see Reconciling Cursor and Offset Pagination Across State Portals.

The Operational Constraint That Forces a Resumable Walk #

BizFile exposes California entity records under the public-inspection regime that also governs the Statement of Information filing itself — the biennial disclosure required by Cal. Corp. Code § 1502, whose omission draws a $250 penalty under § 2204 and, ultimately, suspension of corporate powers under § 2205. The compliance stake is therefore the same asymmetric one the parent area describes: an entity you fail to page in is an entity that never receives a deadline, and its § 1502 filing lapses on your watch. BizFile compounds the risk with two portal-specific behaviours. First, it re-sorts and re-paginates the result set on every request, so a filing accepted by the Secretary of State between your page 8 and page 9 shifts every row after it — silently duplicating one entity and skipping another. Second, it caps how deep you can page regardless of how many rows actually matched, so an unbounded query can appear “complete” while thousands of entities sit past the cap. A correct walk has to pin the ordering, dedupe across the shift, and stop for a reason it can name.

Prerequisites #

• Python 3.10+ — for X | None unions, match where useful, and async/await iteration.
• httpx 0.27+ — async client with connection pooling; BizFile search is a JSON POST, so a typed request/response model matters.
• A durable checkpoint store — Redis, or a single row in your compliance ledger — reachable behind the small CheckpointStore protocol below. It must survive a worker restart.
• A stable per-entity key to dedupe on. BizFile returns an entity_number (the California entity ID); that, not row position, is the identity.
• An already-paced, already-classified request: the async polling & rate limiting layer has granted a token, and transient faults are handled upstream.

Implementation: A Resumable, Deduplicating BizFile Walk #

The module below walks BizFile one page at a time. It requests an explicit sort so page boundaries are reproducible, deduplicates on entity_number so a shifted window never emits the same entity twice, checkpoints its position after emitting each page so an interrupt costs at most one re-fetched page, and stops on a named condition — a short page, a satisfied total, or the hard page cap — never on an ambiguous fall-through. Comments mark the correctness-critical lines. The diagram traces one page through the walk and shows how an interrupt resumes from the last checkpoint.

from __future__ import annotations

import asyncio
import hashlib
import json
import logging
from dataclasses import dataclass, field
from typing import Any, AsyncIterator, Protocol

import httpx

# Structured JSON logging — every line is a parseable audit/observability event.
logging.basicConfig(level=logging.INFO, format="%(message)s")
logger = logging.getLogger("ingestion.bizfile")


class CheckpointStore(Protocol):
    """Minimal durable-state contract; back it with Redis or a compliance-ledger row."""

    async def load(self, key: str) -> dict[str, Any] | None: ...
    async def save(self, key: str, state: dict[str, Any]) -> None: ...


@dataclass(frozen=True)
class WalkPlan:
    query: str                               # the BizFile search term (name fragment or filter)
    page_size: int = 50                      # BizFile caps rows-per-page; 50 is the safe ceiling
    sort_field: str = "INITIAL_FILING_DATE"  # explicit sort → reproducible page boundaries
    hard_page_cap: int = 400                 # BizFile stops paging deep; refuse to walk blind past this


@dataclass
class WalkState:
    plan: WalkPlan
    page_index: int = 0                      # next page to fetch (0-based) — the resume anchor
    high_water: str | None = None            # last sort key emitted; evidence of a shifted window
    seen: set[str] = field(default_factory=set)   # entity ids already emitted — dedupe across pages
    total_expected: int | None = None

    def to_json(self) -> dict[str, Any]:
        # Persist the actual ids (resume needs them to dedupe) plus a digest for tamper-evidence.
        return {
            "page_index": self.page_index,
            "high_water": self.high_water,
            "seen_ids": sorted(self.seen),
            "seen_digest": hashlib.sha256("".join(sorted(self.seen)).encode("utf-8")).hexdigest(),
            "total_expected": self.total_expected,
        }


class BizFilePaginator:
    """Resumable, deduplicating walk of a California BizFile paginated search."""

    SEARCH_PATH = "/api/Records/businesssearch"

    def __init__(self, client: httpx.AsyncClient, store: CheckpointStore) -> None:
        self._client = client
        self._store = store

    async def walk(self, plan: WalkPlan, resume_key: str) -> AsyncIterator[dict[str, Any]]:
        """Yield every unique entity for `plan`, resuming from any prior checkpoint under `resume_key`."""
        state = await self._restore(plan, resume_key)
        while state.page_index < plan.hard_page_cap:
            payload = await self._fetch(state)
            rows = payload.get("rows", [])
            state.total_expected = payload.get("total", state.total_expected)

            if not rows:
                logger.info(json.dumps({"event": "walk_empty_page", "page": state.page_index}))
                break

            emitted = 0
            for row in rows:
                entity_id = row["entity_number"]        # BizFile's stable per-entity key — not row position
                if entity_id in state.seen:
                    # A filing accepted mid-sweep shifted the window and re-served a row we already have.
                    continue
                state.seen.add(entity_id)
                state.high_water = row.get(plan.sort_field, state.high_water)
                emitted += 1
                yield row

            state.page_index += 1
            # Checkpoint AFTER emitting: a crash re-fetches at most one page and skips none.
            await self._checkpoint(resume_key, state, len(rows), emitted)

            # Last-page detection is explicit, never an ambiguous fall-through.
            if len(rows) < plan.page_size:
                logger.info(json.dumps({"event": "walk_short_page", "page": state.page_index}))
                break
            if state.total_expected is not None and len(state.seen) >= state.total_expected:
                logger.info(json.dumps({"event": "walk_count_satisfied",
                                        "seen": len(state.seen), "total": state.total_expected}))
                break
        else:
            # Reached the hard cap without a natural end — alertable, never silent.
            logger.warning(json.dumps({"event": "walk_page_cap_reached",
                                       "cap": plan.hard_page_cap, "seen": len(state.seen)}))

    async def _fetch(self, state: WalkState) -> dict[str, Any]:
        body = {
            "SEARCH_VALUE": state.plan.query,
            "SEARCH_FILTER_TYPE_ID": "0",
            "SORT_FIELD": state.plan.sort_field,        # explicit, stable ordering across requests
            "PAGE": state.page_index + 1,               # BizFile page numbers are 1-based
            "PAGE_SIZE": state.plan.page_size,
        }
        response = await self._client.post(self.SEARCH_PATH, json=body)
        response.raise_for_status()                     # 5xx/429 surface to the retry layer, not here
        payload = response.json()
        logger.info(json.dumps({"event": "page_fetched", "page": state.page_index + 1,
                                "rows": len(payload.get("rows", [])), "total": payload.get("total")}))
        return payload

    async def _restore(self, plan: WalkPlan, resume_key: str) -> WalkState:
        saved = await self._store.load(resume_key)
        if saved is None:
            return WalkState(plan=plan)
        state = WalkState(
            plan=plan,
            page_index=saved["page_index"],
            high_water=saved.get("high_water"),
            seen=set(saved.get("seen_ids", [])),        # rehydrate dedupe set so resume never re-emits
            total_expected=saved.get("total_expected"),
        )
        logger.info(json.dumps({"event": "walk_resumed", "from_page": state.page_index,
                                "already_seen": len(state.seen)}))
        return state

    async def _checkpoint(self, resume_key: str, state: WalkState, fetched: int, emitted: int) -> None:
        await self._store.save(resume_key, state.to_json())
        logger.info(json.dumps({"event": "checkpoint_written", "page": state.page_index,
                                "fetched": fetched, "emitted": emitted, "seen_total": len(state.seen)}))

Two design choices carry the correctness. The dedupe set is keyed on entity_number, the identity BizFile assigns each entity, so a row that reappears because the result window shifted is dropped rather than double-emitted — the same defence the offset grids in Delaware and Texas need. And the checkpoint is written after the page is yielded, which makes the walk resumable with at-least-once semantics: an interrupt costs one re-fetched page whose rows the rehydrated seen set will suppress, so downstream never sees a duplicate and never loses a page. If you flip that and checkpoint before emitting, a crash between the two would skip a whole page — the exact silent gap this design exists to prevent.

Configuration Reference #

These values are portal behaviour and compliance policy, not knobs to branch on inside the loop.

Failure Modes and Fallback Routing #

Each fault maps onto the four-tier scheme in the parent area’s error categorization & retry logic — transient, statutory, data-validation, and system.

1. The result window shifts mid-walk (system, recoverable in-walk). A Statement of Information accepted between two of your requests re-sorts BizFile’s output, so a later page re-serves an entity you already emitted. The seen set drops the duplicate; a rising duplicate ratio across pages is itself a signal that the registry is churning and the sort key should be pinned to something more stable than a filing date.
2. The hard page cap is reached before a natural end (system, needs narrowing). A query broad enough to match past BizFile’s depth cap terminates on walk_page_cap_reached with rows still unseen. The walk refuses to report this as complete; the fix is to split the query — by entity type or filing-date band — and walk each shard, never to raise the cap and pretend the tail does not exist.
3. A worker dies mid-sweep (transient/system). Because the checkpoint holds the page index and the full seen set, the replacement worker calls walk with the same resume_key and continues from the next page. At most one page is re-fetched, and its rows are suppressed by the rehydrated dedupe set, so the resumed sweep is indistinguishable from an uninterrupted one.
4. BizFile returns a malformed row missing entity_number (data-validation). The row["entity_number"] access raises immediately rather than letting an unidentifiable record slip through dedupe. Quarantine the offending payload against your compliance metadata schemas and continue; a record with no stable identity cannot be safely deduplicated and must not be counted toward completeness.

Frequently Asked Questions #

Related #

• Pagination Handling for Bulk Records
• Reconciling Cursor and Offset Pagination Across State Portals
• Error Categorization & Retry Logic
• Async Polling & Rate Limiting
• Secretary of State Portal & API Ingestion