slubprefect-mongodbsolr 1.0.2

Prefect 3.4 Flow-Paket für MongoDB→Solr Synchronisation

MongoDB → Solr Sync

Prefect-3-basiertes Projekt, das Daten aus MongoDB nach Solr synchronisiert. Sämtlicher Code liegt im Python-Paket mongodbsolr unter src/mongodbsolr.

Das Paket wird als Python-Distribution unter dem Namen slubprefect-mongodbsolr veröffentlicht und kann bei Bedarf über PyPI installiert werden.

🎯 Zielgruppe & Einsatzszenario

• Primär: internes Developer-Team der SLUB Dresden (Datenmanagement / DMG)
• Typischer Einsatz: periodische oder ad-hoc Synchronisation von Dokumenten aus MongoDB in einen Solr-Core
• Infrastruktur: Prefect 3, MongoDB-Cluster, Solr-Instanz und Prefect-Server

🚀 Schnellstart

1. Umgebung einrichten

uv sync
source .venv/bin/activate

2. Secrets & Konfiguration

export MONGODB_URI="mongodb://localhost:27017"
export SOLR_URL="http://localhost:8983/solr"
python scripts/create_mongodbsolr_config.py create-secrets

python scripts/create_mongodbsolr_config.py create-variable \
  --mongodb-db my_database \
  --mongodb-collection my_collection \
  --mongodb-query '{"status": "published"}' \
  --solr-core my_core \
  --mongodb-id-field record_id \
  --solr-id-field id

Struktur der Prefect-Variable mongodbsolr-config

Die Variable mongodbsolr-config enthält ein JSON-Objekt mit mehreren Blöcken:

{
  "mongodb": {
    "uri_block": "mongodbsolr-mongodb-uri",
    "database": "...",
    "collection": "...",
    "query": "...",
    "batch_size": 100
  },
  "solr": {
    "url_block": "mongodbsolr-solr-url",
    "core": "my_core",
    "commit_within": 5000,
    "max_concurrent_upserts": 2,
    "finalcommit": false
  },
  "mapping": {
    "mongodb_id_field": "record_id",
    "solr_id_field": "id",
    "exclude_fields": ["_id", "_rev", "_attachments"],
    "required_fields": []
  },
  "mode": "upsert",
  "delete_query": "status:published"
}

Felder im Überblick:

• mongodb.batch_size – Anzahl Dokumente pro gelesenem Batch aus MongoDB (Standard: 100).
• solr.commit_within – commitWithin in Millisekunden, das an Solr übergeben wird (Standard: 5000).
• solr.max_concurrent_upserts (optional) – maximale Anzahl paralleler Solr-Upsert-Tasks im Flow. Wenn nicht gesetzt, wird 2 verwendet.
• solr.finalcommit (optional) – steuert, ob am Ende des Flows ein expliziter commit gegen Solr ausgeführt wird:
  • false (Standard): nur commitWithin wird verwendet, Solr entscheidet selbst, wann Commits passieren.
  • true: nach Abschluss aller Upserts/Deletes wird zusätzlich ein finaler Commit ausgelöst.

Hinweis: mode und delete_query können wie bisher verwendet werden (upsert, delete-before-import, delete-only).

3. Flow ausführen

python -m mongodbsolr.flows.cli sync --config mongodbsolr-config

Optionale Flags:

Flag	Wirkung
--mode	Überschreibt den Modus (upsert, delete-before-import, delete-only)
--explain-indexes	Führt explain('executionStats') aus und prüft Indexnutzung

📦 Installation als Paket

Das Paket wird auf PyPI unter dem Namen slubprefect-mongodbsolr veröffentlicht.

pip install slubprefect-mongodbsolr

Anschließend kann das CLI wie oben beschrieben mit python -m mongodbsolr.flows.cli genutzt werden.

🧱 Struktur

src/
└── mongodbsolr/
    ├── flows/
    │   ├── cli.py             # CLI-Entry für Prefect Flows
    │   └── sync_flow.py       # Haupt-Flow mongodb_to_solr_sync
    ├── tasks/
    │   ├── mongodb.py         # MongoDB-spezifische Tasks
    │   └── solr.py            # Solr-spezifische Tasks
    └── utils/
        ├── config.py          # Prefect Variables & Secret Blocks laden
        └── validation.py      # Pflichtfeldprüfung & Feldfilterung

scripts/
├── create_mongodbsolr_config.py   # Secrets/Variable anlegen
├── deploy_mongodbsolr_eth.py      # Beispiel-Deployment
└── inspect_prefect_variable.py    # Debugging von Prefect-Variablen

doc/plan.md                        # Architektur- und Roadmap-Dokument
prefect.yaml                       # Deployments

⚙️ Flow & Tasks

• flows/sync_flow.py: Implementiert den Flow mongodb_to_solr_sync, erzeugt Prefect-Artifacts und koordiniert Modi upsert, delete-before-import, delete-only.
• tasks/mongodb.py: Optionaler Explain-Check, Cursor-Fetching, Batch-Verarbeitung, ID-Extraktion.
• tasks/solr.py: Delete-by-Query, Delete-by-IDs, Upsert-Batches und Commit.
• utils/config.py: Lädt Prefect Variable, validiert Inhalte und löst Secret Blocks auf.
• utils/validation.py: Pflichtfeld-Validierung, ID-Mapping, Feldfilterung.

🧪 Tests & Linting

ruff format .
ruff check .
python -m mongodbsolr.flows.cli test-config --config mongodbsolr-config

📦 Deployments

Standardflow: mongodbsolr.flows.sync_flow:mongodb_to_solr_sync (siehe prefect.yaml).

Beispiel-Deployment erstellen:

python scripts/deploy_mongodbsolr_eth.py

oder direkt via Prefect CLI:

prefect deploy --name mongodbsolr-sync

Spezielles Deploy-Skript für mongodbsolr

python deploy.py --dockerfile Dockerfile.yml --image-repo registry.git.slub-dresden.de/metadata/mongodbsolr

Parameter:

Flag	Beschreibung
--tag	Optionaler Image-Tag (Standard: Zeitstempel UTC)
--no-push	Überspringt Push ins Registry
--deployment-name	Prefect-Deployment-Name (Standard: mongodbsolr-sync)
--log-level	Log-Level für das Skript

📚 Weitere Quellen

• Technischer Plan: doc/plan.md
• Prefect UI: https://prefect.dmg.slub-dresden.de

📝 Lizenz

Siehe LICENSE im Repository.