# Import the necessary components from flask_restx from flask_restx import Resource, Namespace, fields,Model, reqparse # Import the BrandService from the application's services module from app.services.product_service import ProductService # Import the Product_dict and Productschema from the application's schemas module from app.schema.product import ProductSchema from app.services.brand_service import BrandService from werkzeug.datastructures import FileStorage import os product_namespace = Namespace("products",path='/product', description='Product-related operations') upload_parser = reqparse.RequestParser() upload_parser.add_argument('poster_image', location='files', type=FileStorage, required=True) UPLOAD_FOLDER = 'uploads' product_model = product_namespace.model("Product",{ "name": fields.String( required=True, help_text="Name of the product.", default="Orange Forfait" ), "poster_image": fields.String( help_text="URL/Path Poster image for the spot." ), "brand": fields.String(required=True, help="Name of the product's brand."), # "spots": fields.List( # fields.String(), # required=True, # help_text="References to the spots featuring the product.", # default=["spot_1","spot_2"] # ), }) @product_namespace.route('/') class ProductListResource(Resource): @product_namespace.doc( responses={ 200: 'OK - Product exists', 404: 'Not Found - Product does not exist', }, description='Check if a Product exists by its unique identifier.', security='API Key', tags=['Products'], ) def get(self): self.service = ProductService() try: products = self.service.get_all_products() if not products: return { 'msg': 'No Products found', 'data': [], 'status': 204 },204 else: return { 'msg': 'Products retrieved successfully', 'data': products, 'code_status': 200 }, 200 # OK status code except Exception as e: if hasattr(e, 'code') and e.code is not None: # If the exception has a status code attribute, return it return { 'msg': str(e), 'data': None, 'status': e.code }, e.code # Other status code else: # If no specific status code is provided, return a 500 Internal Server Error return { 'msg': 'An error occurred while retrieving Products: ' + str(e), 'data': None, 'status': 500 }, 500 # Internal Server Error @product_namespace.expect(product_model) @product_namespace.doc( responses={ 201: 'Created - New Product created successfully', 400: 'Bad Request - Invalid data', 409: 'Conflict - Product with this name already exists', 500: 'Internal Server Error - An error occurred while creating a new Product', }, description='Create a new Product using the provided data.', security='API Key', tags=['Products'], ) def post(self): self.service = ProductService() try: # args = upload_parser.parse_args() # # create upload folder # uploaded_file = args['poster_image'] # # Save the uploaded file to your desired location # file_path = f'{UPLOAD_FOLDER}/Products/{uploaded_file.filename}' # # if not os.path.exists(f'{UPLOAD_FOLDER}/Products/'): # # os.makedir(f'{UPLOAD_FOLDER}/Products/') # uploaded_file.save(file_path) # print(file_path) data = product_namespace.payload # data['poster_image'] = { # "size": os.path.getsize(file_path), # "path":file_path # } product_schema = ProductSchema() errors = product_schema.validate(data) if errors: # Return validation errors as a response with a 400 status code return { 'msg': 'Bad Request - Invalid data', 'errors': errors, 'data': None, 'status': 400 }, 400 existing_products = self.service.get_product_by_name(data['name']) if existing_products: return { 'msg': 'Conflict - Product with this name already exists', 'data': None, 'status': 409 }, 409 # Conflict brand = BrandService.get_brand_by_id(data['brand']) if not brand: return { 'msg': 'Brand Not Found', 'data': None, 'status': 404 }, 404 # Conflict new_product = self.service.create_product(data) return { 'msg': 'New Product created successfully', 'data': new_product, 'status': 201 },201 except Exception as e: if hasattr(e, 'code') and e.code is not None: # If the exception has a status code attribute, return it return { 'msg': str(e), 'data': None, 'status': e.code }, e.code # Other status code else: print(e) # If no specific status code is provided, return a 500 Internal Server Error return { 'msg': 'An error occurred while creating a new Product: ' + str(e), 'data': None, 'status': 500 }, 500 # Internal Server Error @product_namespace.doc( responses={ 204: 'No Content - All products deleted successfully', 404: 'Not Found - No products found', 500: 'Internal Server Error - An error occurred while deleting Products', }, description='Delete all products.', security='API Key', tags=['Products'], ) def delete(self): self.service = ProductService() try: deleted_count = self.service.delete_all_products() if deleted_count > 0: # Return a success message with HTTP status 204 (No Content) return { 'msg': f'{deleted_count} Products deleted', 'data': None, 'status': 204 }, 204 # no content else: # Return a message indicating that no Products were found with HTTP status 404 (Not Found) return { 'msg': 'No Products found', 'data': None, 'status': 404 }, 404 # Not Found except Exception as e: if hasattr(e, 'code') and e.code is not None: # If the exception has a status code attribute, return it return { 'msg': str(e), 'data': None, 'status': e.code }, e.code # Other status code else: # If no specific status code is provided, return a 500 Internal Server Error return { 'msg': 'Internal Server Error - An error occurred while deleting Products: ' + str(e), 'data': None, 'status': 500 }, 500 # Internal Server Error @product_namespace.route('/') class ProductResouce(Resource): @product_namespace.doc( responses={ 200: 'OK - Product retrieved successfully', 404: 'Not Found - Product not found', 500: 'Internal Server Error - An error occurred while retrieving the Product', }, description='Retrieve a Product by its unique identifier.', security='API Key', tags=['Product'], ) def get(self,product_id): self.service = ProductService() try: product = self.service.get_product_by_id(product_id=product_id) if not product: return { 'msg': 'Not Found - Product not found', 'data': None, 'status': 404 }, 404 # Not found return { 'msg': 'OK - Product retrieved successfully', 'data': product.to_dict(), 'status': 200 }, 200 # OK code status 200 except Exception as e: if hasattr(e, 'code') and e.code is not None: return { 'msg': str(e), 'data': None, 'status': e.code }, e.code else: return { 'msg': 'Internal Server Error - An error occurred while retrieving the Product: ' + str(e), 'data': None, 'status': 500 }, 500 @product_namespace.doc( responses={ 204: 'No Content - All products deleted successfully', 404: 'Not Found - No products found', 500: 'Internal Server Error - An error occurred while deleting Products', }, description='Delete all products.', security='API Key', tags=['Products'], ) def delete(self,product_id): self.service = ProductService() try: existing_products = self.service.get_product_by_id(product_id=product_id) if not existing_products: return { 'msg': 'Not Found - Product not found', 'data': None, 'status': 404 }, 404 # Not found self.service.delete_product(product_id=product_id) return { 'msg': 'No Content - Product deleted successfully', 'data': None, 'status': 204 },204 except Exception as e: if hasattr(e, 'code') and e.code is not None: return { 'msg': str(e), 'data': None, 'status': e.code }, e.code else: return { 'msg': 'Internal Server Error - An error occurred while deleting the product: ' + str(e), 'data': None, 'status': 500 }, 500 @product_namespace.route("search/") class ProductSearchResource(Resource): @product_namespace.doc( params={ 'query': 'Search query for Products.', }, responses={ 200: 'OK - Product retrieved successfully', 404: 'Not Found - No Product found', }, description='Search for Products using a query string.', security='API Key', tags=['Products'], ) def get(self,query): self.service = ProductService() try: # query = product_namespace.query.get("query") if query: matching_products = self.service.search_products(query) if matching_products: return { 'msg': 'OK - Products retrieved successfully', 'data': matching_products, 'status': 200 }, 200 else: return { 'msg': 'Not Found - No Product found', 'data': [], 'status': 404 }, 404 else: return { 'msg': 'Bad Request - Missing search query', 'data': None, 'status': 400 }, 400 except Exception as e: if hasattr(e, 'code') and e.code is not None: return { 'msg': str(e), 'data': None, 'status': e.code }, e.code else: return { 'msg': 'Internal Server Error - An error occurred while retrieving product count: ' + str(e), 'data': None, 'status': 500 }, 500 @product_namespace.route("/count") class ProductCountResource(Resource): @product_namespace.doc( responses={ 200: 'OK - Product count retrieved successfully', 500: 'Internal Server Error - An error occurred while retrieving Product count', }, description='Retrieve the count of all Products.', security='API Key', tags=['Products'], ) def get(self): self.service = ProductService() try: # Use ProductService to retrieve the count of all Products product_count = self.service.count_products() return { 'msg': 'OK - Product count retrieved successfully', 'data': product_count, 'status': 200 }, 200 except Exception as e: if hasattr(e, 'code') and e.code is not None: return { 'msg': str(e), 'data': None, 'status': e.code }, e.code else: return { 'msg': 'Internal Server Error - An error occurred while retrieving Product count: ' + str(e), 'data': None, 'status': 500 }, 500