Spaces:
Sleeping
Sleeping
File size: 3,337 Bytes
96f792c |
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 |
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from typing import List
from app.database import get_db
from app import models, schemas
from app.auth import (
get_password_hash,
verify_password,
get_current_user,
)
router = APIRouter(tags=["user"])
@router.post("/auth/register", response_model=schemas.UserResponse, status_code=status.HTTP_201_CREATED)
def register(user: schemas.UserCreate, db: Session = Depends(get_db)):
db_user = db.query(models.User).filter(models.User.username == user.username).first()
if db_user:
raise HTTPException(status_code=400, detail="Username already registered")
db_email = db.query(models.User).filter(models.User.email == user.email).first()
if db_email:
raise HTTPException(status_code=400, detail="Email already registered")
hashed_password = get_password_hash(user.password)
db_user = models.User(
username=user.username,
email=user.email,
hashed_password=hashed_password,
role=models.UserRole.USER
)
db.add(db_user)
db.commit()
db.refresh(db_user)
return db_user
@router.post("/auth/login", response_model=schemas.UserResponse)
def login(credentials: schemas.UserLogin, db: Session = Depends(get_db)):
user = db.query(models.User).filter(models.User.username == credentials.username).first()
if not user:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Incorrect username or password"
)
if not verify_password(credentials.password, str(user.hashed_password)):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Incorrect username or password"
)
return user
@router.get("/user/me", response_model=schemas.UserResponse)
def get_current_user_info(current_user: models.User = Depends(get_current_user)):
return current_user
@router.post("/user/health-issues", response_model=schemas.HealthIssueResponse, status_code=status.HTTP_201_CREATED)
def add_health_issue(
health_issue: schemas.HealthIssueCreate,
current_user: models.User = Depends(get_current_user),
db: Session = Depends(get_db)
):
db_health_issue = models.HealthIssue(
user_id=current_user.id,
**health_issue.dict()
)
db.add(db_health_issue)
db.commit()
db.refresh(db_health_issue)
return db_health_issue
@router.get("/user/health-issues", response_model=List[schemas.HealthIssueResponse])
def get_health_issues(
current_user: models.User = Depends(get_current_user),
db: Session = Depends(get_db)
):
return db.query(models.HealthIssue).filter(models.HealthIssue.user_id == current_user.id).all()
@router.delete("/user/health-issues/{issue_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_health_issue(
issue_id: int,
current_user: models.User = Depends(get_current_user),
db: Session = Depends(get_db)
):
health_issue = db.query(models.HealthIssue).filter(
models.HealthIssue.id == issue_id,
models.HealthIssue.user_id == current_user.id
).first()
if not health_issue:
raise HTTPException(status_code=404, detail="Health issue not found")
db.delete(health_issue)
db.commit()
return None
|