code update: update Address, Analytics, Cart, CartLine, Catalogue, Checkout, Communication, Customer, Dashboard, Offer, Order, Partner, Payment, Search, Shipping, Voucher, and Wishlist

Author: ratchanonth60

Dockerfile

@@ -1,9 +1,16 @@
 FROM python:3.12-slim
 
-WORKDIR /app
+WORKDIR /fast-api/app
 
 COPY requirements.txt .
 RUN pip install --no-cache-dir -r requirements.txt
+ENV PYTHONPATH="/fast-api/app"
 
-COPY ./app/ .
+# Copy the entire app directory
+COPY ./app /fast-api/app
 
+# Set PYTHONPATH to ensure imports work correctly
+ENV PYTHONPATH="/fast-api"
+
+
+# CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

Makefile

@@ -51,3 +51,6 @@ log:
 
 ssh:
 	docker-compose exec app bash
+
+restart:
+	docker-compose restart $(target)

app/api/v1/address.py

@@ -0,0 +1,134 @@
+# app/api/v1/address.py
+from fastapi import APIRouter, Depends, HTTPException, Request
+from sqlalchemy.orm import Session
+
+from app.core.db.connect import get_db
+from app.models.address import Address
+from app.models.user import User, UserRole
+from app.schemas.address import AddressCreate, AddressResponse, AddressUpdate
+from app.schemas.base import Failed, Successfully
+from app.services.address import AddressService
+
+router = APIRouter(prefix="/addresses", tags=["addresses"])
+
+
+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.post("/", response_model=Successfully[AddressResponse])
+async def create_address(
+    address: AddressCreate,
+    current_user: User = Depends(get_current_user),
+    db: Session = Depends(get_db),
+):
+    """Create a new address for the authenticated user."""
+    address_service = AddressService(db)
+    address.user_id = current_user.id
+    new_address = address_service.create(address)
+    return Successfully(
+        code=201,
+        msg="Address created successfully",
+        data=AddressResponse.from_orm(new_address),
+    )
+
+
+@router.get("/me", response_model=Successfully[list[AddressResponse]])
+async def get_my_addresses(
+    skip: int = 0,
+    limit: int = 100,
+    current_user: User = Depends(get_current_user),
+    db: Session = Depends(get_db),
+):
+    addresses = (
+        db.query(Address)
+        .filter(Address.user_id == current_user.id)
+        .offset(skip)
+        .limit(limit)
+        .all()
+    )
+    return Successfully(
+        code=200,
+        msg="Addresses retrieved successfully",
+        data=[AddressResponse.from_orm(addr) for addr in addresses],
+    )
+
+
+@router.get("/{address_id}", response_model=Successfully[AddressResponse])
+async def get_address(
+    address_id: int,
+    current_user: User = Depends(get_current_user),
+    db: Session = Depends(get_db),
+):
+    """Get a specific address by ID (restricted to owner or admin)."""
+    address_service = AddressService(db)
+    address = address_service.get(address_id)
+    if not address:
+        return Failed(code=404, msg="Address not found")
+    if address.user_id != current_user.id and current_user.role != UserRole.ADMIN:
+        raise HTTPException(
+            status_code=403, detail="Not authorized to view this address"
+        )
+    return Successfully(
+        code=200,
+        msg="Address retrieved successfully",
+        data=AddressResponse.from_orm(address),
+    )
+
+
+@router.put("/{address_id}", response_model=Successfully[AddressResponse])
+async def update_address(
+    address_id: int,
+    address_update: AddressUpdate,
+    current_user: User = Depends(get_current_user),
+    db: Session = Depends(get_db),
+):
+    """Update an address (restricted to owner or admin)."""
+    address_service = AddressService(db)
+    address = address_service.get(address_id)
+    if not address:
+        return Failed(code=404, msg="Address not found")
+    if address.user_id != current_user.id and current_user.role != UserRole.ADMIN:
+        raise HTTPException(
+            status_code=403, detail="Not authorized to update this address"
+        )
+    updated_address = address_service.update(address, address_update)
+    return Successfully(
+        code=200,
+        msg="Address updated successfully",
+        data=AddressResponse.from_orm(updated_address),
+    )
+
+
+@router.delete("/{address_id}", response_model=Successfully[None])
+async def delete_address(
+    address_id: int,
+    current_user: User = Depends(get_current_user),
+    db: Session = Depends(get_db),
+):
+    """Delete an address (restricted to owner or admin)."""
+    address_service = AddressService(db)
+    address = address_service.get(address_id)
+    if not address:
+        return Failed(code=404, msg="Address not found")
+    if address.user_id != current_user.id and current_user.role != UserRole.ADMIN:
+        raise HTTPException(
+            status_code=403, detail="Not authorized to delete this address"
+        )
+    address_service.remove(address_id)
+    return Successfully(
+        code=200,
+        msg="Address deleted successfully",
+        data=None,
+    )

