Ir para o conteúdo

Workflow

flowchart LR
    subgraph ETL[Pipeline]
        A(Múltiplos Arquivos Excel) --> B[Extract: extract_from_excel]
        B[Extract: extract_from_excel] --> |Gera uma lista de DataFrames| C[Transformation: concatenate_data_frames]
        C[Transformation: concatenate_data_frames] -->|Gera um DataFrame Consolidado| D[Load: Converte para Excel]
        D(Load: Converte para Excel) --> |Salva o consolidado em Excel| E(Pasta Output: Um arquivo único Excel)
    end

Função de extração de dados

Função para ler os arquivos de uma pasta data/input e retornar uma lista de DataFrames

args: input_path (str): caminho da pasta com os arquivos

return: lista de DataFrames

Source code in app\pipeline\extract.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
def extract_from_excel(path: str) -> List[pd.DataFrame]:
    """
    Função para ler os arquivos de uma pasta data/input e retornar uma lista de DataFrames

    args: input_path (str): caminho da pasta com os arquivos

    return: lista de DataFrames
    """

    all_files = glob.glob(os.path.join(path, '*.xlsx'))

    data_frame_list = []

    for file in all_files:
        data_frame_list.append(pd.read_excel(file))

    return data_frame_list

Função de transformação de dados

Função para transformar uma lista de DataFrames em um único DataFrame

Source code in app\pipeline\transform.py 
 6
 7
 8
 9
10
11
12
13
def concatenate_data_frames(
    data_frame_list: List[pd.DataFrame],
) -> pd.DataFrame:
    """
    Função para transformar uma lista de DataFrames em um único DataFrame
    """

    return pd.concat(data_frame_list, ignore_index=True)

Função de carga de dados

Recebendo um DataFrame e salvando-o como arquivo Excel

args: data_frame (pd.DataFrame): DataFrame a ser salvo como Excel output_path (str): caminho onde o arquivo será salvo file_name (str): nome do arquivo a ser salvo

return "Arquivo salvo com sucesso. "

Source code in app\pipeline\load.py 
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
def load_excel(
    data_frame: pd.DataFrame, output_path: str, file_name: str
) -> str:
    """
    Recebendo um DataFrame e salvando-o como arquivo Excel

    args:
    data_frame (pd.DataFrame): DataFrame a ser salvo como Excel
    output_path (str): caminho onde o arquivo será salvo
    file_name (str): nome do arquivo a ser salvo

    return "Arquivo salvo com sucesso. "
    """

    if not os.path.exists(output_path):
        os.makedirs(output_path)

    data_frame.to_excel(f'{output_path}/{file_name}.xlsx', index=False)
    return 'Arquivo salvo com sucesso. '