Python

Question

我有以下方法运行并执行返回 ID 的查询：

def _get_ids_from_snowflake(self):
   sf_hook = SnowflakeHook(warehouse)

   logging.info(f'Query fetching ids...')

   sql = f"""
         SELECT simple_id, id FROM my_table;
         """

  logging.info(f'Query finished!')

  return sf_hook.execute_query(queries=sql, fetch_all=True)

现在我想知道是否有办法使 sql 查询在此方法中更加动态，并使用 if 语句将其拆分为 4 个不同的查询，如果我的其他方法 get_response_2021 正在调用 _get_ids_from_snowflake 它使用 sql1 语句：

sql1 = f"""
             SELECT simple_id, id FROM my_table where YEAR(id_date) = 2021;
             """

方法 get_response_2022 正在调用 _get_ids_from_snowflake 它使用 sql2 语句：

sql2 =  f"""
                 SELECT simple_id, id FROM my_table where YEAR(id_date) = 2022;
                 """

此外，如果这不可能或没有意义，请告诉我为什么我想了解。

Answer 1

可以使用 inspect 模块，但这不是理想或推荐的方法。

最好为方法使用一个参数，作为查询本身的填充物，然后每个调用函数都可以传递不同的填充物作为参数。

例如：

def _get_ids_from_snowflake(self, year=None):
  sf_hook = SnowflakeHook(warehouse)

  if year:
    sql = f"SELECT simple_id, id FROM my_table where YEAR(id_date) = {year};"
  else:
    sql = "SELECT simple_id, id FROM my_table;"

  logging.info('Query fetching ids...')
  logging.info('Query finished!')
  return sf_hook.execute_query(queries=sql, fetch_all=True)

那么每个调用函数看起来像这样：

def get_response_2021(self):
    self._get_ids_from_snowflake(year='2021')

def get_response_2022(self):
    self._get_ids_from_snowflake(year='2022')

以防万一您需要按照您描述的方式进行操作，这就是一个示例：

import inspect

class Klass:
    def _get_ids_from_snowflake(self):
        frame = inspect.stack()[1]
        print(frame.function)  # prints function name of the caller

        if frame.function == "get_response_2022":
           sql = f"SELECT simple_id, id FROM my_table where YEAR(id_date) = 2022;"

        elif frame.function == "get_response_2021":
            ...
        ...
        ...

        logging.info('Query fetching ids...')
        logging.info('Query finished!')
        return sf_hook.execute_query(queries=sql, fetch_all=True)


    def get_response_2021(self):
        self._get_ids_from_snowflake()  # prints get_response_2021

    def get_response_2022(self):
        self._get_ids_from_snowflake()  # prints get_response_2022

a = Klass()
a.get_response_2021()
a.get_response_2022()

Python - 如何检查特定函数是否正在调用您当前的函数，并在当前函数的 if/else 中使用它？

Python - How to check if a specific function is calling your current function, and use that in an if/else within the current function?

python-3.x