swag-shop/app/services/product_service.py

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