66from __future__ import annotations
77
88import os
9+ from dataclasses import dataclass
910from typing import Any
1011
1112
12- def config ():
13- """Return a new Config object"""
14-
15- class Config :
16- etl_columns : list [str ] = [
17- os .getenv ("ETL_LOAD_SRC_COL" , "load_src" ),
18- os .getenv ("ETL_LOAD_ID_COL" , "load_id" ),
19- os .getenv ("ETL_LOAD_DATE_COL" , "load_date" ),
20- os .getenv ("ETL_UPDT_LOAD_SRC_COL" , "updt_load_src" ),
21- os .getenv ("ETL_UPDT_LOAD_ID_COL" , "updt_load_id" ),
22- os .getenv ("ETL_UPDT_LOAD_DATE_COL" , "updt_load_date" ),
23- ]
24-
25- scd1_soft_delete_columns : list [str ] = [
26- os .getenv ("SCD1_SOFT_DELETE_COL" , "delete_f" )
27- ] + etl_columns
28-
29- scd2_columns : list [str ] = [
30- os .getenv ("SCD2_START_DT_COL" , "start_date" ),
31- os .getenv ("SCD2_END_DT_COL" , "end_date" ),
32- os .getenv ("SCD2_DELETE_COL" , "delete_f" ),
33- ] + etl_columns
34-
35- @classmethod
36- def remove_sys_cols (cls , columns : list [str ]):
37- return [col for col in columns if col not in cls .scd2_columns ]
38-
39- @classmethod
40- def export (cls , template_type : str | None = None ) -> dict [str , Any ]:
41- template_type = template_type or 'NOT_SET'
42- etl_vars : dict [str , Any ] = {}
43- if template_type == 'etl' :
44- etl_vars : dict [str , Any ] = {
45- "etl_columns" : cls .etl_columns ,
46- "scd1_soft_delete_columns" : cls .scd1_soft_delete_columns ,
47- "scd2_columns" : cls .scd2_columns ,
48- "only_main" : False ,
49- }
50-
51- return {"only_main" : False } | etl_vars
52-
53- return Config
13+ @dataclass
14+ class Config :
15+ etl_columns : list [str ]
16+ scd1_soft_delete_columns : list [str ]
17+ scd2_columns : list [str ]
18+
19+ def remove_sys_cols (self , columns : list [str ]):
20+ return [col for col in columns if col not in self .scd2_columns ]
21+
22+ def export (self , template_type : str | None = None ) -> dict [str , Any ]:
23+ template_type = template_type or 'NOT_SET'
24+ etl_vars : dict [str , Any ] = {}
25+ if template_type == 'etl' :
26+ etl_vars : dict [str , Any ] = {
27+ "etl_columns" : self .etl_columns ,
28+ "scd1_soft_delete_columns" : self .scd1_soft_delete_columns ,
29+ "scd2_columns" : self .scd2_columns ,
30+ "only_main" : False ,
31+ }
32+
33+ return {"only_main" : False } | etl_vars
34+
35+
36+ def config () -> Config :
37+ """Return a Config dataclass object"""
38+
39+ etl_columns : list [str ] = [
40+ os .getenv ("ETL_LOAD_SRC_COL" , "load_src" ),
41+ os .getenv ("ETL_LOAD_ID_COL" , "load_id" ),
42+ os .getenv ("ETL_LOAD_DATE_COL" , "load_date" ),
43+ os .getenv ("ETL_UPDT_LOAD_SRC_COL" , "updt_load_src" ),
44+ os .getenv ("ETL_UPDT_LOAD_ID_COL" , "updt_load_id" ),
45+ os .getenv ("ETL_UPDT_LOAD_DATE_COL" , "updt_load_date" ),
46+ ]
47+
48+ scd1_soft_delete_columns : list [str ] = [
49+ os .getenv ("SCD1_SOFT_DELETE_COL" , "delete_f" )
50+ ] + etl_columns
51+
52+ scd2_columns : list [str ] = [
53+ os .getenv ("SCD2_START_DT_COL" , "start_date" ),
54+ os .getenv ("SCD2_END_DT_COL" , "end_date" ),
55+ os .getenv ("SCD2_DELETE_COL" , "delete_f" ),
56+ ] + etl_columns
57+
58+ return Config (
59+ etl_columns = etl_columns ,
60+ scd1_soft_delete_columns = scd1_soft_delete_columns ,
61+ scd2_columns = scd2_columns ,
62+ )
0 commit comments