propusk/module/WorkWithDB.py

from sqlalchemy import (MetaData, Table, Column, Integer,
                        DateTime, String, Text, ForeignKey,
                        create_engine, func)
from sqlalchemy.exc import ProgrammingError, OperationalError
import os
from module.MyMessageBox import show_dialog
from PySide6.QtWidgets import QMessageBox
from logger import logger

meta = MetaData()
cam_setting = Table("сam_setting", meta,
                    Column('id', Integer, primary_key=True),
                    Column('type', Integer, nullable=False),
                    Column('mode', String, nullable=False),
                    Column("selected_cam", String, nullable=False),
                    Column('created', DateTime, default=func.now()),
                    Column('update', DateTime,
                           onupdate=func.current_timestamp())
                    )

list_personal = Table("list_personal", meta,
                      Column('id', Integer, primary_key=True),
                      Column('lastname', String, nullable=False),
                      Column('firstname', String, nullable=False),
                      Column('middlename', String, nullable=False),
                      Column('created', DateTime, default=func.now()),
                      Column('update', DateTime,
                             onupdate=func.current_timestamp())
                      )

list_place = Table("list_place", meta,
                   Column('id', Integer, primary_key=True),
                   Column('name_place', String, nullable=False),
                   Column('created', DateTime, default=func.now()),
                   Column('update', DateTime, onupdate=func.current_timestamp())
                   )

list_propusk = Table("list_propusk", meta,
                     Column("id", Integer, primary_key=True),
                     Column("id_propusk", Integer, nullable=False),
                     Column("date_from", DateTime, nullable=False),
                     Column("date_to", DateTime, nullable=False),
                     Column("personal", Integer, ForeignKey(
                         "list_personal.id"), nullable=False),
                     Column("place", Integer, ForeignKey(
                         "list_place.id"), nullable=False),
                     Column("receiving_man", Text, nullable=False),
                     Column("purpose_visite", Text, nullable=False),
                     Column("face", Text, nullable=False),
                     Column("document", Text, nullable=False),
                     Column("created", DateTime, default=func.now()),
                     Column("update", DateTime, default=func.now(),
                            onupdate=func.current_timestamp()))

list_ussued_passes = Table("list_ussued_passes", meta,
                           Column("id", Integer, primary_key=True),
                           Column("used_pass", Integer, nullable=False),
                           Column("id_propusk", Integer, ForeignKey(
                               "list_propusk.id_propusk"), nullable=False),
                           Column("created", DateTime, default=func.now()),
                           Column("update", DateTime, default=func.now(), onupdate=func.current_timestamp()))

def checking_path_db() -> str | None:
    if os.environ.get("DB_DIR"):
        return os.path.join(os.environ.get("DB_DIR"), "propusk.db")
    else:
        if os.environ.get("DEFAULT_PATH"):
            return os.path.join(
                os.environ.get("DEFAULT_PATH"), "propusk.db")
        else:
            logger.error(
                "Не правильно указан путь к базе данных, или вообще отсутствует")
            # show_dialog(
            #     QMessageBox.Icon.Critical,
            #     "Путь к бд",
            #     "Не правильно указан путь к базе данных, или вообще отсутствует"
            # )


engine = create_engine(F"sqlite:///{checking_path_db()}", echo=False)
engine.logging_name = 'PropuskLogger'



def init_db():
    path = checking_path_db()
    if not os.path.exists(os.path.dirname(path)):
        os.mkdir(os.path.dirname(path))

    meta.create_all(engine)


def connect():
    return engine.connect()


def check_error_sql(func):
    def wrapper(*arg, **args):
        try:
            return func(*arg, **args)
        except ProgrammingError as pe:
            show_dialog(QMessageBox.Icon.Critical, 'Ошибка', 'Произошла ошибка в работе БД')
            logger.error(pe)
        except OperationalError as oe:
            show_dialog(QMessageBox.Icon.Critical, 'Ошибка', 'Произошла ошибка в работе БД')
            logger.error(oe)

    return wrapper