feat: create ref_natinfs flow

Author: maximeperraultdev

backend/src/main/resources/db/migration/internal/V0.204__create_ref_natinfs_table.sql

@@ -0,0 +1,8 @@
+CREATE TABLE public.ref_natinfs
+(
+    id            SERIAL PRIMARY KEY,
+    nature        VARCHAR,
+    qualification TEXT,
+    defined_by    VARCHAR,
+    repressed_by  VARCHAR
+);

pipeline/src/flows/ref_natinfs.py

@@ -0,0 +1,58 @@
+from pathlib import Path
+import pandas as pd
+from prefect import flow, get_run_logger, task
+
+from config import LIBRARY_LOCATION
+from src.generic_tasks import load
+
+
+
+@task
+def get_csv_file(directory: str) -> Path:
+    directory = Path(directory)
+    if not directory.exists():
+        raise FileNotFoundError(f"Le dossier {directory} n'existe pas")
+
+    csv_files = list(directory.glob("ref_natinfs.csv"))
+
+    if len(csv_files) == 0:
+        raise FileNotFoundError("Le fichier ref_natinfs.csv est introuvable")
+
+    csv_file = csv_files[0]
+    return csv_file
+
+@task
+def extract_ref_natinfs(filePath):
+    return pd.read_csv(
+        filePath,
+        keep_default_na=False,
+        na_values=[""],
+        sep=";",
+        encoding="latin1"
+    )
+
+
+@task
+def load_ref_natinfs(ref_natinfs: pd.DataFrame):
+    ref_natinfs = ref_natinfs.rename(columns = 
+        {"Numéro NATINF": "id",
+        "Nature de l'infraction": "nature", 
+        "Qualification de l'infraction": "qualification", 
+        "Définie par":"defined_by", 
+        "Réprimée par": "repressed_by"})
+    
+    load(
+        ref_natinfs,
+        table_name="ref_natinfs",
+        schema="public",
+        db_name="monitorenv_remote",
+        how="replace",
+        logger=get_run_logger(),
+    )
+
+
+@flow(name="MonitorEnv - Ref Natinfs")
+def ref_natinfs_flow():
+    csv_filepath = get_csv_file(LIBRARY_LOCATION / "data/")
+    ref_natinfs = extract_ref_natinfs(csv_filepath)
+    load_ref_natinfs(ref_natinfs)

pipeline/tests/mocks.py

@@ -95,3 +95,7 @@ def mock_get_xsd_file(dummy: str):
 @task
 def mock_delete_files(xml_files: List[Path]) -> None:
     pass
+
+@task
+def mock_get_csv_file(dummy: str):
+    return Path(TEST_DATA_LOCATION / "ref_natinfs_csv" / "ref_natinfs.csv")

pipeline/tests/test_data/ref_natinfs_csv/ref_natinfs.csv

@@ -0,0 +1,7 @@
+Numéro NATINF;Nature de l'infraction;Qualification de l'infraction;Définie par;Réprimée par
+7;Délit;RETRAIT DE LA PROVISION D'UN CHEQUE AVEC L'INTENTION DE PORTER ATTEINTE AUX DROITS D'AUTRUI;ART.L.163-2 AL.1 C.M.F.;ART.L.163-2 AL.1, ART.L.163-6 AL.1, AL.2 C.M.F.
+8;Délit;OPPOSITION AU PAIEMENT D'UN CHEQUE AVEC L'INTENTION DE PORTER ATTEINTE AUX DROITS D'AUTRUI;ART.L.163-2 AL.1 C.M.F.;ART.L.163-2 AL.1, ART.L.163-6 AL.1, AL.2 C.M.F. ART.131-30 AL.1 C.PENAL.
+11;Délit;ABANDON DE FAMILLE : NON PAIEMENT D'UNE PENSION OU D'UNE PRESTATION ALIMENTAIRE;ART.227-3 AL.1 C.PENAL.;ART.227-3 AL.1, ART.227-29 C.PENAL.
+13;Délit;INFRACTION A UNE INTERDICTION DE SEJOUR : FREQUENTATION D'UN LIEU INTERDIT;ART.434-38 AL.1, ART.131-31 AL.1 C.PENAL.;ART.434-38 AL.1, ART.434-44 AL.4 C.PENAL.
+16;Délit;AIDE A L'ENTREE, A LA CIRCULATION OU AU SEJOUR IRREGULIERS D'UN ETRANGER EN FRANCE;ART.L.823-1, ART.L.820-1 C.E.S.E.D.A.;ART.L.823-1 AL.1, ART.L.823-4, ART.L.823-6 C.E.S.E.D.A.
+19;Contravention de 5ème classe;NON PAIEMENT D'UNE PENSION ALIMENTAIRE DUE AU CREANCIER, PAR LE TIERS DEBITEUR TENU AU PAIEMENT DIRECT;ART.R.213-5, ART.L.213-1 C.P.C.EX.;ART.R.213-5 C.P.C.EX.

pipeline/tests/test_flows/test_ref_natinfs.py

@@ -0,0 +1,22 @@
+from unittest.mock import patch
+
+import pandas as pd
+
+
+from src.flows.ref_natinfs import (
+    ref_natinfs_flow,
+)
+from src.read_query import read_query
+from tests.mocks import (
+    mock_get_csv_file,
+)
+
+
+@patch("src.flows.ref_natinfs.get_csv_file", mock_get_csv_file)
+def test_flow_ref_natinfs(create_cacem_tables, reset_test_data):
+    state = ref_natinfs_flow(return_state=True)
+    assert state.is_completed()
+
+    ref_natinfs = read_query(query = "SELECT * FROM ref_natinfs", db="monitorenv_remote")
+
+    assert len(ref_natinfs) == 6