|
1 | 1 | from typing import Callable |
2 | | -from datetime import date, timedelta |
| 2 | +from datetime import date, datetime, timedelta |
3 | 3 | import pyarrow |
4 | 4 | import pyarrow.parquet as pq |
5 | 5 | import pyarrow.dataset as pd |
@@ -83,8 +83,11 @@ def create_tableau_parquet(self, partition_template: str) -> bool: |
83 | 83 | ------- |
84 | 84 | True if parquet created, False otherwise |
85 | 85 | """ |
| 86 | + process_logger = ProcessLogger("filtered_hyper_create_parquet") |
86 | 87 | if isinstance(self.start_date, int): |
87 | | - self.end_date = date.today() |
| 88 | + end_datetime = datetime.now() |
| 89 | + process_logger.add_metadata(now=end_datetime) |
| 90 | + self.end_date = end_datetime.date() |
88 | 91 | self.start_date = self.end_date - timedelta(days=self.start_date) |
89 | 92 |
|
90 | 93 | if self.start_date is not None and self.end_date is not None: |
@@ -112,9 +115,7 @@ def create_tableau_parquet(self, partition_template: str) -> bool: |
112 | 115 | format="parquet", |
113 | 116 | filesystem=S3FileSystem(), |
114 | 117 | ) |
115 | | - process_logger = ProcessLogger( |
116 | | - "filtered_hyper_create_parquet", start_date=self.start_date, end_date=self.end_date |
117 | | - ) |
| 118 | + process_logger.add_metadata(start_date=self.start_date, end_date=self.end_date) |
118 | 119 | process_logger.log_start() |
119 | 120 | if len(ds_paths) == 0: |
120 | 121 | process_logger.add_metadata(n_paths_zero=len(ds_paths)) |
|
0 commit comments