33 lines
No EOL
744 B
Python
33 lines
No EOL
744 B
Python
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
import config
|
|
from .sql_generated import *
|
|
import configparser
|
|
|
|
alembic = configparser.ConfigParser()
|
|
alembic.read("alembic.ini")
|
|
|
|
try:
|
|
dburl = alembic.get("alembic", "sqlalchemy.url")
|
|
except:
|
|
dburl = config.DBURL
|
|
|
|
engine = create_engine(
|
|
dburl,
|
|
pool_pre_ping=True,
|
|
pool_recycle=1800
|
|
)
|
|
|
|
ergoEngine = create_engine(
|
|
dburl,
|
|
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() |