83 lines
2.3 KiB
Python
83 lines
2.3 KiB
Python
from fastapi import APIRouter, Depends, HTTPException
|
|
from core.database import fake_users_db
|
|
from typing import List
|
|
from pydantic import BaseModel
|
|
import uuid
|
|
|
|
router = APIRouter()
|
|
|
|
class UserBase(BaseModel):
|
|
username: str
|
|
email: str
|
|
|
|
class UserCreate(UserBase):
|
|
password: str
|
|
|
|
class User(UserBase):
|
|
id: str
|
|
disabled: bool = False
|
|
|
|
@router.post("/users", response_model=User)
|
|
async def create_user(user: UserCreate, db: Session = Depends(get_db)):
|
|
"""Create a new user"""
|
|
if user.username in fake_users_db:
|
|
raise HTTPException(status_code=400, detail="Username already exists")
|
|
|
|
user_id = str(uuid.uuid4())
|
|
new_user = User(
|
|
id=user_id,
|
|
username=user.username,
|
|
email=user.email,
|
|
disabled=False
|
|
)
|
|
fake_users_db[user_id] = new_user
|
|
|
|
return {
|
|
"message": "User created successfully",
|
|
"data": new_user
|
|
}
|
|
|
|
@router.get("/users", response_model=List[User])
|
|
async def read_users(db: Session = Depends(get_db)):
|
|
"""Get all users"""
|
|
return list(fake_users_db.values())
|
|
|
|
@router.get("/users/{user_id}", response_model=User)
|
|
async def read_user(user_id: str, db: Session = Depends(get_db)):
|
|
"""Get a specific user by ID"""
|
|
user = fake_users_db.get(user_id)
|
|
if not user:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
return {
|
|
"message": "User found",
|
|
"data": user
|
|
}
|
|
|
|
@router.put("/users/{user_id}", response_model=User)
|
|
async def update_user(user_id: str, user: UserCreate, db: Session = Depends(get_db)):
|
|
"""Update a user"""
|
|
stored_user = fake_users_db.get(user_id)
|
|
if not stored_user:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
|
|
updated_user = User(
|
|
id=user_id,
|
|
username=user.username,
|
|
email=user.email,
|
|
disabled=stored_user.disabled
|
|
)
|
|
fake_users_db[user_id] = updated_user
|
|
|
|
return {
|
|
"message": "User updated successfully",
|
|
"data": updated_user
|
|
}
|
|
|
|
@router.delete("/users/{user_id}")
|
|
async def delete_user(user_id: str, db: Session = Depends(get_db)):
|
|
"""Delete a user"""
|
|
if user_id not in fake_users_db:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
|
|
del fake_users_db[user_id]
|
|
return {"message": "User deleted successfully"} |