Python - How to check if a specific function is calling your current function, and use that in an if-CodePudding

I have the below method that runs and executes a query returning ids:

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)

now I'm wondering if there's a way to make the sql query to be more dynamic within this method and split it out into 4 different queries with an if statement, that if my other method get_response_2021 is calling _get_ids_from_snowflake it uses the sql1 statement:

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

method get_response_2022 is calling _get_ids_from_snowflake it uses the sql2 statement:

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

Also, if this isn't possible or doesn't make sense please tell me why I'd like to understand.

CodePudding user response：

It is possible using the inspect module, but it is not the ideal or recommended approach.

You are better off using a parameter for the method which will work as a filler for the query itself, then each calling function can pass a different filler as an argument.

For example:

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)

then each of the calling functions can look something like this:

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

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

Just in case you need to do it the way you described this would be an example:

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()