274 lines
8.6 KiB
Python
274 lines
8.6 KiB
Python
"""
|
|
@router CustomerProfileRouter (Backend/app/api/v1/routers/CustomerProfileRouter.py)
|
|
@purpose Database-driven profile management and address book CRUD endpoints for e-commerce customers, protected by JWT access token validation.
|
|
"""
|
|
from fastapi import APIRouter, Depends, HTTPException, status
|
|
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
|
|
from pydantic import BaseModel, EmailStr
|
|
from typing import List, Optional
|
|
from sqlalchemy.orm import Session
|
|
import ulid
|
|
from datetime import datetime, timezone
|
|
|
|
from app.core.database.db_session import get_db
|
|
from app.core.Token import verify_access_token
|
|
from app.models.EcomCustomerModel import EcomCustomer, CustomerAddress
|
|
|
|
router = APIRouter(prefix="/api/v1/customer/profile", tags=["Customer Profile & Addresses"])
|
|
|
|
security = HTTPBearer()
|
|
|
|
# --- Authentication Dependency ---
|
|
|
|
def get_current_customer(credentials: HTTPAuthorizationCredentials = Depends(security), db: Session = Depends(get_db)) -> EcomCustomer:
|
|
token = credentials.credentials
|
|
try:
|
|
payload = verify_access_token(token)
|
|
except Exception as e:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail=f"Token validation failed: {str(e)}",
|
|
headers={"WWW-Authenticate": "Bearer"}
|
|
)
|
|
|
|
customer_id = payload.get("sub")
|
|
role = payload.get("role")
|
|
|
|
if not customer_id or role != "customer":
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Invalid authorization credentials for customer",
|
|
headers={"WWW-Authenticate": "Bearer"}
|
|
)
|
|
|
|
customer = db.query(EcomCustomer).filter(EcomCustomer.customer_id == customer_id).first()
|
|
if not customer:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Customer not found"
|
|
)
|
|
return customer
|
|
|
|
# --- Schemas ---
|
|
|
|
class ProfileUpdatePayload(BaseModel):
|
|
first_name: str
|
|
last_name: str
|
|
phone: Optional[str] = None
|
|
|
|
class ProfileResponse(BaseModel):
|
|
customer_id: str
|
|
email: str
|
|
first_name: str
|
|
last_name: str
|
|
phone: Optional[str] = None
|
|
profile_picture: Optional[str] = None
|
|
|
|
class AddressSchema(BaseModel):
|
|
address_id: Optional[str] = None
|
|
address_type: str = "SHIPPING"
|
|
full_name: str
|
|
phone: str
|
|
street_address: str
|
|
city: str
|
|
state: str
|
|
pincode: str
|
|
is_default: bool = False
|
|
|
|
# --- Profile Endpoints ---
|
|
|
|
@router.get("", response_model=ProfileResponse)
|
|
def get_customer_profile(customer: EcomCustomer = Depends(get_current_customer)):
|
|
"""
|
|
Get current logged-in customer's profile attributes.
|
|
"""
|
|
return ProfileResponse(
|
|
customer_id=customer.customer_id,
|
|
email=customer.email,
|
|
first_name=customer.first_name,
|
|
last_name=customer.last_name,
|
|
phone=customer.phone,
|
|
profile_picture=customer.profile_picture
|
|
)
|
|
|
|
@router.put("", response_model=ProfileResponse)
|
|
def update_customer_profile(
|
|
payload: ProfileUpdatePayload,
|
|
customer: EcomCustomer = Depends(get_current_customer),
|
|
db: Session = Depends(get_db)
|
|
):
|
|
"""
|
|
Update profile details for the logged-in customer.
|
|
"""
|
|
customer.first_name = payload.first_name
|
|
customer.last_name = payload.last_name
|
|
customer.phone = payload.phone
|
|
db.commit()
|
|
db.refresh(customer)
|
|
return ProfileResponse(
|
|
customer_id=customer.customer_id,
|
|
email=customer.email,
|
|
first_name=customer.first_name,
|
|
last_name=customer.last_name,
|
|
phone=customer.phone,
|
|
profile_picture=customer.profile_picture
|
|
)
|
|
|
|
# --- Address Book Endpoints ---
|
|
|
|
@router.get("/addresses", response_model=List[AddressSchema])
|
|
def get_customer_addresses(
|
|
customer: EcomCustomer = Depends(get_current_customer),
|
|
db: Session = Depends(get_db)
|
|
):
|
|
"""
|
|
List all saved shipping/billing addresses for the customer.
|
|
"""
|
|
addresses = db.query(CustomerAddress).filter(CustomerAddress.customer_id == customer.customer_id).all()
|
|
return [
|
|
AddressSchema(
|
|
address_id=a.address_id,
|
|
address_type=a.address_type,
|
|
full_name=a.full_name,
|
|
phone=a.phone,
|
|
street_address=a.street_address,
|
|
city=a.city,
|
|
state=a.state,
|
|
pincode=a.pincode,
|
|
is_default=a.is_default
|
|
) for a in addresses
|
|
]
|
|
|
|
@router.post("/addresses", response_model=AddressSchema)
|
|
def add_customer_address(
|
|
payload: AddressSchema,
|
|
customer: EcomCustomer = Depends(get_current_customer),
|
|
db: Session = Depends(get_db)
|
|
):
|
|
"""
|
|
Save a new address to the customer's address book.
|
|
"""
|
|
# Enforce single default constraint: If new is default, set all others to false.
|
|
if payload.is_default:
|
|
db.query(CustomerAddress).filter(
|
|
CustomerAddress.customer_id == customer.customer_id
|
|
).update({CustomerAddress.is_default: False})
|
|
db.commit()
|
|
|
|
# If this is the customer's first address, force it to be default
|
|
existing_count = db.query(CustomerAddress).filter(CustomerAddress.customer_id == customer.customer_id).count()
|
|
is_default_flag = payload.is_default if existing_count > 0 else True
|
|
|
|
new_address = CustomerAddress(
|
|
address_id=str(ulid.ULID()),
|
|
customer_id=customer.customer_id,
|
|
address_type=payload.address_type,
|
|
full_name=payload.full_name,
|
|
phone=payload.phone,
|
|
street_address=payload.street_address,
|
|
city=payload.city,
|
|
state=payload.state,
|
|
pincode=payload.pincode,
|
|
is_default=is_default_flag
|
|
)
|
|
db.add(new_address)
|
|
db.commit()
|
|
db.refresh(new_address)
|
|
|
|
return AddressSchema(
|
|
address_id=new_address.address_id,
|
|
address_type=new_address.address_type,
|
|
full_name=new_address.full_name,
|
|
phone=new_address.phone,
|
|
street_address=new_address.street_address,
|
|
city=new_address.city,
|
|
state=new_address.state,
|
|
pincode=new_address.pincode,
|
|
is_default=new_address.is_default
|
|
)
|
|
|
|
@router.put("/addresses/{address_id}", response_model=AddressSchema)
|
|
def update_customer_address(
|
|
address_id: str,
|
|
payload: AddressSchema,
|
|
customer: EcomCustomer = Depends(get_current_customer),
|
|
db: Session = Depends(get_db)
|
|
):
|
|
"""
|
|
Modify an existing address in the customer's address book.
|
|
"""
|
|
address = db.query(CustomerAddress).filter(
|
|
CustomerAddress.address_id == address_id,
|
|
CustomerAddress.customer_id == customer.customer_id
|
|
).first()
|
|
|
|
if not address:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Address record not found"
|
|
)
|
|
|
|
if payload.is_default and not address.is_default:
|
|
db.query(CustomerAddress).filter(
|
|
CustomerAddress.customer_id == customer.customer_id
|
|
).update({CustomerAddress.is_default: False})
|
|
db.commit()
|
|
|
|
address.address_type = payload.address_type
|
|
address.full_name = payload.full_name
|
|
address.phone = payload.phone
|
|
address.street_address = payload.street_address
|
|
address.city = payload.city
|
|
address.state = payload.state
|
|
address.pincode = payload.pincode
|
|
address.is_default = payload.is_default
|
|
|
|
db.commit()
|
|
db.refresh(address)
|
|
|
|
return AddressSchema(
|
|
address_id=address.address_id,
|
|
address_type=address.address_type,
|
|
full_name=address.full_name,
|
|
phone=address.phone,
|
|
street_address=address.street_address,
|
|
city=address.city,
|
|
state=address.state,
|
|
pincode=address.pincode,
|
|
is_default=address.is_default
|
|
)
|
|
|
|
@router.delete("/addresses/{address_id}")
|
|
def delete_customer_address(
|
|
address_id: str,
|
|
customer: EcomCustomer = Depends(get_current_customer),
|
|
db: Session = Depends(get_db)
|
|
):
|
|
"""
|
|
Remove an address from the customer's address book.
|
|
"""
|
|
address = db.query(CustomerAddress).filter(
|
|
CustomerAddress.address_id == address_id,
|
|
CustomerAddress.customer_id == customer.customer_id
|
|
).first()
|
|
|
|
if not address:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Address record not found"
|
|
)
|
|
|
|
was_default = address.is_default
|
|
db.delete(address)
|
|
db.commit()
|
|
|
|
# If we deleted the default address, set another address to default if any exists
|
|
if was_default:
|
|
next_address = db.query(CustomerAddress).filter(
|
|
CustomerAddress.customer_id == customer.customer_id
|
|
).first()
|
|
if next_address:
|
|
next_address.is_default = True
|
|
db.commit()
|
|
|
|
return {"detail": "Address deleted successfully"}
|