41 lines
1.0 KiB
Python
41 lines
1.0 KiB
Python
import inspect
|
|
import os
|
|
from config import Config
|
|
from contextlib import contextmanager
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker, scoped_session
|
|
from typing import Generator
|
|
|
|
from log import log
|
|
|
|
MYSQL_CONNECT = os.environ.get("MM_DB")
|
|
if MYSQL_CONNECT is None:
|
|
raise ValueError("MYSQL_CONNECT is undefined")
|
|
else:
|
|
dbname = MYSQL_CONNECT.split("/")[-1]
|
|
log.debug(f"Database: {dbname}")
|
|
|
|
engine = create_engine(
|
|
MYSQL_CONNECT,
|
|
echo=Config.DISPLAY_SQL,
|
|
pool_pre_ping=True,
|
|
future=True,
|
|
connect_args={"charset": "utf8mb4"},
|
|
)
|
|
|
|
|
|
@contextmanager
|
|
def Session() -> Generator[scoped_session, None, None]:
|
|
frame = inspect.stack()[2]
|
|
file = frame.filename
|
|
function = frame.function
|
|
lineno = frame.lineno
|
|
Session = scoped_session(sessionmaker(bind=engine))
|
|
log.debug(
|
|
f"Session acquired: {file}:{function}:{lineno} " f"[{hex(id(Session))}]"
|
|
)
|
|
yield Session
|
|
log.debug(f" Session released [{hex(id(Session))}]")
|
|
Session.commit()
|
|
Session.close()
|