Skip to content

Elect tabular pipeline map

This document maps the ingestion/elec tabular pipeline: file-metadata parsing, Pydantic models (voter registration first), DuckDB interface, and legacy entrypoints. It supports the Dagster migration plan (Step 2: “Deepen understanding of the tabular pipeline”).

1. Directory layout

Path Role
ingestion/elec/file_metadata.py Path → metadata (election, county, base_geo, category, subcategory, header_group).
ingestion/elec/duck.py DuckDB connection, table creation, insert-from-CSV, ingestion log.
ingestion/elec/ingest.py Orchestration: parse paths → filter → verify headers → select model → validate → load.
ingestion/elec/pipeline.py Stub only (print("hello pipeline")); does not call ingest.
ingestion/elec/models.py Job parameters (ElecJobParameters) and ingestion log schema (ElecIngestionLog).
ingestion/elec/data_models/reg_models.py Voter registration Pydantic models + get_data_model, validate_elec_csv.
ingestion/elec/data_models/_generic_reg.py Base registration row model and DuckDB/PyArrow schema helpers.
ingestion/elec/data_models/sov_models.py SOV file layout notes (docstrings/samples only; no Pydantic yet).
ingestion/elec/data_models/conv_models.py Conv file layout notes (docstrings only; no Pydantic yet).

2. File-metadata parsing

