Summary

• add a streaming CLI to migrate MongoDB event collections into the partitioned Parquet dataset
• expose reusable manifest helpers in src/api.py so the CLI matches the fetcher’s normalization
• document the migration workflow and add the required optional dependencies

Testing

• python -m compileall scripts/migrate_mongo_to_parquet.py src/api.py
• python - <<'PY'
  from click.testing import CliRunner
  from datetime import datetime, timezone
  from pathlib import Path
  from tempfile import TemporaryDirectory
  from unittest.mock import patch

import mongomock

from scripts.migrate_mongo_to_parquet import main as migrate
from src import api

client = mongomock.MongoClient()
db = client['fmriprep_stats']
now = datetime(2024, 1, 15, 12, 0, tzinfo=timezone.utc)

for idx, event in enumerate(api.ISSUES, 1):
coll = db[event]
for i in range(3):
coll.insert_one(
{
"_id": f"mongo-{event}-{i}",
"id": f"{event}-{i}",
"dateCreated": now.isoformat(),
"tags": [
{"key": "environment", "value": "prod"},
{"key": "platform", "value": "linux"},
],
}
)

runner = CliRunner()

with TemporaryDirectory() as tmpdir:
dataset_root = Path(tmpdir) / "dataset"

with patch("scripts.migrate_mongo_to_parquet.MongoClient", lambda uri: client):
    result1 = runner.invoke(migrate, ["--mongo-uri", "mongodb://localhost", "--db-name", "fmriprep_stats", str(dataset_root)])
    result2 = runner.invoke(migrate, ["--mongo-uri", "mongodb://localhost", "--db-name", "fmriprep_stats", str(dataset_root)])

assert result1.exit_code == 0, result1.output
assert result2.exit_code == 0, result2.output

manifest_path = api._manifest_path(dataset_root)
manifest = api._load_manifest(manifest_path)
assert len(manifest) == len(api.ISSUES) * 3
assert manifest['id'].nunique() == len(api.ISSUES) * 3

files = list(dataset_root.rglob('*.parquet'))
print('files', len(files))
print('rows', len(manifest))

PY

https://chatgpt.com/codex/tasks/task_e_68db981765988330bb8574b8d28c19ce