core/base.py - master - Ksq/fastapi_app

core/base.py
103 строки4 KB
KSQ-web
Добавление репозитория
20 апр 2025, 16:29
20 апр 2025, 16:29e640632
from sqlalchemy import select, and_, text
from core.models import *
from core.schemas import UserCreate, OrderCreate, PMMessageCreate
from sqlalchemy.orm import joinedload
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from core.config import settings


class DataBase:
    def __init__(self):
        self.connect = settings.db_url
        self.async_engine = create_async_engine(self.connect, echo=False)
        self.Session = async_sessionmaker(
            bind=self.async_engine,
            class_=AsyncSession,
            autoflush=False,
            expire_on_commit=False,
            autocommit=False
        )

    async def create_db(self) -> None:
        async with self.async_engine.begin() as conn:
            await conn.run_sync(Base.metadata.create_all)

    async def check_connection(self) -> bool:
        try:
            async with self.async_engine.begin() as conn:
                await conn.execute(text('SELECT version()'))
                return True
        except SQLAlchemyError as e:
            print(f'Ошибка подключения к БД: {e}')
            return False

    async def check_user_register(self, email: str) -> bool:
        async with self.Session() as session:
            result = await session.scalar(select(Users).where(Users.email == email))
            return True if result else False

    async def create_user(self, user_create: UserCreate) -> Users:
        user = Users(**user_create.model_dump())
        async with self.Session() as session:
            session.add(user)
            await session.commit()
        return user
    
    async def check_user(self, login: str, password: str) -> Users:
        async with self.Session() as session:
            result = await session.scalar(select(Users).where(and_(Users.login == login, Users.password == password)))
            return result
        
    async def get_user_id(self, user_id: int) -> Users:
        async with self.Session() as session:
            result = await session.scalar(select(Users).where(Users.id == user_id))
            return result

    async def check_evacuation(self, author_id: int) -> Orders:
        async with self.Session() as session:
            result = await session.execute(select(Orders).options(joinedload(Orders.staff))
                                           .where(Orders.order_author == author_id)
                                           .order_by(Orders.id.desc()))
            return result.scalars().first()


    async def create_order(self, order_create: OrderCreate):
        order_data = order_create.dict()
        order = Orders(**order_data)
        async with self.Session() as session:
            session.add(order)
            await session.commit()
        return order
    
    async def order_one(self, order_id: int):
        async with self.Session() as session:
            result = await session.execute(select(Orders).options(joinedload(Orders.staff))
                                           .where(Orders.id == order_id)
                                           .order_by(Orders.id.desc()))
            return result.scalars().first()
        

    async def all_order(self, staff_id: int):
        async with self.Session() as session:
            result = await session.scalars(select(Orders).where(Orders.staff_author == staff_id))
            return result.all()
        
    async def user_orders(self, user_id: int):
        async with self.Session() as session:
            result = await session.scalars(select(Orders).where(Orders.order_author == user_id))
            return result.all()
        
    async def pm_message_order(self, order_id: int):
        async with self.Session() as session:
            result = await session.scalars(select(PMMessage).where(PMMessage.order_id == order_id))
            return result.all()
        
    async def create_message(self, message_create: PMMessageCreate):
        message = PMMessage(**message_create.model_dump())
        async with self.Session() as session:
            session.add(message)
            await session.commit()
        return message

db_connect = DataBase()