code update: update fastapi

Author: ratchanonth60

app/api/v1/address.py

@@ -1,4 +1,3 @@
-# app/api/v1/address.py
 from fastapi import APIRouter, Depends, HTTPException, Request
 from sqlalchemy.orm import Session
 
@@ -9,7 +8,7 @@
 from app.schemas.base import Failed, Successfully
 from app.services.address import AddressService
 
-router = APIRouter(prefix="/addresses", tags=["addresses"])
+router = APIRouter(prefix="/addresses", tags=["Addresses"])
 
 
 def get_current_user(request: Request) -> User:

app/api/v1/auth.py

@@ -12,7 +12,7 @@
 from app.schemas.base import Successfully
 from app.services.auth import AuthService
 
-router = APIRouter(prefix="/auth", tags=["auth"])
+router = APIRouter(prefix="/auth", tags=["Auth"])
 
 
 def get_auth_service(db: Session = Depends(get_db)) -> AuthService:

app/api/v1/cart.py

@@ -0,0 +1,161 @@
+from typing import List
+
+from fastapi import APIRouter, Depends, HTTPException, Request
+from sqlalchemy.orm import Session
+
+from app.core.db.connect import get_db
+from app.models.cart import Cart, CartLine
+from app.models.user import User, UserRole
+from app.schemas.base import Failed, Successfully
+from app.services.cart import (
+    CartCreate,
+    CartLineCreate,
+    CartLineService,
+    CartLineUpdate,
+    CartService,
+    CartUpdate,
+)
+
+router = APIRouter(prefix="/carts", tags=["Cart"])
+
+
+def get_current_user(request: Request) -> User:
+    """Dependency to get the current user from request.state."""
+    user = request.state.user
+    if not user:
+        raise HTTPException(status_code=401, detail="Authentication required")
+    return user
+
+
+def get_admin_user(current_user: User = Depends(get_current_user)) -> User:
+    """Dependency to ensure the user is an admin."""
+    if current_user.role != UserRole.ADMIN:  # type: ignore[comparison-overlap]
+        raise HTTPException(status_code=403, detail="Admin access required")
+    return current_user
+
+
+@router.get("/carts/{cart_id}", response_model=Successfully[Cart])
+def get_cart(cart_id: int, db: Session = Depends(get_db)):
+    service = CartService(db)
+    cart = service.get(cart_id)
+    if not cart:
+        return Failed(msg="Cart not found", code=404)
+    return Successfully(data=cart, msg="Cart retrieved successfully", code=200)
+
+
+@router.get("/carts/", response_model=Successfully[List[Cart]])
+def get_carts(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
+    service = CartService(db)
+    list_cart = service.get_multi(skip, limit)
+    return Successfully(data=list_cart, msg="Carts retrieved successfully", code=200)
+
+
+@router.post("/carts/", response_model=Successfully[Cart])
+def create_cart(cart: CartCreate, db: Session = Depends(get_db)):
+    service = CartService(db)
+    obj = service.create(cart)
+    return Successfully(data=obj, msg="Cart created successfully", code=201)
+
+
+@router.put("/carts/{cart_id}", response_model=Successfully[Cart])
+def update_cart(cart_id: int, cart: CartUpdate, db: Session = Depends(get_db)):
+    service = CartService(db)
+    db_cart = service.get(cart_id)
+    if not db_cart:
+        return Failed(msg="Cart not found", code=404)
+    obj = service.update(db_cart, cart)
+    return Successfully(data=obj, msg="Cart updated successfully", code=200)
+
+
+@router.delete("/carts/{cart_id}", response_model=Successfully[Cart])
+def delete_cart(cart_id: int, db: Session = Depends(get_db)):
+    service = CartService(db)
+    cart = service.remove(cart_id)
+    if not cart:
+        return Failed(msg="Cart not found", code=404)
+    return Successfully(data=cart, msg="Cart deleted successfully", code=200)
+
+
+@router.get("/carts/{cart_id}/total/")
+def get_cart_total(cart_id: int, tax_rate: float = 0.07, db: Session = Depends(get_db)):
+    service = CartService(db)
+    total = service.calculate_total(cart_id, tax_rate)
+    json = {"cart_id": cart_id, "total_amount": total}
+    return Successfully(data=json, msg="Cart total retrieved successfully", code=200)
+
+
+@router.post("/carts/{cart_id}/apply-discount/")
+def apply_cart_discount(
+    cart_id: int, discount_rate: float, db: Session = Depends(get_db)
+):
+    service = CartService(db)
+    total_discount = service.apply_discount(cart_id, discount_rate)
+    json = {"cart_id": cart_id, "total_discount": total_discount}
+    return Successfully(data=json, msg="Cart discount applied successfully", code=200)
+
+
+# 4. CartLine Endpoints
+@router.get("/cart-lines/{cart_line_id}", response_model=Successfully[CartLine])
+def get_cart_line(cart_line_id: int, db: Session = Depends(get_db)):
+    service = CartLineService(db)
+    cart_line = service.get(cart_line_id)
+    if not cart_line:
+        return Failed(msg="CartLine not found", code=404)
+    return Successfully(data=cart_line, msg="CartLine retrieved successfully", code=200)
+
+
+@router.get("/cart-lines/", response_model=Successfully[List[CartLine]])
+def get_cart_lines(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
+    service = CartLineService(db)
+    obj = service.get_multi(skip, limit)
+    return Successfully(data=obj, msg="CartLines retrieved successfully", code=200)
+
+
+@router.post("/cart-lines/", response_model=Successfully[CartLine])
+def create_cart_line(cart_line: CartLineCreate, db: Session = Depends(get_db)):
+    service = CartLineService(db)
+    obj = service.create(cart_line)
+    return Successfully(data=obj, msg="CartLine created successfully", code=201)
+
+
+@router.put("/cart-lines/{cart_line_id}", response_model=Successfully[CartLine])
+def update_cart_line(
+    cart_line_id: int, cart_line: CartLineUpdate, db: Session = Depends(get_db)
+):
+    service = CartLineService(db)
+    db_cart_line = service.get(cart_line_id)
+    if not db_cart_line:
+        return Failed(msg="CartLine not found", code=404)
+    obj = service.update(db_cart_line, cart_line)
+    return Successfully(data=obj, msg="CartLine updated successfully", code=200)
+
+
+@router.delete("/cart-lines/{cart_line_id}", response_model=Successfully[CartLine])
+def delete_cart_line(cart_line_id: int, db: Session = Depends(get_db)):
+    service = CartLineService(db)
+    cart_line = service.remove(cart_line_id)
+    if not cart_line:
+        return Failed(msg="CartLine not found", code=404)
+    return Successfully(data=cart_line, msg="CartLine deleted successfully", code=200)
+
+
+@router.post("/cart-lines/{cart_line_id}/apply-discount/")
+def apply_cart_line_discount(
+    cart_line_id: int, discount_rate: float, db: Session = Depends(get_db)
+):
+    service = CartLineService(db)
+    discount = service.apply_discount(cart_line_id, discount_rate)
+    json = {"cart_line_id": cart_line_id, "discount": discount}
+    return Successfully(
+        data=json, msg="CartLine discount applied successfully", code=200
+    )
+
+
+@router.get("/cart-lines/{cart_line_id}/tax/")
+def get_cart_line_tax(
+    cart_line_id: int, tax_rate: float = 0.07, db: Session = Depends(get_db)
+):
+    service = CartLineService(db)
+    tax = service.calculate_tax(cart_line_id, tax_rate)
+    json = {"cart_line_id": cart_line_id, "tax": tax}
+    return Successfully(data=json, msg="CartLine tax calculated successfully", code=200)

app/api/v1/catalogue.py

@@ -0,0 +1,58 @@
+from typing import List
+
+from fastapi import APIRouter, Depends
+from sqlalchemy.orm import Session
+
+from app.core.db.connect import get_db
+from app.models.catalogue import Catalogue
+from app.schemas.base import Failed, Successfully
+from app.schemas.catalogue import CatalogueCreate, CatalogueUpdate
+from app.services.catalogue import CatalogueService
+
+router = APIRouter(prefix="/catalogues", tags=["Catalogue"])
+
+
+@router.get("/catalogues/{catalogue_id}", response_model=Successfully[Catalogue])
+def get_catalogue(catalogue_id: int, db: Session = Depends(get_db)):
+    service = CatalogueService(db)
+    catalogue = service.get(catalogue_id)
+    if not catalogue:
+        return Failed(code=404, msg="Catalogue not found")
+    return Successfully(
+        data=catalogue, msg="Catalogue retrieved successfully", code=200
+    )
+
+
+@router.get("/catalogues/", response_model=Successfully[List[Catalogue]])
+def get_catalogues(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
+    service = CatalogueService(db)
+    obj = service.get_multi(skip, limit)
+    return Successfully(data=obj, msg="Catalogues retrieved successfully", code=200)
+
+
+@router.post("/catalogues/", response_model=Successfully[Catalogue])
+def create_catalogue(catalogue: CatalogueCreate, db: Session = Depends(get_db)):
+    service = CatalogueService(db)
+    obj = service.create(catalogue)
+    return Successfully(data=obj, msg="Catalogue created successfully", code=201)
+
+
+@router.put("/catalogues/{catalogue_id}", response_model=Successfully[Catalogue])
+def update_catalogue(
+    catalogue_id: int, catalogue: CatalogueUpdate, db: Session = Depends(get_db)
+):
+    service = CatalogueService(db)
+    db_catalogue = service.get(catalogue_id)
+    if not db_catalogue:
+        return Failed(code=404, msg="Catalogue not found")
+    obj = service.update(db_catalogue, catalogue)
+    return Successfully(data=obj, msg="Catalogue updated successfully", code=200)
+
+
+@router.delete("/catalogues/{catalogue_id}", response_model=Successfully[Catalogue])
+def delete_catalogue(catalogue_id: int, db: Session = Depends(get_db)):
+    service = CatalogueService(db)
+    catalogue = service.remove(catalogue_id)
+    if not catalogue:
+        return Failed(code=404, msg="Catalogue not found")
+    return Successfully(data=catalogue, msg="Catalogue deleted successfully", code=200)

app/api/v1/checkout.py

@@ -0,0 +1,58 @@
+from typing import List
+
+from fastapi import APIRouter, Depends
+from sqlalchemy.orm import Session
+
+from app.core.db.connect import get_db
+from app.models.checkout import Checkout
+from app.schemas.checkout import CheckoutCreate, CheckoutUpdate
+from app.services.checkout import CheckoutService
+from app.schemas.base import Failed, Successfully
+
+router = APIRouter(prefix="/checkouts", tags=["Checkout"])
+
+
+@router.get("/checkouts/{checkout_id}", response_model=Successfully[Checkout])
+def get_checkout(checkout_id: int, db: Session = Depends(get_db)):
+    service = CheckoutService(db)
+    checkout = service.get(checkout_id)
+    if not checkout:
+        return Failed(code=404, msg="Checkout not found")
+    return Successfully(
+        status="success", code=200, msg="Checkout retrieved", data=checkout
+    )
+
+
+@router.get("/checkouts/", response_model=Successfully[List[Checkout]])
+def get_checkouts(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
+    service = CheckoutService(db)
+    obj = service.get_multi(skip, limit)
+    return Successfully(data=obj, msg="Checkouts retrieved successfully", code=200)
+
+
+@router.post("/checkouts/", response_model=Successfully[Checkout])
+def create_checkout(checkout: CheckoutCreate, db: Session = Depends(get_db)):
+    service = CheckoutService(db)
+    obj = service.create(checkout)
+    return Successfully(data=obj, msg="Checkout created successfully", code=201)
+
+
+@router.put("/checkouts/{checkout_id}", response_model=Successfully[Checkout])
+def update_checkout(
+    checkout_id: int, checkout: CheckoutUpdate, db: Session = Depends(get_db)
+):
+    service = CheckoutService(db)
+    db_checkout = service.get(checkout_id)
+    if not db_checkout:
+        return Failed(code=404, msg="Checkout not found")
+    obj = service.update(db_checkout, checkout)
+    return Successfully(data=obj, msg="Checkout updated successfully", code=200)
+
+
+@router.delete("/checkouts/{checkout_id}", response_model=Successfully[Checkout])
+def delete_checkout(checkout_id: int, db: Session = Depends(get_db)):
+    service = CheckoutService(db)
+    checkout = service.remove(checkout_id)
+    if not checkout:
+        return Failed(code=404, msg="Checkout not found")
+    return Successfully(data=checkout, msg="Checkout deleted successfully", code=200)

app/api/v1/users.py

@@ -7,7 +7,7 @@
 from app.schemas.user import UserResponse, UserUpdate
 from app.services.user import UserService
 
-router = APIRouter(prefix="/users", tags=["users"])
+router = APIRouter(prefix="/users", tags=["Users"])
 
 
 def get_current_user(request: Request) -> User:

app/core/security.py

@@ -1,17 +1,8 @@
-from fastapi import Depends
-from fastapi.security import OAuth2PasswordBearer
 from passlib.context import CryptContext
 
-from .auth_handler import decode_jwt
 
 pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
-oauth2_scheme = OAuth2PasswordBearer(tokenUrl="v1/auth/login")
 
 
 def get_password_hash(password: str) -> str:
     return pwd_context.hash(password)
-
-
-async def get_current_user(token: str = Depends(oauth2_scheme)):
-    payload = decode_jwt(token)
-    return payload.get("id")

app/main.py

@@ -1,10 +1,13 @@
 import logging
 
 from fastapi import FastAPI
-
+from fastapi.security import OAuth2AuthorizationCodeBearer
+from app.api.v1.address import router as address_router
 from app.api.v1.auth import router as auth_router
+from app.api.v1.cart import router as cart_router
 from app.api.v1.users import router as users_router
-from app.api.v1.address import router as address_router
+from app.api.v1.catalogue import router as catalogue_router
+from app.api.v1.checkout import router as checkout_router
 from app.core.config import (
     configure_app,
     configure_exception_handlers,
@@ -13,6 +16,10 @@
 from app.core.db.connect import Base, get_engine
 
 logger = logging.getLogger(__name__)
+oauth2_scheme = OAuth2AuthorizationCodeBearer(
+    authorizationUrl="/v1/auth/login",  # URL สำหรับการขอ token (ในที่นี้เป็น placeholder)
+    tokenUrl="/v1/auth/login",  # URL สำหรับการขอ token (ในที่นี้เป็น placeholder)
+)
 
 
 def create_app(db_engine=None) -> FastAPI:
@@ -38,7 +45,10 @@ async def shutdown_event():
     configure_exception_handlers(app)
     app.include_router(auth_router, prefix="/v1")
     app.include_router(users_router, prefix="/v1")
-    app.include_router(address_router, prefix="/v1")  # Add this line
+    app.include_router(address_router, prefix="/v1")
+    app.include_router(cart_router, prefix="/v1")
+    app.include_router(catalogue_router, prefix="/v1")
+    app.include_router(checkout_router, prefix="/v1")
 
     @app.get("/")
     def read_root():

app/models/user.py

@@ -31,7 +31,6 @@ class User(BaseModel):
 
     last_login = Column(DateTime(timezone=True), nullable=True)
 
-    addresses = relationship("Address", back_populates="user")
 
     # # ความสัมพันธ์กับตารางอื่น
     # orders = relationship("Order", back_populates="user")  # ความสัมพันธ์กับคำสั่งซื้อ