27 lines
No EOL
708 B
Python
27 lines
No EOL
708 B
Python
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
from .sql_generated import *
|
|
import configparser
|
|
|
|
alembic = configparser.ConfigParser()
|
|
alembic.read("alembic.ini")
|
|
|
|
|
|
engine = create_engine(
|
|
alembic.get("alembic", "sqlalchemy.url"),
|
|
pool_pre_ping=True,
|
|
pool_recycle=1800
|
|
)
|
|
|
|
ergoEngine = create_engine(
|
|
alembic.get("alembic", "sqlalchemy.url"),
|
|
pool_pre_ping=True,
|
|
pool_recycle=1800
|
|
)
|
|
|
|
SessionMaker = sessionmaker(autocommit=False, autoflush=False, bind=engine, )
|
|
|
|
def ergoQueryFetchOne(q: str, **kwargs):
|
|
with ergoEngine.connect() as connection:
|
|
connection.execute(text("use ergo"))
|
|
return connection.execute(text(q), kwargs).fetchone() |