ETL_Project/transform.py at main · El-Manaa/ETL_Project · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
from polars import col, LazyFrame, UInt32, UInt64, nth
from extract import read_data
from os import getenv
from dotenv import load_dotenv
load_dotenv()


def transform_Date_df(df: LazyFrame) -> LazyFrame:
    df_Dim_Date = df.with_columns(
        D_Id=col("D_Date").dt.strftime("%Y%m%d").cast(UInt32),
        D_Jour=col("D_Date").dt.day(),
        D_Mois=col("D_Date").dt.month(),
        D_Année=col("D_Date").dt.year(),
    )
    return df_Dim_Date.select(nth(1, 2, 3, 4, 0))


def transform_Client_df(df: LazyFrame) -> LazyFrame:
    return df.with_columns(
        col("Cli_Id").cast(UInt32)
    )


def transform_Article_df(df: LazyFrame) -> LazyFrame:
    return df.with_columns(
        col("Art_Id").cast(UInt32)
    )


def transform_Livreur_df(df: LazyFrame) -> LazyFrame:
    return df.with_columns(
        col("Liv_Id").cast(UInt32)
    )


def transform_FctTab(lbl: LazyFrame, bl: LazyFrame) -> LazyFrame:
    return lbl.filter(
        col("Art_Id").is_in(
            read_data(getenv("DEST_DB") + ".dbo.Dim_Article",
                      ("Art_Id",)).collect().to_series()
        )
    ).join(
        bl.filter(
            col("Cli_Id").is_in(
                read_data(getenv("DEST_DB") + ".dbo.Dim_Client",
                          ("Cli_Id",)).collect()
            )
            & col("Liv_Id").is_in(
                read_data(getenv("DEST_DB") + ".dbo.Dim_Livreur",
                          ("Liv_Id",)).collect()
            )
        ),
        right_on=col("BLiv_Id"),
        left_on=col("BLiv_Id")
    ).with_columns(
        col("BLiv_DateCreation").dt.strftime(
            "%Y%m%d").alias("D_Id").cast(UInt32),
        col("BLiv_Id").cast(UInt64),
        col("Cli_Id").cast(UInt64),
        col("Art_Id").cast(UInt32),
        col("Liv_Id").cast(UInt32),
        col("LBLiv_Id").cast(UInt32)
    ).drop("BLiv_DateCreation").filter(
        col("D_Id").is_in(
            read_data(getenv("DEST_DB") + ".dbo.Dim_Date", ("D_Id",)).collect()
        )
    ).select(
        read_data(getenv("DEST_DB") +
                  ".dbo.Fct_BonLivraison").collect_schema().names()
    )