Staging/sms farmacia estoque refactoring

.gitignore

@@ -5,6 +5,7 @@ replit.nix
 test_local.py
 pylint.txt
 test.py
+test*.ipynb
 test/*
 test/*.ipynb
 test/*.csv

pipelines/rj_sms/__init__.py

@@ -4,4 +4,10 @@
 """
 
 from pipelines.rj_sms.dump_db_sivep.flows import *
-from pipelines.rj_sms.pubsub.flows import *
+from pipelines.rj_sms.dump_api_prontuario_vitacare.flows import *
+from pipelines.rj_sms.dump_api_prontuario_vitai.flows import *
+from pipelines.rj_sms.dump_azureblob_estoque_tpc.flows import *
+from pipelines.rj_sms.dump_ftp_cnes.flows import *
+from pipelines.rj_sms.dump_ftp_sigtap.flows import *
+from pipelines.rj_sms.dump_sheets.flows import *
+from pipelines.rj_sms.materialize_datalake.flows import *

pipelines/rj_sms/dump_api_prontuario_vitacare/constants.py

@@ -0,0 +1,31 @@
+# -*- coding: utf-8 -*-
+# pylint: disable=C0103
+"""
+Constants for utils.
+"""
+from enum import Enum
+
+
+class constants(Enum):
+    """
+    Constant values for the dump vitai flows
+    """
+
+    VAULT_PATH = "estoque_vitai"
+    VAULT_KEY = "token"
+    DATASET_ID = "brutos_prontuario_vitacare"
+    TABLE_POSICAO_ID = "estoque_posicao"
+    TABLE_MOVIMENTOS_ID = "estoque_movimento"
+    ENDPOINT_BASE_URL = [
+        "http://consolidado-ap10.pepvitacare.com:8088",
+        "http://consolidado-ap21.pepvitacare.com:8088",
+        "http://consolidado-ap22.pepvitacare.com:8088",
+        "http://consolidado-ap31.pepvitacare.com:8089",
+        "http://consolidado-ap32.pepvitacare.com:8088",
+        "http://consolidado-ap33.pepvitacare.com:8089",
+        "http://consolidado-ap40.pepvitacare.com:8089",
+        "http://consolidado-ap51.pepvitacare.com:8089",
+        "http://consolidado-ap52.pepvitacare.com:8088",
+        "http://consolidado-ap53.pepvitacare.com:8090"]
+    ENDPOINT_POSICAO = "/reports/pharmacy/stocks"
+    ENDPOINT_MOVIMENTOS = "/reports/pharmacy/movements"

pipelines/rj_sms/dump_api_prontuario_vitacare/flows.py

@@ -0,0 +1,140 @@
+# -*- coding: utf-8 -*-
+from prefect import Parameter
+from prefect.run_configs import KubernetesRun
+from prefect.storage import GCS
+from pipelines.utils.decorators import Flow
+from pipelines.constants import constants
+from pipelines.rj_sms.dump_api_prontuario_vitacare.constants import (
+    constants as vitacare_constants,
+)
+from pipelines.rj_sms.utils import (
+    create_folders,
+    from_json_to_csv,
+    download_from_api,
+    add_load_date_column,
+    create_partitions,
+    upload_to_datalake,
+)
+from pipelines.rj_sms.dump_api_prontuario_vitacare.tasks import (
+    build_params,
+    download_multiple_files)
+
+from pipelines.rj_sms.dump_api_prontuario_vitai.schedules import every_day_at_six_am
+
+
+with Flow(
+    name="SMS: Dump VitaCare - Captura Posição de Estoque", code_owners=["thiago"]
+) as dump_vitacare_posicao:
+    # Set Parameters
+    #  Vault
+    vault_path = vitacare_constants.VAULT_PATH.value
+    vault_key = vitacare_constants.VAULT_KEY.value
+    #  GCP
+    dataset_id = vitacare_constants.DATASET_ID.value
+    table_id = vitacare_constants.TABLE_POSICAO_ID.value
+
+    # Vitacare API
+    endpoint_base_url = vitacare_constants.ENDPOINT_BASE_URL.value
+    endpoint_posicao = vitacare_constants.ENDPOINT_POSICAO.value
+    date = Parameter("date", default="today")
+
+    # Start run
+    create_folders_task = create_folders()
+
+    build_params_task = build_params(date_param=date)
+    build_params_task.set_upstream(create_folders_task)
+
+    download_multiple_files_task = download_multiple_files(
+        base_urls=endpoint_base_url,
+        endpoint=endpoint_posicao,
+        params=build_params_task,
+        table_id=table_id,
+        vault_path=vault_path,
+        vault_key=vault_key)
+    download_multiple_files_task.set_upstream(build_params_task)
+
+    create_partitions_task = create_partitions(
+        data_path="./data/raw", partition_directory="./data/partition_directory"
+    )
+    create_partitions_task.set_upstream(download_multiple_files_task)
+
+    upload_to_datalake_task = upload_to_datalake(
+        input_path="./data/partition_directory",
+        dataset_id=dataset_id,
+        table_id=table_id,
+        if_exists="replace",
+        csv_delimiter=";",
+        if_storage_data_exists="replace",
+        biglake_table=True,
+    )
+    upload_to_datalake_task.set_upstream(create_partitions_task)
+
+
+dump_vitacare_posicao.storage = GCS(constants.GCS_FLOWS_BUCKET.value)
+dump_vitacare_posicao.run_config = KubernetesRun(
+    image=constants.DOCKER_IMAGE.value,
+    labels=[
+        constants.RJ_SMS_DEV_AGENT_LABEL.value,
+    ],
+)
+
+dump_vitacare_posicao.schedule = every_day_at_six_am
+
+
+with Flow(
+    name="SMS: Dump VitaCare - Captura Posição de Estoque", code_owners=["thiago"]
+) as dump_vitacare_movimento:
+    # Set Parameters
+    #  Vault
+    vault_path = vitacare_constants.VAULT_PATH.value
+    vault_key = vitacare_constants.VAULT_KEY.value
+    #  GCP
+    dataset_id = vitacare_constants.DATASET_ID.value
+    table_id = vitacare_constants.TABLE_MOVIMENTOS_ID.value
+
+    # Vitacare API
+    endpoint_base_url = vitacare_constants.ENDPOINT_BASE_URL.value
+    endpoint_movimentos = vitacare_constants.ENDPOINT_MOVIMENTOS.value
+    date = Parameter("date", default="yesterday")
+
+    # Start run
+    create_folders_task = create_folders()
+
+    build_params_task = build_params(date_param=date)
+    build_params_task.set_upstream(create_folders_task)
+
+    download_multiple_files_task = download_multiple_files(
+        base_urls=endpoint_base_url,
+        endpoint=endpoint_movimentos,
+        params=build_params_task,
+        table_id=table_id,
+        vault_path=vault_path,
+        vault_key=vault_key)
+    download_multiple_files_task.set_upstream(build_params_task)
+
+    create_partitions_task = create_partitions(
+        data_path="./data/raw", partition_directory="./data/partition_directory"
+    )
+    create_partitions_task.set_upstream(download_multiple_files_task)
+
+    upload_to_datalake_task = upload_to_datalake(
+        input_path="./data/partition_directory",
+        dataset_id=dataset_id,
+        table_id=table_id,
+        if_exists="replace",
+        csv_delimiter=";",
+        if_storage_data_exists="replace",
+        biglake_table=True,
+    )
+    upload_to_datalake_task.set_upstream(create_partitions_task)
+
+
+dump_vitacare_movimento.storage = GCS(constants.GCS_FLOWS_BUCKET.value)
+dump_vitacare_movimento.run_config = KubernetesRun(
+    image=constants.DOCKER_IMAGE.value,
+    labels=[
+        constants.RJ_SMS_DEV_AGENT_LABEL.value,
+    ],
+)
+
+dump_vitacare_movimento.schedule = every_day_at_six_am

pipelines/rj_sms/dump_api_prontuario_vitacare/tasks.py

@@ -0,0 +1,63 @@
+# -*- coding: utf-8 -*-
+"""
+Tasks for dump_api_prontuario_vitacare
+"""
+
+from datetime import date, timedelta
+import re
+from prefect import task
+from pipelines.utils.utils import log
+from pipelines.rj_sms.utils import (
+    from_json_to_csv,
+    download_from_api,
+    add_load_date_column
+)
+
+
+@task
+def build_params(date_param="today"):
+    if date_param == "today":
+        params = {"date": str(date.today())}
+    elif date_param == "yesterday":
+        params = {"date": str(date.today() - timedelta(days=1))}
+    else:
+        {"date": date_param}
+    log(f"Params built: {params}")
+    return params
+
+
+@task
+def download_multiple_files(
+    base_urls: list, endpoint: str, params: dict, table_id:str, vault_path: str, vault_key: str
+):
+
+    pattern = r"ap\d{2}"
+
+    for n, base_url in enumerate(base_urls):
+
+        ap = re.findall(pattern, base_url)[0]
+
+        log(f"Downloading {ap} ({n+1}/{len(base_urls)})")
+
+        download_task = download_from_api.run(
+            url=f"{base_url}{endpoint}",
+            params=params,
+            file_folder="./data/raw",
+            file_name=f"{table_id}-{ap}",
+            vault_path=vault_path,
+            vault_key=vault_key,
+            add_load_date_to_filename=True,
+            load_date=params["date"],
+        )
+
+        #
+        with open(download_task, 'r', encoding="UTF-8") as f:
+            first_line = f.readline().strip()
+        if first_line == '[]':
+            log("The json content is empty.")
+        else:
+            conversion_task = from_json_to_csv.run(input_path=download_task, sep=";")
+
+            add_load_date_column_task = add_load_date_column.run(
+                input_path=conversion_task, sep=";"
+            )

pipelines/rj_sms/dump_api_prontuario_vitai/constants.py

@@ -0,0 +1,18 @@
+# -*- coding: utf-8 -*-
+# pylint: disable=C0103
+"""
+Constants for utils.
+"""
+from enum import Enum
+
+
+class constants(Enum):
+    """
+    Constant values for the dump vitai flows
+    """
+
+    VAULT_PATH = "estoque_vitai"
+    VAULT_KEY = "token"
+    DATASET_ID = "brutos_prontuario_vitai"
+    TABLE_POSICAO_ID = "estoque_posicao"
+    TABLE_MOVIMENTOS_ID = "estoque_movimento"