119 lines
3.0 KiB
Python
119 lines
3.0 KiB
Python
|
from typing import Optional
|
||
|
|
|
||
|
|
from app.extensions import db_connection
|
||
|
|
|
||
|
|
from app.models.product_model import Product
|
||
|
|
|
||
|
|
|
||
|
|
def fetch_products(page: int = 0) -> Optional[list[Product]]:
|
||
|
|
cursor = db_connection.cursor(dictionary=True)
|
||
|
|
|
||
|
|
offset = 10 * page
|
||
|
|
cursor.execute(
|
||
|
|
"select product.id, user.displayname as seller, product.name, product.price_pc from product inner join user on user.id = product.seller_id order by product.id limit 10 offset %s",
|
||
|
|
(offset,),
|
||
|
|
)
|
||
|
|
results = cursor.fetchall()
|
||
|
|
|
||
|
|
if len(results) < 1:
|
||
|
|
return None
|
||
|
|
|
||
|
|
result_products: list[Product] = []
|
||
|
|
|
||
|
|
for row in results:
|
||
|
|
result_products.append(
|
||
|
|
Product(
|
||
|
|
product_id=row["id"],
|
||
|
|
seller_id=row["seller_id"],
|
||
|
|
name=row["name"],
|
||
|
|
price=row["price"],
|
||
|
|
creation_date=row["creation_date"],
|
||
|
|
)
|
||
|
|
)
|
||
|
|
|
||
|
|
return result_products
|
||
|
|
|
||
|
|
def fetch_product_by_id(product_id: int) -> Optional[Product]:
|
||
|
|
"""
|
||
|
|
Fetches specific product info
|
||
|
|
|
||
|
|
:param product_id: ID of product to be updated.
|
||
|
|
:type product_id: int
|
||
|
|
"""
|
||
|
|
|
||
|
|
cursor = db_connection.cursor(dictionary=True)
|
||
|
|
|
||
|
|
cursor.execute("select * from product where id = %s", (product_id,))
|
||
|
|
result = cursor.fetchone()
|
||
|
|
|
||
|
|
if cursor.rowcount != 1:
|
||
|
|
return None
|
||
|
|
|
||
|
|
result_product = Product(
|
||
|
|
product_id=result["id"],
|
||
|
|
seller_id=result["seller_id"],
|
||
|
|
name=result["name"],
|
||
|
|
price=result["price"],
|
||
|
|
creation_date=result["creation_date"],
|
||
|
|
)
|
||
|
|
|
||
|
|
return result_product
|
||
|
|
|
||
|
|
|
||
|
|
def fetch_product_extended_by_id(product_id: int) -> Optional[Product]:
|
||
|
|
"""
|
||
|
|
Fetches specific product info including the seller n
|
||
|
|
|
||
|
|
:param product_id: ID of product to be updated.
|
||
|
|
:type product_id: int
|
||
|
|
"""
|
||
|
|
|
||
|
|
cursor = db_connection.cursor(dictionary=True)
|
||
|
|
|
||
|
|
cursor.execute("select * from product inner join user on user.id = product.seller_id where product.id = %s", (product_id,))
|
||
|
|
result = cursor.fetchone()
|
||
|
|
|
||
|
|
if cursor.rowcount != 1:
|
||
|
|
return None
|
||
|
|
|
||
|
|
result_product = Product(
|
||
|
|
product_id=result["id"],
|
||
|
|
seller_id=result["seller_id"],
|
||
|
|
seller_name=result["displayname"],
|
||
|
|
name=result["name"],
|
||
|
|
price=result["price"],
|
||
|
|
creation_date=result["creation_date"],
|
||
|
|
)
|
||
|
|
|
||
|
|
return result_product
|
||
|
|
|
||
|
|
def insert_product(product: Product):
|
||
|
|
"""
|
||
|
|
Creates a new product listing
|
||
|
|
|
||
|
|
:param seller_id: User ID
|
||
|
|
:type seller_id: str
|
||
|
|
:param name: New product's name
|
||
|
|
:type name: str
|
||
|
|
:param price: New product's price
|
||
|
|
:type price: float
|
||
|
|
"""
|
||
|
|
|
||
|
|
cursor = db_connection.cursor()
|
||
|
|
|
||
|
|
cursor.execute(
|
||
|
|
"insert into product(seller_id, name, price_pc) values (%s, %s, %s)",
|
||
|
|
(product.seller_id, product.name, round(product.price, 2)),
|
||
|
|
)
|
||
|
|
db_connection.commit()
|
||
|
|
|
||
|
|
|
||
|
|
def delete_product(product: Product):
|
||
|
|
cursor = db_connection.cursor()
|
||
|
|
|
||
|
|
cursor.execute(
|
||
|
|
"delete from product where id = %s",
|
||
|
|
(product.product_id,),
|
||
|
|
)
|
||
|
|
db_connection.commit()
|