import base64 import duckdb class Q(str): UNSAFE = ["CREATE", "DELETE", "DROP", "INSERT", "UPDATE"] def __init__(self, query: str): self.is_safe = not any([cmd in query.upper() for cmd in self.UNSAFE]) def run(self, sql_engine=None): try: if sql_engine is None: return self.run_duckdb() else: return self.run_sql(sql_engine) except Exception as e: pass def run_duckdb(self): return duckdb.sql(self) def df(self, sql_engine=None): result = self.run(sql_engine=sql_engine) if result is None: return result_df = result.df() return result_df @property def base64(self): return base64.b64encode(self.encode()).decode() @classmethod def from_base64(cls, b64): """Initializing from base64-encoded URL paths.""" return cls(base64.b64decode(b64).decode())