67 lines
2.1 KiB
Python
67 lines
2.1 KiB
Python
import os
|
|
from alembic.config import Config
|
|
from alembic.script import ScriptDirectory
|
|
from alembic.runtime.migration import MigrationContext
|
|
from alembic import command
|
|
from sqlalchemy import create_engine
|
|
from alembic import command
|
|
from dotenv import load_dotenv
|
|
import logging
|
|
|
|
#Logging
|
|
|
|
logging.basicConfig(
|
|
filename='app.log', # Log to a file
|
|
level=logging.DEBUG, # Log everything (DEBUG and above)
|
|
format='%(asctime)s %(levelname)s %(message)s'
|
|
)
|
|
|
|
logging.info('Application started')
|
|
try:
|
|
logging.debug('Debug info here')
|
|
except Exception as e:
|
|
logging.exception('An error occurred')
|
|
|
|
|
|
load_dotenv()
|
|
db_name = os.getenv("DB_NAME")
|
|
db_psswd = os.getenv("DB_PASSWORD")
|
|
db_user = os.getenv("DB_USER")
|
|
db_port = os.getenv("DB_PORT") or "3306"
|
|
db_host = os.getenv("DB_HOST")
|
|
if db_host == None:
|
|
os._exit(1)
|
|
engine_url = f"mariadb+mariadbconnector://{db_user}:{db_psswd}@{db_host}:{db_port}/{db_name}"
|
|
|
|
base_dir = os.path.dirname(__file__)
|
|
ini_path = os.path.join(base_dir, "..", "alembic.ini")
|
|
alembic_cfg = Config(ini_path)
|
|
alembic_cfg.set_section_option(alembic_cfg.config_ini_section, "sqlalchemy.url", engine_url)
|
|
script = ScriptDirectory.from_config(alembic_cfg)
|
|
|
|
engine = create_engine(engine_url)
|
|
|
|
def check_db():
|
|
with engine.connect() as conn:
|
|
context = MigrationContext.configure(conn)
|
|
current_rev = context.get_current_revision()
|
|
head_rev = script.get_current_head()
|
|
if current_rev != head_rev:
|
|
command.upgrade(alembic_cfg, "head")
|
|
return False
|
|
|
|
def auto_lookup():
|
|
command.revision(alembic_cfg, message="add tables and rows", autogenerate=True)
|
|
script = ScriptDirectory.from_config(alembic_cfg)
|
|
head_rev = script.get_revision(script.get_current_head())
|
|
with open(head_rev.path) as f:
|
|
content = f.read()
|
|
if "pass\n # ### end Alembic commands ###" in content:
|
|
os.remove(head_rev.path)
|
|
command.stamp(alembic_cfg, "head")
|
|
logging.info("No changes DB detected")
|
|
else:
|
|
command.upgrade(alembic_cfg, "head")
|
|
logging.info("New DB changes, upgrade table")
|
|
|
|
auto_lookup() |