Milestone 2 : Alpha Release Report

[kshitijrajsharma]

Report goes here !

[gauravbarall]

Pydantic datamodel migration complete #16 . Closed by #31 #34

Background

Prior to this change, all data structures were bare global dictionaries:

users_temp: dict        = {}
users: dict             = {}
summary_interval: dict  = {}
summary_interval_temp:  = {}
processed_changesets:   = {}

These offered no type guarantees, no schema documentation, and no safe serialisation path. They were dynamically shaped, and mutated. The shape of a user record was only known by reading the entire function.

---

What changed

All mutable accumulator state has been replaced with Pydantic models. Each accumulator model owns its fields, default values, and a to_row() method that produces a fixed-width tuple for the database layer.

---

Final schema

User

Field	Type	Description
uid	int	OSM user ID
username	str	OSM display name

---

Changeset

Field	Type	Default	Description
changeset_id	int	—	OSM changeset ID
uid	int	—	Author's OSM user ID
created_at	datetime | None	None	Changeset open timestamp (UTC)
hashtags	list[str]	[]	Hashtags extracted from comment + hashtags tag
editor	str | None	None	Raw created_by tag value
bbox	tuple[float,float,float,float] | None	None	(min_lon, min_lat, max_lon, max_lat)

---

ElementStat

Tracks create / modify / delete counts for a single OSM element type (nodes, ways, relations).

Field	Type	Default	Description
c	int	0	Create count
m	int	0	Modify count
d	int	0	Delete count

---

TagValueStat

Tracks create / modify counts and optional accumulated length for a single (tag_key, tag_value) pair.

Field	Type	Default	Description
c	int	0	Create count
m	int	0	Modify count
len	float | None	None	Accumulated haversine length in metres (ways only, create action)

---

ChangesetStats

Per-changeset accumulator. One instance per changeset ID encountered in a changefile.

Field	Type	Default	Description
changeset_id	int	—	OSM changeset ID
uid	int	—	Author's OSM user ID
seq_id	int	—	Replication sequence number this stat came from
nodes	ElementStat	ElementStat()	Node create/modify/delete counts
ways	ElementStat	ElementStat()	Way create/modify/delete counts
rels	ElementStat	ElementStat()	Relation create/modify/delete counts
poi_created	int	0	Nodes with tags created
poi_modified	int	0	Nodes with tags modified
tag_stats	dict[str, dict[str, TagValueStat]]	{}	Nested tag key → tag value → TagValueStat; populated when --all_tags or --tags is active

to_row() flattens this to a 15-column tuple and serialises tag_stats to a JSON string, making it directly insertable into the DB layer .

---

Migration mapping (old → new)

Old global dict	New model / location
users: dict	ChangefileHandler.users: dict[int, User]
users_temp: dict (changeset dedup)	removed — dedup is now done at the DB aggregation layer
processed_changesets: dict	ChangesetHandler.changesets: dict[int, Changeset]
summary_interval: dict	removed — summary is derived by DB queries over ChangesetStats
summary_interval_temp: dict	removed — same as above

---

Benefits

• Type safety — field types are validated on assignment; passing the wrong type raises a ValidationError immediately rather than silently producing a wrong number downstream.
• Explicit schema — every accumulator's shape is documented in one place (models.py), not inferred by reading handler logic.
• Safe serialisation — to_row() on each model produces a deterministic, fixed-width tuple; the DB schema and the model schema cannot drift independently.
• No more global mutable state — all accumulators live on handler instances, making the pipeline stateless between runs and safe for concurrent use in the ProcessPoolExecutor.

[gauravbarall]

DuckDB as the persistence and query layer #37 — closed by #31 #34

Why DuckDB

The old codebase accumulated all stats in plain Python dicts (users, users_temp, summary_interval) held in memory for the entire run. At the end of main(), everything was dumped into a pandas DataFrame for sorting, filtering, and export. This meant the full dataset had to fit in RAM, there was no resume capability, and querying anything post-execution required re-running the tool.

DuckDB fits naturally here, it's an embedded analytical database with no server, zero config, and native Parquet read/write. It gives us persistent state, SQL queries over the results, and a clean separation between ingestion and querying.

---

How it's structured

The DB layer lives in osmsg/db/:

osmsg/db/
├── __init__.py
├── schema.py    # table definitions + state/resume helpers
├── ingest.py    # per-worker parquet writers + bulk merge into DuckDB
└── queries.py   # user_stats, attach_metadata, attach_tag_stats, daily_summary

Schema (db/schema.py)

Four tables:

Ingest (db/ingest.py)

Workers run in separate processes, each worker flushes its results to small Parquet files (temp_{pid}_{name}_{batch}.parquet) where pid is the process id for each unique process, name is the name of the table and batch is the unique batch id for each file processed within the process. Once all workers finish, merge_parquet_files() reads all files in one INSERT … SELECT FROM read_parquet(glob) transaction and cleans up.

# Each worker writes its own parquet file
flush_rows_to_parquet(parquet_dir=..., pid=os.getpid(), batch_index=..., users=[...], changesets=[...])

# Pipeline merges everything at the end
dbmod.merge_parquet_files(conn, cs_dir, cleanup=True)

This avoids any cross-process locking — workers never touch the DuckDB file
directly, only write Parquet. The merge is a single-writer operation that runs
after the pool closes.

Queries (db/queries.py)

All result queries are plain SQL against the four tables — no pandas involved. user_stats() does the aggregation in DuckDB and returns plain list[dict]. attach_metadata() and attach_tag_stats() enrich those rows in-place with a second pass. daily_summary() runs a separate GROUP BY DATE_TRUNC('day', created_at) query to produce one row per UTC day which is used by --summary to write the _summary output files.

