127 lines
3.4 KiB
Python
127 lines
3.4 KiB
Python
import base64
|
|
|
|
from flask import abort
|
|
from mysql.connector import Error
|
|
|
|
from decimal import Decimal
|
|
|
|
from app.extensions import db_connection
|
|
|
|
class ProductService:
|
|
|
|
@staticmethod
|
|
def get_products(page: int):
|
|
"""
|
|
Fetches 10 products
|
|
|
|
:param page: Page, aka offset of fetching
|
|
:type page: int
|
|
"""
|
|
|
|
try:
|
|
with db_connection.cursor(dictionary=True) as cursor:
|
|
|
|
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 {"Failed": "Failed to fetch products. You've probably selected too far with pages"}, 400
|
|
|
|
result_obj = []
|
|
for row in results:
|
|
mid_result = {
|
|
"id": row['id'],
|
|
"seller": row['seller'],
|
|
"name": row['name'],
|
|
"price": row['price_pc']
|
|
}
|
|
|
|
result_obj.append(mid_result)
|
|
|
|
return result_obj, 200
|
|
|
|
except Error as e:
|
|
return {"Failed": f"Failed to fetch products. Error: {e}"}, 500
|
|
|
|
@staticmethod
|
|
def get_product_info(fields: list[str], product_id: int):
|
|
"""
|
|
Fetches specific product info
|
|
|
|
:param fields: array of fields that can be fetched
|
|
:type fields: list[str]
|
|
:param product_id: ID of product to be updated.
|
|
:type product_id: int
|
|
"""
|
|
|
|
try:
|
|
with db_connection.cursor(dictionary=True) as cursor:
|
|
fields_to_sql = {
|
|
"name": "product.name",
|
|
"price": "product.price_pc as price",
|
|
"image": "product.image",
|
|
"image_name": "product.image_name",
|
|
"seller": "user.displayname as seller"
|
|
}
|
|
field_sql = []
|
|
|
|
for field in fields:
|
|
field_sql.append(fields_to_sql[field])
|
|
|
|
select_params = ", ".join(field_sql)
|
|
|
|
if "seller" in fields:
|
|
cursor.execute(f"select {select_params} from product inner join user on user.id = product.seller_id where product.id = %s", (product_id,))
|
|
else:
|
|
cursor.execute(f"select {select_params} from product where id = %s", (product_id,))
|
|
|
|
result = cursor.fetchone()
|
|
|
|
if cursor.rowcount != 1:
|
|
return {"Failed": "Failed to fetch product. Product likely doesn't exist"}, 400
|
|
|
|
result_obj = {}
|
|
|
|
for field in fields:
|
|
if field == "image": # Encode image into base64
|
|
result_obj[field] = base64.b64encode(result[field]).decode('utf-8')
|
|
else:
|
|
result_obj[field] = result[field]
|
|
|
|
return result_obj, 200
|
|
except Error as e:
|
|
return {"Failed": f"Failed to fetch product info. Error: {e}"}, 500
|
|
|
|
@staticmethod
|
|
def create_listing(seller_id: str, name: str, price: float):
|
|
"""
|
|
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
|
|
"""
|
|
|
|
try:
|
|
with db_connection.cursor() as cursor:
|
|
rounded_price = round(price, 2)
|
|
|
|
cursor.execute("insert into product(seller_id, name, price_pc) values (%s, %s, %s)", (seller_id, name, Decimal(str(rounded_price))))
|
|
db_connection.commit()
|
|
except Error as e:
|
|
return {"Failed": f"Failed to create product. {e}"}, 400
|
|
|
|
return {"Success": "Successfully created new product listing"}, 200
|
|
|
|
@staticmethod
|
|
def __is_base64_jpg(decoded_string):
|
|
try:
|
|
# Checking if the decoded data represents a valid JPEG image
|
|
image_type = imghdr.what(None, decoded_data)
|
|
return image_type == 'jpeg'
|
|
except Exception:
|
|
return False |