in jobs/kpi-forecasting/kpi_forecasting/metric_hub.py [0:0]
def __post_init__(self) -> None:
self.start_date = pd.to_datetime(self.start_date).date()
self.end_date = pd.to_datetime(parse_end_date(self.end_date)).date()
# Set useful attributes based on the Metric Hub definition
metric = ConfigLoader.get_metric(
metric_slug=self.slug,
app_name=self.app_name,
)
self.metric = metric
self.alias = self.alias or metric.name
self.submission_date_column = metric.data_source.submission_date_column
# Modify the metric source table string so that it formats nicely in the query.
self.from_expression = self.metric.data_source._from_expr.replace(
"\n", "\n" + " " * 19
)
# Add query snippets for segments
self.segment_select_query = ""
self.segment_groupby_query = ""
if self.segments:
segment_select_query = []
segments = dict(self.segments)
for alias, sql in segments.items():
segment_select_query.append(f" {sql} AS {alias},")
self.segment_select_query = "," + "\n ".join(
segment_select_query
)
self.segment_groupby_query = "," + "\n ,".join(
self.segments.keys()
)
self.where = f"AND {self.where}" if self.where else ""