ORMhw.py - master - fess/BaseDate

ORMhw.py
129 строк4 KB
Станислав Константинов
upload files for ORM HW
03 фев 2026, 17:19
03 фев 2026, 17:196bae2d4
import sqlalchemy
import sqlalchemy as sq
from sqlalchemy import false
from sqlalchemy.orm import declarative_base, relationship, sessionmaker
from sqlalchemy import TIMESTAMP

import json

Base = declarative_base()

class Publisher(Base):
    __tablename__ = "publisher"

    id = sq.Column(sq.Integer, primary_key=True)
    name = sq.Column(sq.String(length=40), unique=True)

    def __str__(self):
        return f'ID: {self.id}, name {self.name}'

class Book(Base):
    __tablename__ = "book"

    id = sq.Column(sq.Integer, primary_key=True)
    title = sq.Column(sq.String(length=40), nullable=False)
    id_publisher = sq.Column(sq.Integer, sq.ForeignKey("publisher.id"), nullable=False)

    book_publisher = relationship(Publisher, backref="book")

    def __str__(self):
        return f'ID: {self.id}, title {self.title}, id_publisher {self.id_publisher}'


class Shop(Base):
    __tablename__ = "shop"

    id = sq.Column(sq.Integer, primary_key=True)
    name = sq.Column(sq.String(length=40), unique=True)

    def __str__(self):
        return f'ID: {self.id}, name {self.name}'

class Stock(Base):
    __tablename__ = "stock"

    id = sq.Column(sq.Integer, primary_key=True)
    id_book = sq.Column(sq.Integer, sq.ForeignKey("book.id"), nullable=False)
    id_shop = sq.Column(sq.Integer, sq.ForeignKey("shop.id"), nullable=False)
    count = sq.Column(sq.Integer, nullable=False)

    stok_shop = relationship(Shop, backref="stock")
    stok_book = relationship(Book, backref="stock")

    def __str__(self):
        return f'ID: {self.id}, id_book {self.id_book}, id_shop {self.id_shop}, count {self.count}'

class Sale(Base):
    __tablename__ = "sale"

    id = sq.Column(sq.Integer, primary_key=True)
    price = sq.Column(sq.Float, nullable=False)
    date_sale = sq.Column(sq.Date, nullable=False)
    id_stock = sq.Column(sq.Integer, sq.ForeignKey("stock.id"), nullable=False)
    count = sq.Column(sq.Integer, nullable=False)

    sale_stock = relationship(Stock, backref="sale")

    def __str__(self):
        return f'ID: {self.id}, prise {self.price}, data _sale {self.date_sale}, id_stock {self.id_stock}, count {self.count}'

def create_tables(engine):
    Base.metadata.drop_all(engine)
    Base.metadata.create_all(engine)


DSN = "postgresql://postgres:   @localhost:5432/books_db"
engine = sq.create_engine(DSN)
create_tables(engine)

# сессия
Session = sessionmaker(bind=engine)
session = Session()

with open('tests_data.json', encoding='utf-8') as file:
    data = json.load(file)

for record in data:
    model = {
        'publisher': Publisher,
        'shop': Shop,
        'book': Book,
        'stock': Stock,
        'sale': Sale,
    }[record.get('model')]
    session.add(model(id=record.get('pk'), **record.get('fields')))
session.commit()

print("Данные успешно загружены!")
for c in session.query(Publisher).all():
    print(c)
find_publisher = 'Pearson'
 # поиск по издателю по форме название книги | название магазина, в котором была куплена эта книга | стоимость покупки | дата покупки
def get_publisher_sales_subquery(publisher_name):
    # Создаем подзапрос для поиска издателя
    publisher_subq = session.query(Publisher.id).filter(Publisher.name.ilike(f"%{publisher_name}%")).subquery()

    # Основной запрос
    results = (
        session.query(
            Book.title,
            Shop.name,
            Sale.price,
            Sale.date_sale
        )
        .join(publisher_subq, Book.id_publisher == publisher_subq.c.id)
        .join(Stock, Stock.id_book == Book.id)
        .join(Shop, Shop.id == Stock.id_shop)
        .join(Sale, Sale.id_stock == Stock.id)
        .all()
    )

    if results:
        print(f"\nПокупки книг издателя: {publisher_name}")
        print("=" * 80)
        for title, shop, price, date in results:
            print(f"{title:30} | {shop:15} | {price:8.2f} руб. | {date.strftime('%d.%m.%Y')}")
    else:
        print(f"Для издателя '{publisher_name}' не найдено записей о покупках")
get_publisher_sales_subquery(find_publisher)
session.close()