How annuaire-entreprises-data-gouv-fr/search-infra works
This repository is the data-ingestion and indexing infrastructure behind the Annuaire des Entreprises search API. Airflow workflows collect French public business data, preprocess it into object storage, build and enrich a SQLite ETL database, then create Elasticsearch indexes and snapshots used by API instances. The graph emphasizes those documented product stages and representative source-specific pipelines; the README establishes the overall flow, while excerpts show selected storage, ETL, and indexing details. Many individual DAGs and processors were not sampled, so their internal links are not inferred.
Main entry points: object_storage.py Object storage · index.py Index creation and loading · dag.py ETL workflow · database_constructor.py SQLite table construction
Components 20
Data ingestion
- Source pipelines
workflows/data_pipelines - SIRENE pipelines
workflows/data_pipelines/sirene - RNE flux ingestion
workflows/data_pipelines/rne/flux/flux_tasks.py - DataGouv processing
workflows/data_pipelines/data_gouv/processor.py - Object storage
helpers/object_storage.py
ETL and enrichment
- ETL workflow
workflows/data_pipelines/etl/dag.py - ETL table tasks
workflows/data_pipelines/etl/task_functions - SQLite table construction
helpers/database_constructor.py - SQLite client
helpers/sqlite_client.py - SQLite database
workflows/data_pipelines/etl/sqlite - Search data enrichment
workflows/data_pipelines/elasticsearch/data_enrichment.py
Search publishing
- Index workflow
workflows/data_pipelines/elasticsearch/dag_index_data.py - Index creation and loading
workflows/data_pipelines/elasticsearch/task_functions/index.py - Index selection queries
workflows/data_pipelines/elasticsearch/sqlite - Elasticsearch indexes
workflows/data_pipelines/elasticsearch - Snapshot workflow
workflows/data_pipelines/elasticsearch/dag_snapshot_data.py
Data operations
- Storage maintenance
workflows/maintenance/dag_clean_object_storage.py
Other components
- Public data sources
- Airflow workflows
- Search API instances
Connections 24
- Airflow workflows → Source pipelines: runs
- Public data sources → Source pipelines: provides data
- Source pipelines → Object storage: stores preprocessed data
- SIRENE pipelines → Object storage: stores flux and stock
- RNE flux ingestion → Object storage: downloads and uploads
workflows/data_pipelines/rne/flux/flux_tasks.py - DataGouv processing → Object storage: uploads processed files
workflows/data_pipelines/data_gouv/processor.py - Airflow workflows → ETL workflow: runs
- Object storage → ETL workflow: supplies stored data
- ETL workflow → ETL table tasks: coordinates
- ETL table tasks → SQLite table construction: builds tables
- SQLite table construction → Object storage: reads preprocessed CSV
helpers/database_constructor.py - SQLite table construction → SQLite client: executes table operations
helpers/database_constructor.py - SQLite client → SQLite database: reads and writes
- ETL workflow → Search data enrichment: enriches processed data
- Airflow workflows → Index workflow: runs
- Index workflow → Index creation and loading: coordinates indexing
- Index creation and loading → SQLite database: reads processed records
workflows/data_pipelines/elasticsearch/task_functions/index.py - Index creation and loading → Index selection queries: executes selection queries
workflows/data_pipelines/elasticsearch/task_functions/index.py - Index creation and loading → Elasticsearch indexes: creates and fills indexes
workflows/data_pipelines/elasticsearch/task_functions/index.py - Index workflow → Snapshot workflow: triggers snapshots
- Snapshot workflow → Elasticsearch indexes: snapshots index
- Elasticsearch indexes → Search API instances: feeds search API
- Airflow workflows → Storage maintenance: runs scheduled cleanup
workflows/maintenance/dag_clean_object_storage.py - Storage maintenance → Object storage: lists and deletes old files
workflows/maintenance/dag_clean_object_storage.py