module_27_postgres_migrations/lessons/3/sync/flask_app.py - master - Russid/python_advanced

module_27_postgres_migrations/lessons/3/sync/flask_app.py
106 строк3 KB
User
Rename folder to module_27_postgres_migrations
27 ноя 2023, 18:27
27 ноя 2023, 18:27fec1888
from sqlalchemy import Column, Integer, String, Float, \
    create_engine, Sequence, Identity, ForeignKey, delete
from sqlalchemy.orm import sessionmaker, declarative_base, relationship
from flask import Flask, jsonify
from typing import Dict, Any
from sqlalchemy.dialects.postgresql import insert

app = Flask(__name__)
engine = create_engine('postgresql+psycopg2://admin:admin@localhost')
Base = declarative_base()
Session = sessionmaker(bind=engine)
session = Session()


class Product(Base):
    __tablename__ = 'products'

    id = Column(Integer, Sequence('product_id_seq'), primary_key=True)
    title = Column(String(200), nullable=False)
    count = Column(Integer, default=0)
    price = Column(Float, default=0)
    user_id = Column(Integer, ForeignKey('users.id'))
    user = relationship("User", backref="products")

    def __repr__(self):
        return f"Товар {self.title}"

    def to_json(self) -> Dict[str, Any]:
        return {c.name: getattr(self, c.name) for c in
                self.__table__.columns}


class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    name = Column(String(50), nullable=False)
    surname = Column(String(50), nullable=True)
    num = Column(Integer, Identity(minvalue=100, maxvalue=1000, cycle=True))

    def __repr__(self):
        return f"Пользователь {self.username}"

    def to_json(self) -> Dict[str, Any]:
        return {c.name: getattr(self, c.name) for c in
                self.__table__.columns}


@app.before_request
def before_request_func():
    Base.metadata.drop_all(engine)
    Base.metadata.create_all(engine)

    objects = [
        User(name='u1'),
        User(name='u2'),
        User(name='u3'),
        Product(title="p1", user_id=1),
        Product(title="p2", user_id=2),
        Product(title="p3", user_id=3)
    ]
    session.bulk_save_objects(objects)
    session.commit()


@app.route('/products/<int:id>', methods=['DELETE'])
def delete_product_handler(id: int):
    result = delete(Product).returning(Product.id, Product.title). \
        where(Product.id == id)
    deleted_row = session.execute(result).fetchone()
    if deleted_row:
        deleted_row_json = dict(id=deleted_row[0], title=deleted_row[1])
        return jsonify(delete_row_attrs=deleted_row_json)


@app.route('/products', methods=['POST'])
def insert_product_handler():
    insert_query = insert(Product).values(id='2',
                                         title='новый продукт')
    # do nothing
    # do_nothing_stmt = insert_query \
    #     .on_conflict_do_nothing(index_elements=['id'])
    # session.execute(do_nothing_stmt)

    # do update
    do_update_stmt = insert_query \
        .on_conflict_do_update(constraint='products_pkey',
                               set_=dict(title='обновленный продукт'))
    session.execute(do_update_stmt)
    session.commit()
    return '',200


@app.route('/products', methods=['GET'])
def get_products_handler():
    products = session.query(Product).all()
    products_list = []
    for p in products:
        product_obj = p.to_json()
        product_obj['user'] = p.user.to_json()
        products_list.append(product_obj)
    return jsonify(products_list)


if __name__ == '__main__':
    app.run()