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

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查询在这个方法中更加动态，并将其分割成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()