[REFACTOR]Clean and organise data processing

[HAEKADI]

• [REFACTO]Nettoyer la config infra #392
• Implement Airflow best practices
• Use a RessourceManager to download, process and save datasets

[HAEKADI]

The target organisation for the repository will look something like this:

dag_datalake_sirene/
│
├── config/
│   ├── __init__.py
│   ├── constants.py  # Global constants
│   └── settings.py   # Environment-specific settings
│
├── helpers/
│   ├── __init__.py
│   ├── data_processor.py
│   ├── config_models.py
│   ├── minio_helpers.py
│   ├── tchap.py
│   ├── utils.py
│   └── sqlite_client.py
│
├── workflows/
│   ├── __init__.py
│   │
│   ├── data_pipelines/
│   │   ├── __init__.py
│   │   │
│   │   ├── egapro/
│   │   │   ├── __init__.py
│   │   │   ├── dag_egapro.py
│   │   │   ├── egapro_processor.py
│   │   │   └── task_functions.py
│   │   │
│   │   ├── rge/
│   │   │   ├── __init__.py
│   │   │   ├── dag_rge.py
│   │   │   ├── rge_processor.py
│   │   │   └── task_functions.py
│   │   │
│   │   ├── colter/
│   │   │   ├── __init__.py
│   │   │   ├── dag_colter.py
│   │   │   ├── colter_processor.py
│   │   │   └── task_functions.py
│   │   │
│   │   └── ... (other data sources)
│   │
│   └── common/
│       ├── __init__.py
│       └── common_tasks.py  # Tasks that might be shared across DAGs
│
├── tests/
│   ├── __init__.py
│   ├── conftest.py
│   │
│   ├── helpers/
│   │   ├── __init__.py
│   │   ├── test_data_processor.py
│   │   └── ... (tests for other helpers)
│   │
│   └── workflows/
│       ├── __init__.py
│       │
│       └── data_pipelines/
│           ├── __init__.py
│           ├── test_egapro.py
│           ├── test_rge.py
│           └── ... (tests for other data pipelines)
│
├── requirements.txt
├── setup.py
└── README.md

[HAEKADI]

This refactoring would involve creating a universal client or framework that can handle different data sources with similar processing patterns. Something that would look like this:

class DataProcessor(ABC):
    def __init__(self, config):
        self.config = config
        self.minio_client = minio_client

    @abstractmethod
    def download_data(self):
        pass

    @abstractmethod
    def process_data(self):
        pass

    @abstractmethod
    def save_to_minio(self):
        pass

    @abstractmethod
    def compare_files_minio(self):
        pass

    def send_notification(self, message):
        send_message(message)):
            self.send_notification()

This will be a huge undertaking given the size of this codebase.

Here's a step-by-step plan that focuses on gradually refactoring configuration management, then using a universal client for processing data sources, without drastically changing the existing structure:

1. Introduce data classes for configuration management
2. Create a universal data processing client (like shown above)
3. Refactor one data source using data classes and DataProcessor
4. Gradually refactor other data sources (each data source gets a PR)
5. Enhance DataProcessor as needed
6. Update tests
7. Documentation updates

In order to avoid bugs as much as possible, and simplify reviewing efforts, each step will preferably have its own PR.

[HAEKADI]

@XavierJp Any feedback so far ?

[XavierJp]

It is... beautiful 🥹

[XavierJp]

Honestly sounds very relevant. Step by step is always a good choice. Will start the migration with the most complicated clients or the simpler ones ?

[HAEKADI]

Data sources :

• EGAPRO
• RGE
• COLTER
• UAI
• FINESS
• FORMATION
• SPECTACLE
• AGENCE_BIO
• ESS_FRANCE
• SIRENE
• RNE
• BILANS FINANCIERS
• MARCHÉ INCLUSION
• CONVENTION COLLECTIVE
• ETL

[HAEKADI]

I’m thinking of implementing an initial version of the client focused on a straightforward data source, such as EGAPRO (see related draft).

Then each PR will add a new data source, potentially introducing additional layers of complexity.

This is very much a work in progress. Many improvements are coming (so don't mind the naming conventions etc).

[HAEKADI]

@XavierJp What do you think?

[XavierJp]

Tottally agree. You could even do one or two basic sources, then the hard ones like insee and rne. Thus ensuring the model is both straightforward and flexible enough.

[HAEKADI]

Potential enhancement :

• Add a MinIOFile and an AirflowFile to distinguish types of files while processing.
• Use Dataclass and Pydantic when possible instead of TypeDict.