database code

import configparser
import logging

from sqlalchemy import create_engine
from sqlalchemy.exc import DatabaseError
from sqlalchemy.orm import Session

config = configparser.ConfigParser()
config.read("config.ini", encoding='utf-8')

logger = logging.getLogger("main")

USERNAME = config["DB"]["login"]
PASSWORD = config["DB"]["password"]
HOST = config["DB"]["server"]
PORT = config["DB"].getint("port")
DATABASE = config["DB"]["database"]

engine = create_engine(f"mysql://{USERNAME}:{PASSWORD}@{HOST}:{PORT}/{DATABASE}", pool_pre_ping=True, pool_recycle=3600)

session = Session(bind=engine)


def run_stored_procedure(name: str):
    logger.debug(f"A stored procedure {name} has been started")
    try:
        data = session.execute(f"CALL {name}").one()
        return data
    except DatabaseError as e:
        logger.error(f"Stored procedure run error {e}")


def select_all(model: object, params: tuple = None):
    logger.debug(f"SELECT one {model} with params {params}")
    query = session.query(model)
    if params is not None:
        query = query.filter(params)
    return query


def select_one(model: object, id: int):
    logger.debug(f"SELECT one {model} with key {id}")
    data = session.query(model).get(id)
    return data


def insert(data: object):
    logger.debug(f"UPDATE {data}")
    session.add(data)
    try:
        session.commit()
    except DatabaseError as e:
        session.rollback()
        logger.error(f"INSERT error {e}")


def update(data: object):
    logger.debug(f"UPDATE {data}")
    session.add(data)
    try:
        session.commit()
    except DatabaseError as e:
        session.rollback()
        logger.error(f"UPDATE error {e}")


def ping():
    session.execute("SELECT 1")