| 12345678910111213141516171819202122232425262728293031323334 |
- # -*- coding: utf-8 -*-
- """
- @author: yq
- @time: 2024/10/31
- @desc: sql直接统计指标直出
- """
- import pandas as pd
- from commom import get_logger, f_fill_placeholder
- from data import DataLoaderBase
- from entitys import MetricFucEntity
- from .metric_base import MetricBase
- logger = get_logger()
- class MetricBySqlGeneral(MetricBase):
- def __init__(self, sql: str, *args, **kwargs):
- super().__init__(*args, **kwargs)
- self._sql = sql
- def _fill_placeholder(self, sql: str):
- return sql
- def _load_data(self, data_loader: DataLoaderBase, *args, **kwargs) -> pd.DataFrame:
- sql = f_fill_placeholder(self._sql)
- logger.info(f"sql: {sql}")
- data = data_loader.get_data(sql)
- logger.info(f"sql execute result: {data.head(5)}")
- return data
- def calculate(self, *args, **kwargs) -> MetricFucEntity:
- return MetricFucEntity(table=self._load_data(*args, **kwargs))
|