app/api/v1/auth.py

@@ -1,81 +1,83 @@
-from fastapi import APIRouter, Depends
-import jwt
-from sqlalchemy.orm import Session
+from fastapi import APIRouter, BackgroundTasks, Depends
 from fastapi.security import OAuth2PasswordRequestForm
+from sqlalchemy.orm import Session
 
-
-from app.core.auth_handler import sign_jwt, sign_refresh_token, decode_jwt
 from app.core.db.connect import get_db
-from app.core.security import get_password_hash
-from app.models.user import User
 from app.schemas.auth import (
-    JWTBearer,
-    UserCreate,
-    UserResponse,
+    ConfirmPasswordRequest,
     RefreshTokenRequest,
+    ResetPasswordRequest,
+    UserCreate,
 )
-from app.schemas.base import Failed, Successfully
+from app.schemas.base import Successfully
+from app.services.auth import AuthService
 
 router = APIRouter(prefix="/auth", tags=["auth"])
 
 
+def get_auth_service(db: Session = Depends(get_db)) -> AuthService:
+    return AuthService(db)
+
+
 @router.post("/login")
 def login(
     form_data: OAuth2PasswordRequestForm = Depends(),
-    db: Session = Depends(get_db),
+    auth_service: AuthService = Depends(get_auth_service),
 ):
-    user = db.query(User).filter(User.username == form_data.username).first()
-    print(user)
-    if not user or not user.verify_password(form_data.password):
-        return Failed(status="fail", code=404, msg="Invalid username or password")
-    user.update_last_login()
-    db.commit()
-    id = str(user.id)
-    access_token, refresh_token = sign_jwt(id), sign_refresh_token(id)
-    jwt_token = JWTBearer(
-        access_token=access_token, refresh_token=refresh_token, token_type="bearer"
-    )
+    jwt_token = auth_service.login(form_data.username, form_data.password)
     return Successfully(
         code=200, msg="Login successfully", data=jwt_token, status="success"
     )
 
 
-@router.post("/register", response_model=UserResponse)
-async def register(user: UserCreate, db: Session = Depends(get_db)):
-    existing_user = (
-        db.query(User)
-        .filter((User.username == user.username) | (User.email == user.email))
-        .first()
+@router.post("/register")
+async def register(
+    user: UserCreate,
+    background_tasks: BackgroundTasks,
+    auth_service: AuthService = Depends(get_auth_service),
+):
+    new_user = auth_service.register(user, background_tasks)
+    return Successfully(status="success", code=201, msg="User created", data=new_user)
+
+
+@router.post("/refresh")
+def refresh_token(
+    refresh_token_request: RefreshTokenRequest,
+    auth_service: AuthService = Depends(get_auth_service),
+):
+    new_access_token = auth_service.refresh_token(refresh_token_request)
+    return Successfully(
+        status="success", code=200, msg="Refresh token", data=new_access_token
     )
 
-    if existing_user:
-        return Failed(status="fail", code=400, msg="Username or email already exists")
-    hashed_password = get_password_hash(user.password)
 
-    new_user = User(
-        username=user.username, email=user.email, hashed_password=hashed_password
+@router.post("/reset-password-request")
+def reset_password_request(
+    email: str,
+    auth_service: AuthService = Depends(get_auth_service),
+):
+    reset_token = auth_service.reset_password_request(email)
+    return Successfully(
+        code=200,
+        msg="Reset password token sent",
+        data=reset_token,
+        status="success",
     )
 
-    db.add(new_user)
-    db.commit()
-    db.refresh(new_user)
 
-    return Successfully(status="fail", code=201, msg="User created", data=user)
+@router.post("/reset-password")
+def reset_password(
+    request: ResetPasswordRequest,
+    auth_service: AuthService = Depends(get_auth_service),
+):
+    auth_service.reset_password(request)
+    return Successfully(code=200, msg="Password reset successfully", status="success")
 
 
-@router.post("/refresh")
-def refresh_token(refresh_token_request: RefreshTokenRequest):
-    try:
-        payload = decode_jwt(refresh_token_request.refresh_token)
-        user_id = payload.get("id")
-        if not user_id:
-            return Failed(status="fail", code=401, msg="Invalid refresh token")
-        new_access_token = sign_jwt(user_id)
-
-        return Successfully(
-            status="success", code=200, msg="Refresh token", data=new_access_token
-        )
-    except jwt.ExpiredSignatureError:
-        return Failed(status="fail", code=401, msg="Refresh token is expired")
-    except jwt.PyJWKError:
-        return Failed(status="fail", code=401, msg="Invalid refresh token")
+@router.post("/confirm-password")
+def confirm_password(
+    request: ConfirmPasswordRequest,
+    auth_service: AuthService = Depends(get_auth_service),
+):
+    auth_service.confirm_password(request)
+    return Successfully(code=200, msg="Password updated successfully", status="success")