16 lines
511 B
Python
16 lines
511 B
Python
|
from typing import Any
|
||
|
|
from sqlalchemy import text
|
||
|
|
from app.db.mysql_router import get_engine_for_company
|
||
|
|
|
||
|
|
def execute_query(company_id: str, sql_query: str) -> list[dict[Any, Any]]:
|
||
|
|
engine = get_engine_for_company(company_id)
|
||
|
|
|
||
|
|
try:
|
||
|
|
with engine.connect() as connection:
|
||
|
|
result = connection.execute(text(sql_query))
|
||
|
|
columns = result.keys()
|
||
|
|
rows = [dict(zip(columns, row)) for row in result]
|
||
|
|
return rows
|
||
|
|
except Exception as e:
|
||
|
|
raise e
|