45 lines
1.3 KiB
Python
45 lines
1.3 KiB
Python
import time
|
|
from contextlib import contextmanager
|
|
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
from sqlalchemy.sql import text
|
|
|
|
from app.core._config import MYSQL_URI, MYSQL_DATABASE
|
|
from app.core.logger import make_log
|
|
|
|
engine = create_engine(MYSQL_URI) #, echo=True)
|
|
Session = sessionmaker(bind=engine)
|
|
|
|
|
|
database_initialized = False
|
|
while not database_initialized:
|
|
try:
|
|
with Session() as session:
|
|
databases_list = session.execute(text("SHOW DATABASES;"))
|
|
databases_list = [row[0] for row in databases_list]
|
|
make_log("SQL", 'Database list: ' + str(databases_list), level='debug')
|
|
assert MYSQL_DATABASE in databases_list, 'Database not found'
|
|
database_initialized = True
|
|
except Exception as e:
|
|
make_log("SQL", 'MariaDB is not ready yet: ' + str(e), level='debug')
|
|
time.sleep(1)
|
|
|
|
engine = create_engine(f"{MYSQL_URI}/{MYSQL_DATABASE}")
|
|
Session = sessionmaker(bind=engine)
|
|
|
|
|
|
@contextmanager
|
|
def db_session(auto_commit=False):
|
|
_session = Session()
|
|
try:
|
|
yield _session
|
|
if auto_commit is True:
|
|
_session.commit()
|
|
except BaseException as e:
|
|
_session.rollback()
|
|
raise e
|
|
finally:
|
|
_session.close()
|
|
|