feat(fastapi-4-project): add fastapi fourth practice project

fastapi-projects/4-project/.python-version

@@ -0,0 +1 @@
+3.14

fastapi-projects/4-project/data/data_generation_guide.md

@@ -0,0 +1,17 @@
+# Guide
+
+Use [Mockaroo](https://www.mockaroo.com/) and generate `commerce` type data in `json` or `csv` format. You can get help from llm for this, how to generate fake+free data using Mockaroo. 
+
+# Fields
+
+Select following fields to get the application to work properly, as these this product is defined and whole application is built on top of it. So, use exact fields in Mockaroo Api for data generation.
+
+```py
+class Product:
+    id: int
+    name: str
+    description: str
+    price: float
+    category: str
+    image_url: str
+```

fastapi-projects/4-project/pyproject.toml

@@ -0,0 +1,12 @@
+[project]
+name = "4-project"
+version = "0.1.0"
+description = "4-project"
+authors = [
+    { name = "Saqib Bedar", email = "124094939+saqibbedar@users.noreply.github.com" }
+]
+requires-python = ">=3.14"
+dependencies = [
+    "fastapi>=0.141.1",
+    "uvicorn>=0.54.0",
+]

fastapi-projects/4-project/src/core/__init__.py

@@ -0,0 +1,3 @@
+from .config import JSON_DATA_FILE_PATH
+
+__all__ = ["JSON_DATA_FILE_PATH"]

fastapi-projects/4-project/src/core/config.py

@@ -0,0 +1,12 @@
+from pathlib import Path
+
+curr_file_dir = Path(__file__).parent
+
+# go two level up (to access data dir)
+Json_Data_File_Path = (curr_file_dir.parent.parent / "data" / "data.json").resolve()
+
+JSON_DATA_FILE_PATH: Path | None = (
+    Json_Data_File_Path
+    if (Json_Data_File_Path.exists() and Json_Data_File_Path.is_file() and Json_Data_File_Path.suffix == ".json")
+    else None
+)

fastapi-projects/4-project/src/main.py

@@ -0,0 +1,64 @@
+import json
+
+from fastapi import FastAPI, HTTPException
+
+from src.repositories import ProductRepository
+from src.schemas import ApiResponse
+
+# instantiate app
+app = FastAPI()
+
+
+# test-api
+@app.get("/")
+def home():
+    return ApiResponse(status_code=200, message="Homepage", data=None)
+
+
+# instantiate product repository
+product = ProductRepository()
+
+
+@app.get("/products")
+def get_all_products():
+    try:
+        # read data
+        payload: dict | None = product.get_all()
+
+        # breakpoint() # debugging
+
+        if payload:
+            return ApiResponse(status_code=200, message="All products", data=product.get_all())
+
+        raise HTTPException(status_code=404, detail="No product found")
+    except (FileNotFoundError, json.JSONDecodeError, OSError) as e:
+        print(e)
+        raise HTTPException(status_code=400, detail="Internal server error")
+
+
+@app.get("/products/search")
+def search_product(q: str, skip: int = 0, limit: int = 10):
+    try:
+        payload: dict | None = product.search(q, skip, limit)
+
+        if not payload:
+            raise HTTPException(status_code=404, detail="No product found")
+
+        return ApiResponse(status_code=200, message="Matched products", data=payload)
+    except (FileNotFoundError, json.JSONDecodeError, OSError) as e:
+        print(e)
+        raise HTTPException(status_code=500, detail="Internal server error: Database issue")
+
+
+@app.get("/products/{id}")
+def get_product_by_id(id: int):
+    try:
+        payload: dict | None = product.get_by_id(id=id)
+
+        if payload:
+            return ApiResponse(status_code=200, message="Matched product", data=payload)
+
+        raise HTTPException(status_code=404, detail="Product not found")
+    except (FileNotFoundError, json.JSONDecodeError, OSError) as e:
+        print(e)
+        raise HTTPException(status_code=500, detail="Internal server error: Database issue")

fastapi-projects/4-project/src/repositories/__init__.py

@@ -0,0 +1,3 @@
+from .product import ProductRepository
+
+__all__ = ["ProductRepository"]

fastapi-projects/4-project/src/repositories/product.py

@@ -0,0 +1,97 @@
+import json
+
+from src.core import JSON_DATA_FILE_PATH
+
+
+class ProductRepository:
+    # 1. get all products
+    def get_all(self) -> dict | None:
+
+        if not JSON_DATA_FILE_PATH:
+            raise FileNotFoundError("Invalid path or /data/data.json file not found")
+
+        # read file
+        with open(JSON_DATA_FILE_PATH, mode="r") as file:
+            payload = json.load(file)
+
+        if not payload:
+            return None
+
+        # DEBUG: inspect few values
+        # print("--- DEBUG Inspect payload ---")
+        # print(payload[1:5])
+        # print("-"*25)
+
+        return {"total_products": len(payload), "products": payload}
+
+    # 2. get by id
+    def get_by_id(self, id: int) -> dict | None:
+
+        if not JSON_DATA_FILE_PATH:
+            raise FileNotFoundError("Invalid path or /data/data.json file not found")
+
+        # read file
+        with open(JSON_DATA_FILE_PATH, mode="r") as file:
+            payload = json.load(file)
+
+        if not payload:
+            return None
+
+        return next((p for p in payload if p.get("id") == id), None)
+
+    # 3. search product
+    def search(self, q: str, skip: int = 0, limit: int = 10) -> dict | None:
+
+        if not JSON_DATA_FILE_PATH:
+            raise FileNotFoundError("Invalid path or /data/data.json file not found")
+
+        # read file
+        with open(JSON_DATA_FILE_PATH, mode="r") as file:
+            payload = json.load(file)
+
+        if not payload:
+            return None
+
+        # lowercase query
+        q = q.lower()
+
+        products = []
+
+        for p in payload:
+            if (
+                q in str(p.get("id"))
+                or q in str(p.get("price"))
+                or q in p.get("name").lower()
+                or q in p.get("description").lower()
+                or q in p.get("category").lower()
+            ):
+                products.append(p)
+
+        # handle if products found are with in default limit
+        if len(products) >= 1 and len(products) <= 10:
+            return {
+                "total_products": len(payload),
+                "found_products": len(products),
+                "skip": 0,
+                "limit": len(products),
+                "products": products[0 : 0 + len(products)],
+            }
+        else:
+            return {
+                "total_products": len(payload),
+                "found_products": len(products),
+                "skip": skip,
+                "limit": limit,
+                "products": products[skip : skip + limit],
+            }
+
+
+# # Debugging
+# product = ProductRepository()
+
+# # Get all products
+# print("--- DEBUG product repository ---")
+# all_products: dict | None = product.get_all()
+# if all_products:
+#     # print some products
+#     print(all_products["products"][1:2])

fastapi-projects/4-project/src/schemas/__init__.py

@@ -0,0 +1,3 @@
+from .response import ApiResponse
+
+__all__ = ["ApiResponse"]

fastapi-projects/4-project/src/schemas/response.py

@@ -0,0 +1,11 @@
+from typing import Generic, Optional, TypeVar
+
+from pydantic import BaseModel
+
+T = TypeVar("T")
+
+
+class ApiResponse(BaseModel, Generic[T]):
+    status_code: int = 200
+    message: str
+    data: Optional[T] = None