126 lines
3.9 KiB
Python
126 lines
3.9 KiB
Python
from datetime import datetime
|
|
|
|
from fastapi import HTTPException, status
|
|
from sqlalchemy.orm import Session, selectinload
|
|
|
|
from app.models.user import User
|
|
from app.models.user_profile import UserProfile
|
|
from app.schemas.user import UserCreate, UserUpdate
|
|
from app.security import hash_password
|
|
|
|
|
|
def get_user_list(db: Session, status: str = None):
|
|
query = db.query(User).options(selectinload(User.profile)).filter(User.deleted_at.is_(None))
|
|
if status is not None:
|
|
query = query.filter(User.status == status)
|
|
return query.order_by(User.created_at.desc()).all()
|
|
|
|
|
|
def get_user_by_id(db: Session, id: str):
|
|
user = db.query(User).options(selectinload(User.profile)).filter(User.id == id, User.deleted_at.is_(None)).first()
|
|
if not user:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail={"message": "The item does not exist"},
|
|
)
|
|
return user
|
|
|
|
|
|
def create_user(db: Session, data: UserCreate):
|
|
existing = db.query(User).filter(
|
|
(User.email == data.email) | (User.username == data.username)
|
|
).first()
|
|
if existing:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_409_CONFLICT,
|
|
detail={"message": "Email or username already registered"},
|
|
)
|
|
|
|
user = User(
|
|
email=data.email,
|
|
username=data.username,
|
|
password=hash_password(data.password),
|
|
)
|
|
db.add(user)
|
|
db.flush()
|
|
|
|
try:
|
|
profile_data = {
|
|
"user_id": user.id,
|
|
"full_name": data.full_name,
|
|
"phone": data.phone,
|
|
}
|
|
if data.address is not None:
|
|
profile_data["address"] = data.address
|
|
if data.pob is not None:
|
|
profile_data["pob"] = data.pob
|
|
if data.dob is not None:
|
|
profile_data["dob"] = data.dob
|
|
profile = UserProfile(**profile_data)
|
|
db.add(profile)
|
|
db.commit()
|
|
except:
|
|
db.rollback()
|
|
raise
|
|
|
|
return user
|
|
|
|
|
|
def update_user(db: Session, id: str, data: UserUpdate):
|
|
user = db.query(User).filter(User.id == id, User.deleted_at.is_(None)).first()
|
|
if not user:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail={"message": "The item does not exist"},
|
|
)
|
|
|
|
update_data = data.model_dump(exclude_unset=True)
|
|
|
|
profile_fields = ["full_name", "phone", "address", "pob", "dob"]
|
|
user_fields = {k: v for k, v in update_data.items() if k not in profile_fields}
|
|
profile_update = {k: v for k, v in update_data.items() if k in profile_fields}
|
|
|
|
if user_fields.get("email"):
|
|
dup = db.query(User).filter(User.email == user_fields["email"], User.id != id).first()
|
|
if dup:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_409_CONFLICT,
|
|
detail={"message": "Email already used"},
|
|
)
|
|
|
|
if user_fields.get("username"):
|
|
dup = db.query(User).filter(User.username == user_fields["username"], User.id != id).first()
|
|
if dup:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_409_CONFLICT,
|
|
detail={"message": "Username already used"},
|
|
)
|
|
|
|
if "password" in user_fields:
|
|
user_fields["password"] = hash_password(user_fields["password"])
|
|
|
|
for field, value in user_fields.items():
|
|
setattr(user, field, value)
|
|
|
|
if profile_update:
|
|
profile = db.query(UserProfile).filter(UserProfile.user_id == id).first()
|
|
if profile:
|
|
for field, value in profile_update.items():
|
|
setattr(profile, field, value)
|
|
|
|
db.commit()
|
|
db.refresh(user)
|
|
|
|
return user
|
|
|
|
|
|
def delete_user(db: Session, id: str):
|
|
user = db.query(User).filter(User.id == id, User.deleted_at.is_(None)).first()
|
|
if not user:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail={"message": "The item does not exist"},
|
|
)
|
|
|
|
user.deleted_at = datetime.now()
|
|
db.commit() |