Я хотел бы использовать QueryChecker.check_query
декоратор, чтобы определить, есть ли у них старые имена таблиц, которые были заменены новыми.
class QueryChecker:
@staticmethod
def tables_in_query(sql_str):
# remove the /* */ comments
q = re.sub(r"/\*[^*]*\*+(?:[^*/][^*]*\*+)*/", "", sql_str)
# remove whole line -- and # comments
lines = [line for line in q.splitlines() if not re.match("^\s*(--|#)", line)]
# remove trailing -- and # comments
q = " ".join([re.split("--|#", line)[0] for line in lines])
# split on blanks, parens and semicolons
tokens = re.split(r"[\s)(;]+", q)
# scan the tokens. if we see a FROM or JOIN, we set the get_next
# flag, and grab the next one (unless it's SELECT).
result = set()
get_next = False
for tok in tokens:
if get_next:
if tok.lower() not in ["", "select"]:
result.add(tok)
get_next = False
get_next = tok.lower() in ["from", "join"]
return result
@staticmethod
def check_tables(tables):
if tables == set():
return
prefixes = {
'old_table': 'new_table',
'second_old_table': 'second_new_table'
}
for table in tables:
# remove db name if exist
table = table.split('.')[-1]
for prefix in prefixes.keys():
if table.startswith(prefix):
print(f"!! {table} to {table.replace(prefix, prefixes[prefix])} !!")
def check_query(self, run_query):
def wrapper(new_self, query) :
tables = self.tables_in_query(query)
print(f'Tables being queried: {tables}')
self.check_tables(tables)
new_self.run_query( query)
return wrapper
class Palim:
@QueryChecker.check_query
def run_query(self, query):
print(query)
palim = Palim()
palim.run_query("""
SELECT
*
FROM database
""")
TypeError: check_query () отсутствует 1 обязательный позиционный аргумент : 'run_query'
У меня он работал вне класса, но я хотел бы, чтобы он работал внутри класса, чтобы сигнализировать, что эти функции принадлежат друг другу.
class Palim
is отличается в кодовой базе и запускает запрос, здесь это простая печать для воспроизводимости.