Skip to content

scripts

BaseEtl

Bases: AbstractEtl

This class is used to create an ETL object.

Source code in pypdtools\core\scripts.py 
 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
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
class BaseEtl(AbstractEtl):
    """This class is used to create an ETL object."""

    def __init__(self) -> None:
        """This method is used to initialize the ETL object."""
        super().__init__()
        self.data = None
        self.executed = False

    def extract(self):
        """This method is used to get data from the source."""
        raise NotImplementedError

    def transform(self):
        """This method is used to transform the data."""
        raise NotImplementedError

    def load(self):
        """This method is used to load the data."""
        raise NotImplementedError

    def execute(self):
        """This method is used to execute the etl."""
        if self.executed:
            return self

        self.extract()
        self.transform()
        self.load()
        self.executed = True
        return self

    @property
    def data(self) -> pd.DataFrame:
        """This method is used to get the data."""
        self.execute()
        return self._df

    def __len__(self) -> int:
        """This method is used to get the length of the data."""
        self.execute()
        return super().__len__()

    def __iter__(self) -> Iterator:
        """This method is used to iterate over the data."""
        self.execute()
        return super().__iter__()

    def __aiter__(self) -> Iterator:
        """This method is used to async iterate over the data."""
        self.execute()
        return super().__aiter__()

    def __enter__(self) -> pd.DataFrame:
        """This method is used to enter the context."""
        self.execute()
        return super().__enter__()

    async def __aenter__(self) -> pd.DataFrame:
        """This method is used to async enter the context."""
        self.execute()
        return super().__aenter__()

    def __add__(self, other) -> PtDataFrame:
        """This method is used to add two dataframes."""
        self.execute()
        other.execute()
        return super().__add__(other)

    def __radd__(self, other) -> PtDataFrame:
        """This method is used to add two dataframes."""
        self.execute()
        other.execute()
        return super().__radd__(other)

    def col_to_list(self, col_name, drop_duplicates=True) -> List:
        """This method is used to get a column as a list."""
        self.execute()
        return super().col_to_list(col_name, drop_duplicates)

data: pd.DataFrame property

This method is used to get the data.

__add__(other)

This method is used to add two dataframes.

Source code in pypdtools\core\scripts.py 
72
73
74
75
76
def __add__(self, other) -> PtDataFrame:
    """This method is used to add two dataframes."""
    self.execute()
    other.execute()
    return super().__add__(other)

__aenter__() async

This method is used to async enter the context.

Source code in pypdtools\core\scripts.py 
67
68
69
70
async def __aenter__(self) -> pd.DataFrame:
    """This method is used to async enter the context."""
    self.execute()
    return super().__aenter__()

__aiter__()

This method is used to async iterate over the data.

Source code in pypdtools\core\scripts.py 
57
58
59
60
def __aiter__(self) -> Iterator:
    """This method is used to async iterate over the data."""
    self.execute()
    return super().__aiter__()

__enter__()

This method is used to enter the context.

Source code in pypdtools\core\scripts.py 
62
63
64
65
def __enter__(self) -> pd.DataFrame:
    """This method is used to enter the context."""
    self.execute()
    return super().__enter__()

__init__()

This method is used to initialize the ETL object.

Source code in pypdtools\core\scripts.py 
12
13
14
15
16
def __init__(self) -> None:
    """This method is used to initialize the ETL object."""
    super().__init__()
    self.data = None
    self.executed = False

__iter__()

This method is used to iterate over the data.

Source code in pypdtools\core\scripts.py 
52
53
54
55
def __iter__(self) -> Iterator:
    """This method is used to iterate over the data."""
    self.execute()
    return super().__iter__()

__len__()

This method is used to get the length of the data.

Source code in pypdtools\core\scripts.py 
47
48
49
50
def __len__(self) -> int:
    """This method is used to get the length of the data."""
    self.execute()
    return super().__len__()

__radd__(other)

This method is used to add two dataframes.

Source code in pypdtools\core\scripts.py 
78
79
80
81
82
def __radd__(self, other) -> PtDataFrame:
    """This method is used to add two dataframes."""
    self.execute()
    other.execute()
    return super().__radd__(other)

col_to_list(col_name, drop_duplicates=True)

This method is used to get a column as a list.

Source code in pypdtools\core\scripts.py 
84
85
86
87
def col_to_list(self, col_name, drop_duplicates=True) -> List:
    """This method is used to get a column as a list."""
    self.execute()
    return super().col_to_list(col_name, drop_duplicates)

execute()

This method is used to execute the etl.

Source code in pypdtools\core\scripts.py 
30
31
32
33
34
35
36
37
38
39
def execute(self):
    """This method is used to execute the etl."""
    if self.executed:
        return self

    self.extract()
    self.transform()
    self.load()
    self.executed = True
    return self

extract()

This method is used to get data from the source.

Source code in pypdtools\core\scripts.py 
18
19
20
def extract(self):
    """This method is used to get data from the source."""
    raise NotImplementedError

load()

This method is used to load the data.

Source code in pypdtools\core\scripts.py 
26
27
28
def load(self):
    """This method is used to load the data."""
    raise NotImplementedError

transform()

This method is used to transform the data.

Source code in pypdtools\core\scripts.py 
22
23
24
def transform(self):
    """This method is used to transform the data."""
    raise NotImplementedError