Module: ingestion/elec/file_metadata.py

  • Input: Paths under ELEC_ROOT with shape {elec_code}/c{county}/…csv (e.g. G22/c001/c001_g22_All_VBM_by_g22_rgprec.csv).

  • Entrypoints:

  • get_elec_paths_list(elec_code) — globs {ELEC_ROOT}/{elec_code}/c*/*.csv, returns paths stripped of ELEC_ROOT.

  • extract_file_metadata(elec_path) — parses one path into a dict.

  • parse_elec(elec) — paths for one election → full metadata DataFrame.

  • parse_elec_paths(elec_file_paths) — list of paths → DataFrame with metadata columns.

  • Output dict keys: key, base_geo, category, subcategory, header_group, modify_time.

  • Conventions:

  • election = first 3 chars of path (e.g. g22).

  • county = chars 5:8 (e.g. 001).

  • base_geo from regex _by_{elec}_([a-z0-9]+)\.csv$: one of block20, srprec, rgprec, rrprec, svprec; no match → conv.

  • category: sov_state (key sov_data), sov_local (key sov_local_*), reg (e.g. key registration, nonvoters, All_VBM), or conv (conversion/lookup files).

  • header_group: {category}_{base_geo} for reg/sov; conv_{subcategory} for conv (e.g. reg_rgprec, conv_rg_blk_map).

  • Header checks: collect_headers(elec|files_df) builds a dict header_group → DataFrame(path + header row). verify_headers(...) ensures within each group all files share the same column list; raises ValueError('Inconsistent county headers') otherwise.

3. Pydantic models (voter registration)

Base: ingestion/elec/data_models/_generic_reg.py

  • RegistrationFile — Base row model: many Annotated[NonNegativeInt, Field(le=65535)] columns (totreg_r, dem, rep, aip, … demographic/registration counts). No geography columns; those are in subclasses.

  • DuckDB: duckdb_schema(table_name, base_geo_fields, ...) produces CREATE TABLE with metadata fields (ingestion_id, filename, election, county), type, base_geo columns, then the same numeric columns as USMALLINT. Enum for type is defined but stored as VARCHAR in practice.

  • PyArrow: pyarrow_schema(base_geo_fields) is commented out; ingestion uses DuckDB + CSV, not Arrow.

Concrete models: ingestion/elec/data_models/reg_models.py

  • RegistrationFileRGprec, RegistrationFileRRprec, RegistrationFileSRprec — add election, type (voter group), and one of rgprec / rrprec / srprec; tablenames reg_rgprec, reg_rrprec, reg_srprec.

  • RegistrationFileBlock — adds tract, block; tablename reg_block20.

  • Voter group type: D20RegVoterGroupsT = Literal of subcategory strings (e.g. All_Poll, All_VBM, registration, nonvoters, voters, …). All current reg models use the same D20 set (no era-specific branching yet).

Model selection: get_data_model(election, category, base_geo, subcategory=None)

  • Only category == "reg" is implemented. For reg:
  • base_geo == "block20" → RegistrationFileBlock[Literal[election], D20RegVoterGroupsT]
  • base_geo == "rgprec" → RegistrationFileRGprec[...]
  • base_geo == "rrprec" → RegistrationFileRRprec[...]
  • base_geo == "srprec" → RegistrationFileSRprec[...]
  • Else → ValueError.
  • Other categories (sov_state, sov_local, conv) → ValueError (no model yet).

Validation: validate_elec_csv(path, election, category, base_geo, subcategory=None) — resolves model via get_data_model, opens CSV under ELEC_ROOT, validates each row with data_mdl.model_validate(row); on ValidationError logs and re-raises.

Pydantic vs DuckDB (voter reg):

  • Same logical fields: metadata (election, county, etc.) are added at insert time in DuckDB; CSV has optional county/election which are excluded from the inserted row and replaced by the ingest layer. Row counts and types align; constraints (e.g. 0–65535) live in Pydantic only; DuckDB uses USMALLINT without CHECKs.

4. SOV and conv (current state)

  • SOV (data_models/sov_models.py): Docstrings and sample CSV snippets for sov_state (e.g. TOTREG, TOTVOTE, contest columns) and sov_local (srprec/svprec, contest, jurisdiction, office, candidate, votes). No Pydantic models or get_data_model branch yet.

  • Conv (data_models/conv_models.py): Docstrings for conv file layouts (e.g. blk_mprec, rg_blk_map, sr_blk_map, srprec_to_city, mprec_srprec). No Pydantic models or DuckDB table definitions in code yet.

5. DuckDB interface

Module: ingestion/elec/duck.py

  • ElecIngestDb — One instance per (header_group / table): holds a DuckDB connection (in-memory by default), a separate log_db connection for the ingestion log, and table/schema/category.

  • Initialization: initialize_connection(duckdb_schema, database=None) — if not dryrun, runs CREATE TABLE …; database=None → in-memory DB.

  • Insert: insert(filename, election, county):

  • Deletes existing rows for that filename in the table.

  • Reads CSV from ELEC_ROOT; if CSV has county/election columns, they are excluded and replaced by the passed election/county.

  • For category == "sov_local", applies WHERE votes > 0 on the CSV read.

  • Inserts via read_csv(...) with ingestion_id, filename, election, county prepended.

  • Counts file lines (wc -l) and table row count; calls log_file_ingestion(...).

  • Writes table to Parquet under INGESTION_LOCAL_WRITE_DIR partitioned by (election, county).

  • Log: log_file_ingestion(filename, election, county, file_lines, rows_deleted, rows_inserted, seconds_elapsed) — inserts one row into elec_ingestion_log on log_db. Log table schema is defined in models.ElecIngestionLog.sql_schema() (ingestion_id, ingestion_time, table_name, election, county, file_mtime, filename, file_lines, rows_deleted, rows_inserted, seconds_elapsed).

Environment: ELEC_ROOT (CSV location), INGESTION_LOCAL_WRITE_DIR (Parquet output). Log DB is currently hardcoded in ingest.py as ingestion_log_elec.dev.db.

6. Legacy entrypoints and flow

Makefile

  • make elec-ingest runs: uv run python -m ingestion.elec.pipeline.

  • Pipeline module: ingestion/elec/pipeline.py only defines main() that prints "hello pipeline" and does not call any ingest logic. So the Makefile target does not currently run the full ingestion.

Actual ingestion flow (used by tests and any direct Python usage)

  1. Build file list: parse_elec(elec) or parse_elec_paths(path_list) from file_metadata.

  2. Optional filter: ElecJobParameters (election, category, subcategory, base_geo, county) applied to the metadata DataFrame in ingest.import_files.

  3. Header verification: collect_headers(files_df=files_df) then verify_headers(header_dict=header_dict).

  4. Per header_group: Take first row of the group; get_data_model(election, category, base_geo, subcategory) → data model class; instantiate ElecIngestDb(duckdb_schema=..., table_name=..., category=..., log_db=ingestion_log_conn, ...).

  5. Per file in group: validate_elec_csv(path, election, category, base_geo, subcategory) then duck_ingest.insert(filename=path, election=..., county=...).

Entrypoint function: ingestion.elec.ingest.import_files(params=None, files_df=None). If files_df is None, it is not populated (caller must pass it). If params is set, filtering by category/subcategory/base_geo/county is applied. Currently only the first header group is processed (header_groups[0]), so only one table per run (e.g. reg_rgprec).

Tests

  • tests/test_ingest.py: uses parse_elec_paths + import_files(files_df=...) with a single CSV path.

  • tests/test_file_metadata.py: parse_elec_paths, extract_file_metadata, collect_headers, verify_headers with fixture mock_env_elec_root pointing at ingestion/elec/tests/fixtures/elec.

  • tests/test_data_models_reg.py / test_data_models.py: reg model and validation tests.

7. Summary diagram

flowchart LR

paths["Paths under ELEC_ROOT"]

paths --> fileMetadata["file_metadata: parse_elec / parse_elec_paths"]

fileMetadata --> meta["DataFrame: path, election, county, base_geo, category, subcategory, header_group"]

meta --> filter["Optional: ElecJobParameters filter"]

filter --> headers["collect_headers + verify_headers"]

headers --> group["Per header_group"]

group --> getModel["reg_models.get_data_model"]

getModel --> duckDb["ElecIngestDb(duckdb_schema, table_name, log_db)"]

group --> forFile["For each file in group"]

forFile --> validate["validate_elec_csv (Pydantic)"]

validate --> insert["duck_ingest.insert"]

insert --> duckTables["DuckDB tables (reg_*)"]

insert --> log["elec_ingestion_log"]

insert --> parquet["Parquet under INGESTION_LOCAL_WRITE_DIR"]

8. Model-selection branching (reference)

| category | base_geo | Model class (current) |

|-----------|----------|------------------------|

| reg | block20 | RegistrationFileBlock[elec, D20RegVoterGroupsT] |

| reg | rgprec | RegistrationFileRGprec[elec, D20RegVoterGroupsT] |

| reg | rrprec | RegistrationFileRRprec[elec, D20RegVoterGroupsT] |

| reg | srprec | RegistrationFileSRprec[elec, D20RegVoterGroupsT] |

| reg | svprec | Not implemented (ValueError) |

| sov_state | * | Not implemented |

| sov_local | * | Not implemented |

| conv | conv | Not implemented |

Election/era: only one voter-group set (D20) is wired; future work can add election-specific Literals and branches in get_data_model.