database/db_utils.py - master - AlexF88/FFB

database/db_utils.py
162 строки6 KB
Alex
Add clearing the cart after payment
16 июл 2026, 20:28
16 июл 2026, 20:2899ba0ec
from statistics import quantiles
from typing import Iterable

from sqlalchemy.orm import Session
from sqlalchemy import update, delete, select, DECIMAL
from sqlalchemy.sql.functions import sum
from sqlalchemy.exc import IntegrityError

from .models import Users, Categories, Carts, Finally_carts, Products, engine


with Session(engine) as session:
    db_session = session


def db_register_user(full_name, chat_id) -> bool:
    '''Первая регистрация пользователя с доступными данными'''
    try:
        query = Users(name=full_name, telegram=chat_id)
        db_session.add(query)
        db_session.commit()
        return False
    except IntegrityError:
        db_session.rollback()
        return True


def dp_update_user(chat_id: int, phone: str):
    '''Дополняем данные пользоватиеля телефонным номером'''
    query = update(Users).where(Users.telegram == chat_id).values(phone=phone)
    db_session.execute(query)
    db_session.commit()


def db_creat_user_cart(chat_id: int):
    '''Создание временной корзины пользователя'''
    try:
        subquery = db_session.scalar(select(Users).where(Users.telegram == chat_id))
        query = Carts(user_id=subquery.id)
        db_session.add(query)
        db_session.commit()
        return True
    except IntegrityError:
        '''Если карта уже существует'''
        db_session.rollback()
    except AttributeError:
        '''Если отправил контакт анонимный пользователь'''
        db_session.rollback()


def db_get_all_category() -> Iterable:
    '''Получаем все категории'''
    query = select(Categories)
    return db_session.scalars(query)


def db_get_products(category_id) -> Iterable:
    '''Получаем продукты категории по id категории'''
    query = select(Products).where(Products.category_id == category_id)
    return db_session.scalars(query)


def db_get_product_by_id(product_id: int) -> Products:
    '''Получаем продукт по его id'''
    query = select(Products).where(Products.ia == product_id)
    return db_session.scalar(query)


def db_get_user_cart(chat_id: int) -> Carts:
    '''Получаем id корзинки по связанной таблице Users'''
    query = select(Carts).join(Users).where(Users.telegram == chat_id)
    return db_session.scalar(query)


def db_update_to_cart(price: DECIMAL, cart_id: int, quantity:int=1) -> None:
    '''Обновляем данные временной корзины'''
    query = update(Carts).where(Carts.id == cart_id).values(total_price=price,
                                                            total_products=quantity)
    db_session.execute(query)
    db_session.commit()


def db_get_product_by_name(product_name: str) -> Products:
    '''Получаем продукт по его имени'''
    query = select(Products).where(Products.product_name == product_name)
    return db_session.scalar(query)


def db_ins_or_update_finally_cart(cart_id, product_name, total_products, total_price):
    '''Постоянная корзинкаб Вноим новую запись либо редактируем существующею если такой товар есть'''
    try:
        query = Finally_carts(cart_id=cart_id,
                              product_name=product_name,
                              quantity=total_products,
                              final_price=total_price)
        db_session.add(query)
        db_session.commit()
    except IntegrityError:
        db_session.rollback()
        query = update(Finally_carts
                       ).where(Finally_carts.product_name == product_name
                               ).where(Finally_carts.cart_id == cart_id
                                       ).values(quantity=total_products,
                                                final_price=total_price)
        db_session.execute(query)
        db_session.commit()
        return False


def db_get_finally_price(chat_id: int) -> DECIMAL:
    '''Получение общей суммы пользователя с постоянной корзины'''
    query = select(sum(Finally_carts.final_price)
                   ).join(Carts
                          ).join(Users
                                 ).where(Users.telegram == chat_id)
    return db_session.execute(query).fetchone()[0]


def db_get_FinallyCart_products(chat_id: int) -> Iterable:
    '''Получаем список товаров с корзины по id пользователя'''
    query = (select(Finally_carts.product_name,
                   Finally_carts.quantity,
                   Finally_carts.final_price,
                   Finally_carts.cart_id
                   )
             .join(Finally_carts.user_cart)
             .join(Carts.user_cart)
             .where(Users.telegram == chat_id))
             # .join(Users.telegram == chat_id))
    return db_session.execute(query).fetchall()


def db_get_product_for_delete(chat_id: int) -> Iterable:
    '''Получаем порядковй номер и название товара для удаления с корзиныы'''
    query = select(Finally_carts.id,
                   Finally_carts.product_name
                   ).join(Carts
                          ).join(Users
                                 ).where(Users.telegram == chat_id)

    return db_session.execute(query).fetchall()


def db_delete_product(finally_id: int) -> None:
    '''Удаление товара'''
    query = delete(Finally_carts).where(Finally_carts.id == finally_id)
    db_session.execute(query)
    db_session.commit()


def db_get_user_info(chat_id: int) -> Users:
    '''Для получения информации о пользователе'''
    query = select(Users).where(Users.telegram == chat_id)

    return db_session.scalar(query)


def clear_funally_cart(cart_id: int) -> None:
    '''Очистка корзины после оплаты'''
    query = delete(Finally_carts).where(Finally_carts.cart_id == cart_id)
    db_session.execute(query)
    db_session.commit()