# annuaire-entreprises-data-gouv-fr/search-infra architecture

> Architecture diagram of the GitHub repository annuaire-entreprises-data-gouv-fr/search-infra, generated by GitDiagram from its file tree, README and a sample of source files (last updated 2026-09-30). It is an AI-made overview: check details against the source.

- Interactive diagram: https://gitdiagram.com/annuaire-entreprises-data-gouv-fr/search-infra
- Repository: https://github.com/annuaire-entreprises-data-gouv-fr/search-infra

## Overview

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: [`helpers/object_storage.py`](https://github.com/annuaire-entreprises-data-gouv-fr/search-infra/blob/main/helpers/object_storage.py) (Object storage), [`workflows/data_pipelines/elasticsearch/task_functions/index.py`](https://github.com/annuaire-entreprises-data-gouv-fr/search-infra/blob/main/workflows/data_pipelines/elasticsearch/task_functions/index.py) (Index creation and loading), [`workflows/data_pipelines/etl/dag.py`](https://github.com/annuaire-entreprises-data-gouv-fr/search-infra/blob/main/workflows/data_pipelines/etl/dag.py) (ETL workflow), [`helpers/database_constructor.py`](https://github.com/annuaire-entreprises-data-gouv-fr/search-infra/blob/main/helpers/database_constructor.py) (SQLite table construction)

## Diagram (Mermaid)

```mermaid
flowchart TD

subgraph group_ingestion["Data ingestion"]
  node_source_pipelines["Source pipelines"]
  node_sirene_pipeline["SIRENE pipelines"]
  node_rne_flux["RNE flux ingestion<br/>[flux_tasks.py]"]
  node_datagouv_processor["DataGouv processing<br/>[processor.py]"]
  node_object_storage[("Object storage<br/>[object_storage.py]")]
end

subgraph group_transform["ETL and enrichment"]
  node_etl_workflow["ETL workflow<br/>[dag.py]"]
  node_etl_tasks["ETL table tasks"]
  node_sqlite_constructor["SQLite table construction"]
  node_sqlite_client["SQLite client<br/>[sqlite_client.py]"]
  node_sqlite_db[("SQLite database")]
  node_search_enrichment["Search data enrichment<br/>[data_enrichment.py]"]
end

subgraph group_search["Search publishing"]
  node_index_workflow["Index workflow<br/>[dag_index_data.py]"]
  node_index_tasks["Index creation and loading<br/>[index.py]"]
  node_index_queries["Index selection queries"]
  node_elasticsearch[("Elasticsearch indexes")]
  node_snapshot_workflow["Snapshot workflow"]
end

subgraph group_operations["Data operations"]
  node_maintenance["Storage maintenance"]
end

node_sources(("Public data sources"))
node_airflow(("Airflow workflows"))
node_api_instances(("Search API instances"))

node_airflow -->|"runs"| node_source_pipelines
node_sources -->|"provides data"| node_source_pipelines
node_source_pipelines -->|"stores preprocessed data"| node_object_storage
node_sirene_pipeline -.->|"stores flux and stock"| node_object_storage
node_rne_flux -->|"downloads and uploads"| node_object_storage
node_datagouv_processor -->|"uploads processed files"| node_object_storage
node_airflow -->|"runs"| node_etl_workflow
node_object_storage -->|"supplies stored data"| node_etl_workflow
node_etl_workflow -.->|"coordinates"| node_etl_tasks
node_etl_tasks -.->|"builds tables"| node_sqlite_constructor
node_sqlite_constructor -->|"reads preprocessed CSV"| node_object_storage
node_sqlite_constructor -->|"executes table operations"| node_sqlite_client
node_sqlite_client -.->|"reads and writes"| node_sqlite_db
node_etl_workflow -->|"enriches processed data"| node_search_enrichment
node_airflow -->|"runs"| node_index_workflow
node_index_workflow -.->|"coordinates indexing"| node_index_tasks
node_index_tasks -->|"reads processed records"| node_sqlite_db
node_index_tasks -->|"executes selection queries"| node_index_queries
node_index_tasks -->|"creates and fills indexes"| node_elasticsearch
node_index_workflow -->|"triggers snapshots"| node_snapshot_workflow
node_snapshot_workflow -->|"snapshots index"| node_elasticsearch
node_elasticsearch -->|"feeds search API"| node_api_instances
node_airflow -->|"runs scheduled cleanup"| node_maintenance
node_maintenance -->|"lists and deletes old files"| node_object_storage

click node_source_pipelines "https://github.com/annuaire-entreprises-data-gouv-fr/search-infra/tree/main/workflows/data_pipelines"
click node_sirene_pipeline "https://github.com/annuaire-entreprises-data-gouv-fr/search-infra/tree/main/workflows/data_pipelines/sirene"
click node_rne_flux "https://github.com/annuaire-entreprises-data-gouv-fr/search-infra/blob/main/workflows/data_pipelines/rne/flux/flux_tasks.py"
click node_datagouv_processor "https://github.com/annuaire-entreprises-data-gouv-fr/search-infra/blob/main/workflows/data_pipelines/data_gouv/processor.py"
click node_object_storage "https://github.com/annuaire-entreprises-data-gouv-fr/search-infra/blob/main/helpers/object_storage.py"
click node_etl_workflow "https://github.com/annuaire-entreprises-data-gouv-fr/search-infra/blob/main/workflows/data_pipelines/etl/dag.py"
click node_etl_tasks "https://github.com/annuaire-entreprises-data-gouv-fr/search-infra/tree/main/workflows/data_pipelines/etl/task_functions"
click node_sqlite_constructor "https://github.com/annuaire-entreprises-data-gouv-fr/search-infra/blob/main/helpers/database_constructor.py"
click node_sqlite_client "https://github.com/annuaire-entreprises-data-gouv-fr/search-infra/blob/main/helpers/sqlite_client.py"
click node_sqlite_db "https://github.com/annuaire-entreprises-data-gouv-fr/search-infra/tree/main/workflows/data_pipelines/etl/sqlite"
click node_search_enrichment "https://github.com/annuaire-entreprises-data-gouv-fr/search-infra/blob/main/workflows/data_pipelines/elasticsearch/data_enrichment.py"
click node_index_workflow "https://github.com/annuaire-entreprises-data-gouv-fr/search-infra/blob/main/workflows/data_pipelines/elasticsearch/dag_index_data.py"
click node_index_tasks "https://github.com/annuaire-entreprises-data-gouv-fr/search-infra/blob/main/workflows/data_pipelines/elasticsearch/task_functions/index.py"
click node_index_queries "https://github.com/annuaire-entreprises-data-gouv-fr/search-infra/tree/main/workflows/data_pipelines/elasticsearch/sqlite"
click node_elasticsearch "https://github.com/annuaire-entreprises-data-gouv-fr/search-infra/tree/main/workflows/data_pipelines/elasticsearch"
click node_snapshot_workflow "https://github.com/annuaire-entreprises-data-gouv-fr/search-infra/blob/main/workflows/data_pipelines/elasticsearch/dag_snapshot_data.py"
click node_maintenance "https://github.com/annuaire-entreprises-data-gouv-fr/search-infra/blob/main/workflows/maintenance/dag_clean_object_storage.py"

classDef toneNeutral fill:#f8fafc,stroke:#334155,stroke-width:1.5px,color:#0f172a
classDef toneBlue fill:#dbeafe,stroke:#2563eb,stroke-width:1.5px,color:#172554
classDef toneAmber fill:#fef3c7,stroke:#d97706,stroke-width:1.5px,color:#78350f
classDef toneMint fill:#dcfce7,stroke:#16a34a,stroke-width:1.5px,color:#14532d
classDef toneRose fill:#ffe4e6,stroke:#e11d48,stroke-width:1.5px,color:#881337
classDef toneIndigo fill:#e0e7ff,stroke:#4f46e5,stroke-width:1.5px,color:#312e81
classDef toneTeal fill:#ccfbf1,stroke:#0f766e,stroke-width:1.5px,color:#134e4a
class node_source_pipelines,node_sirene_pipeline,node_rne_flux,node_datagouv_processor,node_object_storage toneBlue
class node_etl_workflow,node_etl_tasks,node_sqlite_constructor,node_sqlite_client,node_sqlite_db,node_search_enrichment toneAmber
class node_index_workflow,node_index_tasks,node_index_queries,node_elasticsearch,node_snapshot_workflow,node_api_instances toneMint
class node_maintenance toneRose
class node_sources,node_airflow toneIndigo
```

## Components

### Data ingestion

- [Source pipelines](https://github.com/annuaire-entreprises-data-gouv-fr/search-infra/tree/main/workflows/data_pipelines): `workflows/data_pipelines`
- [SIRENE pipelines](https://github.com/annuaire-entreprises-data-gouv-fr/search-infra/tree/main/workflows/data_pipelines/sirene): `workflows/data_pipelines/sirene`
- [RNE flux ingestion](https://github.com/annuaire-entreprises-data-gouv-fr/search-infra/blob/main/workflows/data_pipelines/rne/flux/flux_tasks.py): `workflows/data_pipelines/rne/flux/flux_tasks.py`
- [DataGouv processing](https://github.com/annuaire-entreprises-data-gouv-fr/search-infra/blob/main/workflows/data_pipelines/data_gouv/processor.py): `workflows/data_pipelines/data_gouv/processor.py`
- [Object storage](https://github.com/annuaire-entreprises-data-gouv-fr/search-infra/blob/main/helpers/object_storage.py): `helpers/object_storage.py`

### ETL and enrichment

- [ETL workflow](https://github.com/annuaire-entreprises-data-gouv-fr/search-infra/blob/main/workflows/data_pipelines/etl/dag.py): `workflows/data_pipelines/etl/dag.py`
- [ETL table tasks](https://github.com/annuaire-entreprises-data-gouv-fr/search-infra/tree/main/workflows/data_pipelines/etl/task_functions): `workflows/data_pipelines/etl/task_functions`
- [SQLite table construction](https://github.com/annuaire-entreprises-data-gouv-fr/search-infra/blob/main/helpers/database_constructor.py): `helpers/database_constructor.py`
- [SQLite client](https://github.com/annuaire-entreprises-data-gouv-fr/search-infra/blob/main/helpers/sqlite_client.py): `helpers/sqlite_client.py`
- [SQLite database](https://github.com/annuaire-entreprises-data-gouv-fr/search-infra/tree/main/workflows/data_pipelines/etl/sqlite): `workflows/data_pipelines/etl/sqlite`
- [Search data enrichment](https://github.com/annuaire-entreprises-data-gouv-fr/search-infra/blob/main/workflows/data_pipelines/elasticsearch/data_enrichment.py): `workflows/data_pipelines/elasticsearch/data_enrichment.py`

### Search publishing

- [Index workflow](https://github.com/annuaire-entreprises-data-gouv-fr/search-infra/blob/main/workflows/data_pipelines/elasticsearch/dag_index_data.py): `workflows/data_pipelines/elasticsearch/dag_index_data.py`
- [Index creation and loading](https://github.com/annuaire-entreprises-data-gouv-fr/search-infra/blob/main/workflows/data_pipelines/elasticsearch/task_functions/index.py): `workflows/data_pipelines/elasticsearch/task_functions/index.py`
- [Index selection queries](https://github.com/annuaire-entreprises-data-gouv-fr/search-infra/tree/main/workflows/data_pipelines/elasticsearch/sqlite): `workflows/data_pipelines/elasticsearch/sqlite`
- [Elasticsearch indexes](https://github.com/annuaire-entreprises-data-gouv-fr/search-infra/tree/main/workflows/data_pipelines/elasticsearch): `workflows/data_pipelines/elasticsearch`
- [Snapshot workflow](https://github.com/annuaire-entreprises-data-gouv-fr/search-infra/blob/main/workflows/data_pipelines/elasticsearch/dag_snapshot_data.py): `workflows/data_pipelines/elasticsearch/dag_snapshot_data.py`

### Data operations

- [Storage maintenance](https://github.com/annuaire-entreprises-data-gouv-fr/search-infra/blob/main/workflows/maintenance/dag_clean_object_storage.py): `workflows/maintenance/dag_clean_object_storage.py`

### Other components

- **Public data sources**
- **Airflow workflows**
- **Search API instances**

## Connections

- 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 (evidence: `workflows/data_pipelines/rne/flux/flux_tasks.py`)
- DataGouv processing → Object storage: uploads processed files (evidence: `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 (evidence: `helpers/database_constructor.py`)
- SQLite table construction → SQLite client: executes table operations (evidence: `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 (evidence: `workflows/data_pipelines/elasticsearch/task_functions/index.py`)
- Index creation and loading → Index selection queries: executes selection queries (evidence: `workflows/data_pipelines/elasticsearch/task_functions/index.py`)
- Index creation and loading → Elasticsearch indexes: creates and fills indexes (evidence: `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 (evidence: `workflows/maintenance/dag_clean_object_storage.py`)
- Storage maintenance → Object storage: lists and deletes old files (evidence: `workflows/maintenance/dag_clean_object_storage.py`)