Resume with --update (state table)

The state table stores the last processed sequence number and timestamp per source URL. --update reads this to pick up exactly where the previous run ended, and upsert_state() writes it back after each URL is processed. If a source URL has no
prior state row, --update raises early with a clear error rather than
silently re-processing from an arbitrary point.

---

What this replaced

Old	New
In-memory Python dicts for the whole run	DuckDB tables, flushed incrementally
pandas DataFrame for sorting/aggregation	SQL GROUP BY, ORDER BY in DuckDB
Results lost after export	.duckdb file stays on disk, queryable anytime
No multi-process coordination	Per-worker Parquet files merged at end
No resume capability	state table + --update flag for incremental runs

[gauravbarall]

Implemented parallel processing and migrated from pandas to PyArrow #17 — closed by #31 #34

---

pandas is completely removed from the codebase. We went with PyArrow for per-worker shard writing and DuckDB for aggregation and querying.

How it works now

The old approach accumulated all parsed rows in memory across every file, then built a pandas DataFrame at the end and dumped it. For a multi-day run that meant holding the entire result set in RAM before writing a single byte to disk.

Now each worker flushes to disk immediately after processing one file:

1. Worker parses one OSM file via osmium, accumulates rows only for that file.
2. flush_rows_to_parquet() writes a small typed temp_{pid}_{name}_{batch}.parquet shard via PyArrow and the worker's memory is freed.
3. Once all workers finish, merge_parquet_files() runs a single DuckDB INSERT … SELECT … FROM read_parquet('temp_*_…_*.parquet') glob — DuckDB streams the shards in without pulling everything into Python heap.

Peak RAM per worker is now proportional to one file's worth of rows, not the entire run.

PyArrow schemas are strict

The old pandas path let column types drift silently. PyArrow schemas in db/ingest.py are explicit and enforced at write time — type mismatches raise immediately at the worker rather than producing subtly wrong output.

CHANGESET_STATS_SCHEMA = pa.schema([
    pa.field("changeset_id", pa.int64(), nullable=False),
    pa.field("nodes_created", pa.int32()),
    pa.field("tag_stats",     pa.string()),
    # …
])

Parallelism

File processing uses ProcessPoolExecutor. Each process writes shards named with its own PID so there are no cross-worker write conflicts.

[gauravbarall]

Export layer extracted from monolith to dedicated export/ subpackage #24 - Closed by #34

What we did

We pulled all output/export logic out of the monolithic app.py and into a dedicated osmsg/export/ subpackage. Each format is now its own small module that the pipeline calls after query results are ready — the replication → processor → output layering the issue asked for.

osmsg/export/
├── __init__.py
├── csv.py
├── json.py
├── markdown.py
├── parquet.py
└── psql.py

The pipeline just does:

if "parquet" in cfg.formats:
    written["parquet"] = str(to_parquet(rows, out / f"{cfg.name}.parquet"))
if "csv" in cfg.formats:
    written["csv"] = str(to_csv(rows, out / f"{cfg.name}.csv"))
# ... and so on

No exporter knows anything about parsing or replication. New formats can be added by dropping a file in export/ and registering it in EXPORT_FORMATS inside __init__.py

---

What changed from the old code

The old app.py had file.write(f"...") and df.to_csv() calls sitting directly inside main(). It pulled in pandas, matplotlib, geopandas, and humanize just to serialize results. Adding a new format meant editing the same monolithic function.

Key improvements:

• Parquet is now the default output (was CSV). Uses PyArrow directly — no pandas needed.
• CSV and JSON use stdlib only. csv.DictWriter and json.dumps — no heavy deps.
• Markdown no longer needs humanize or matplotlib. A small local _human() helper handles compact number display.
• PostgreSQL is a new format ( -f psql). Uses DuckDB's built-in postgres extension to stream all tables directly into a Postgres target — nothing like this existed before.

---

What we didn't carry forward

• --charts — chart generation was tightly coupled to matplotlib and pandas DataFrame columns. Dropped from the output step. Users can visualize the Parquet/DuckDB output downstream with whatever tool they prefer.

---

[gauravbarall]

Authentication upgrade to OAuth 2.0 completed #32 - Closed by #33

OAuth 1.0a and HTTP Basic Auth were shut down by OSM in 2024, breaking Geofabrik authenticated downloads entirely. This PR removes the old auth path and replaces it with a proper OAuth 2.0 flow.

[gauravbarall]

Added support to dump stats to Postgresql #10 - Closed by #33

How to use it

Pass -f psql and --psql-dsn along with a libpq connection string. Example:

uv run osmsg --last day -f psql --psql-dsn "host=localhost dbname=osmsg user=osmsg password=secret port=5433" --changeset

Or combine it with other formats in the same run:

uv run osmsg --last day -f parquet -f psql --psql-dsn "host=localhost dbname=osmsg user=osmsg password=secret port=5433" --changeset

What was implemented

The export lives in osmsg/export/psql.py. It uses DuckDB's built-in postgres
extension to attach the target database and stream all four osmsg tables
(users, changesets, changeset_stats, state) directly — no intermediate
files, no extra Python dependencies beyond what was already required.

The Postgres schema mirrors the DuckDB schema exactly (same column names, types,
indexes, and foreign keys), so both backends are queryable with identical SQL.
users, changesets, and changeset_stats use ON CONFLICT DO NOTHING for
safe re-runs; the state table upserts so the resume marker always reflects the
latest run.

The --psql-dsn flag is validated at CLI startup — if -f psql is passed
without --psql-dsn, osmsg exits early with a clear error rather than failing
mid-run.

[gauravbarall]

Integrated CI for Ruff formatting #14 and added ty static checks instead of mypy #13 - Closed by #33