64 lines
2.5 KiB
Python
64 lines
2.5 KiB
Python
from typing import List
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.db import get_db
|
|
from app.models.user import User as UserModel
|
|
from app.schemas.user import User, UserCreate, UserUpdate
|
|
|
|
router = APIRouter()
|
|
|
|
@router.post("/users", response_model=User, status_code=201)
|
|
def create_user(user: UserCreate, db: Session = Depends(get_db)):
|
|
db_user = UserModel(**user.dict())
|
|
db.add(db_user)
|
|
db.commit()
|
|
db.refresh(db_user)
|
|
return db_user
|
|
|
|
@router.get("/users", response_model=List[User])
|
|
def get_users(db: Session = Depends(get_db)):
|
|
users = db.query(UserModel).all()
|
|
return users
|
|
|
|
@router.get("/users/{user_id}", response_model=User)
|
|
def get_user(user_id: int, db: Session = Depends(get_db)):
|
|
db_user = db.query(UserModel).filter(UserModel.id == user_id).first()
|
|
if not db_user:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
return db_user
|
|
|
|
@router.put("/users/{user_id}", response_model=User)
|
|
def update_user(user_id: int, user: UserUpdate, db: Session = Depends(get_db)):
|
|
db_user = db.query(UserModel).filter(UserModel.id == user_id).first()
|
|
if not db_user:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
update_data = user.dict(exclude_unset=True)
|
|
for key, value in update_data.items():
|
|
setattr(db_user, key, value)
|
|
db.commit()
|
|
db.refresh(db_user)
|
|
return db_user
|
|
|
|
@router.delete("/users/{user_id}", status_code=204)
|
|
def delete_user(user_id: int, db: Session = Depends(get_db)):
|
|
db_user = db.query(UserModel).filter(UserModel.id == user_id).first()
|
|
if not db_user:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
db.delete(db_user)
|
|
db.commit()
|
|
return {"message": "User deleted successfully"}
|
|
```
|
|
|
|
|
|
The endpoints are:
|
|
|
|
1. `POST /users`: Creates a new user based on the provided `UserCreate` model data.
|
|
2. `GET /users`: Retrieves a list of all users.
|
|
3. `GET /users/{user_id}`: Retrieves a specific user by ID.
|
|
4. `PUT /users/{user_id}`: Updates an existing user with the provided `UserUpdate` model data.
|
|
5. `DELETE /users/{user_id}`: Deletes a user by ID.
|
|
|
|
The code uses dependency injection to obtain a SQLAlchemy database session from the `get_db` function (assumed to be defined elsewhere). It also handles HTTP exceptions for cases where a user is not found (404 Not Found).
|
|
|
|
Note that this is a basic implementation, and you may need to add additional functionality, such as authentication, authorization, and input validation, depending on your application's requirements. |