Spaces:
Running
Running
File size: 3,651 Bytes
75e2b6c |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 |
from fastapi import APIRouter, Depends, HTTPException, Body
from pydantic import BaseModel, EmailStr, validator
from typing import Optional
from werkzeug.security import generate_password_hash
from app.database.database_query import DatabaseQuery
from app.middleware.auth import get_current_user
router = APIRouter()
query = DatabaseQuery()
class ProfileUpdateRequest(BaseModel):
email: Optional[EmailStr] = None
password: Optional[str] = None
name: Optional[str] = None
age: Optional[int] = None
@validator('password')
def password_length(cls, v):
if v is not None and len(v) < 6:
raise ValueError('Password must be at least 6 characters')
return v
@validator('age')
def age_range(cls, v):
if v is not None and (v < 13 or v > 120):
raise ValueError('Age must be between 13 and 120')
return v
@router.get('/profile')
async def get_profile(username: str = Depends(get_current_user)):
try:
user = query.get_user_profile(username)
if not user:
raise HTTPException(status_code=404, detail="User not found")
return {
'username': user['username'],
'email': user['email'],
'name': user['name'],
'age': user['age'],
'created_at': user['created_at']
}
except Exception as e:
if isinstance(e, HTTPException):
raise e
raise HTTPException(status_code=500, detail=str(e))
@router.put('/profile')
async def update_profile(
update_data: ProfileUpdateRequest = Body(...),
username: str = Depends(get_current_user)
):
try:
update_fields = {}
if update_data.email:
if not query.is_valid_email(update_data.email):
raise HTTPException(status_code=400, detail="Invalid email format")
update_fields['email'] = update_data.email
if update_data.password:
update_fields['password'] = generate_password_hash(update_data.password)
if update_data.name:
update_fields['name'] = update_data.name
if update_data.age is not None:
update_fields['age'] = update_data.age
if update_fields:
if query.update_user_profile(username, update_fields):
return {"message": "Profile updated successfully"}
return {"message": "No changes made"}
except Exception as e:
if isinstance(e, HTTPException):
raise e
raise HTTPException(status_code=500, detail=str(e))
@router.delete('/profile')
async def delete_account(username: str = Depends(get_current_user)):
try:
if query.delete_user_account(username):
return {"message": "Account deleted successfully"}
raise HTTPException(status_code=404, detail="User not found")
except Exception as e:
if isinstance(e, HTTPException):
raise e
raise HTTPException(status_code=500, detail=str(e))
@router.delete('/delete-account-permanently')
async def delete_account_permanently(username: str = Depends(get_current_user)):
try:
result = query.delete_account_permanently(username)
if result['success']:
return {
'message': 'Account and all associated data deleted successfully',
'details': result['deleted_data']
}
else:
raise HTTPException(status_code=500, detail="Failed to delete account")
except Exception as e:
if isinstance(e, HTTPException):
raise e
raise HTTPException(status_code=500, detail=str(e)) |