utils/Database.py - master - Ksq/app_flet

utils/Database.py
48 строк2 KB
KSQ-web
добавление файлов в репозиторий
05 ноя 2024, 13:45
05 ноя 2024, 13:4590ecde1
from sqlalchemy import create_engine, Table, MetaData, select, insert, and_
from sqlalchemy.orm import sessionmaker, declarative_base
from dotenv import load_dotenv
import os


class Database:
    def __init__(self):
        load_dotenv()
        self.db_host = os.getenv('DB_HOST')
        self.db_user = os.getenv('DB_USER')
        self.db_password = os.getenv('DB_PASSWORD')
        self.db_name = os.getenv('DB_NAME')
        self.connect = f'mysql+pymysql://{self.db_user}:{self.db_password}@{self.db_host}/{self.db_name}?charset=utf8mb4'
        self.engine = create_engine(self.connect)
        self.metadata = MetaData()
        self.Base = declarative_base()

        self.adminUser = Table('admin_user', self.metadata, autoload_with=self.engine)
        self.postPending = Table('pending_post', self.metadata, autoload_with=self.engine)

        self.Session = sessionmaker(bind=self.engine)
        self.session = self.Session()

    def check_email(self, email):
        result = self.session.execute(select(self.adminUser).where(self.adminUser.c.email == email))
        return result.fetchone()

    def check_login(self, login):
        result = self.session.execute(select(self.adminUser).where(self.adminUser.c.login == login))
        return result.fetchone()

    def insert_user(self, login, email, password):
        self.session.execute(insert(self.adminUser).values(login=login, email=email, password=password))
        self.session.commit()

    def authorization(self, email, password):
        result = self.session.execute(select(self.adminUser).where(and_(self.adminUser.c.email == email, self.adminUser.c.password == password)))
        return result.fetchone()

    def insert_post(self, message, path_img, link_post, time):
        self.session.execute(insert(self.postPending).values(message=message, path_img=path_img, link_post=link_post,
                                                             time=time))
        self.session.commit()

    def get_post(self, link):
        result = self.session.execute(select(self.postPending).where(self.postPending.c.link_post == link))
        return result.fetchone()