Zhi
f60dc68b22
- app/api/deps.py: shared auth dependencies - app/api/routers/auth.py: login, me - app/api/routers/issues.py: CRUD, transition, assign, relations, tags, batch, search - app/api/routers/projects.py: CRUD, members, worklog summary - app/api/routers/users.py: CRUD, worklogs - app/api/routers/comments.py: CRUD - app/api/routers/webhooks.py: CRUD, logs, retry - app/api/routers/misc.py: API keys, activity, milestones, notifications, worklogs, export, dashboard - main.py: 1165 lines → 51 lines - Version bump to 0.2.0
81 lines
2.7 KiB
Python
81 lines
2.7 KiB
Python
"""Users router."""
|
|
from typing import List
|
|
from fastapi import APIRouter, Depends, HTTPException, status
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.core.config import get_db
|
|
from app.models import models
|
|
from app.schemas import schemas
|
|
from app.api.deps import get_password_hash
|
|
|
|
router = APIRouter(prefix="/users", tags=["Users"])
|
|
|
|
|
|
@router.post("", response_model=schemas.UserResponse, status_code=status.HTTP_201_CREATED)
|
|
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
|
|
existing = db.query(models.User).filter(
|
|
(models.User.username == user.username) | (models.User.email == user.email)
|
|
).first()
|
|
if existing:
|
|
raise HTTPException(status_code=400, detail="Username or email already exists")
|
|
hashed_password = get_password_hash(user.password) if user.password else None
|
|
db_user = models.User(
|
|
username=user.username, email=user.email, full_name=user.full_name,
|
|
hashed_password=hashed_password, is_admin=user.is_admin
|
|
)
|
|
db.add(db_user)
|
|
db.commit()
|
|
db.refresh(db_user)
|
|
return db_user
|
|
|
|
|
|
@router.get("", response_model=List[schemas.UserResponse])
|
|
def list_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
|
|
return db.query(models.User).offset(skip).limit(limit).all()
|
|
|
|
|
|
@router.get("/{user_id}", response_model=schemas.UserResponse)
|
|
def get_user(user_id: int, db: Session = Depends(get_db)):
|
|
user = db.query(models.User).filter(models.User.id == user_id).first()
|
|
if not user:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
return user
|
|
|
|
|
|
@router.patch("/{user_id}", response_model=schemas.UserResponse)
|
|
def update_user(user_id: int, db: Session = Depends(get_db), full_name: str = None, email: str = None):
|
|
user = db.query(models.User).filter(models.User.id == user_id).first()
|
|
if not user:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
if full_name is not None:
|
|
user.full_name = full_name
|
|
if email is not None:
|
|
user.email = email
|
|
db.commit()
|
|
db.refresh(user)
|
|
return user
|
|
|
|
|
|
# ---- User worklogs ----
|
|
|
|
from app.models.worklog import WorkLog
|
|
from pydantic import BaseModel
|
|
from datetime import datetime
|
|
|
|
|
|
class WorkLogResponse(BaseModel):
|
|
id: int
|
|
issue_id: int
|
|
user_id: int
|
|
hours: float
|
|
description: str | None = None
|
|
logged_date: datetime
|
|
created_at: datetime
|
|
class Config:
|
|
from_attributes = True
|
|
|
|
|
|
@router.get("/{user_id}/worklogs", response_model=List[WorkLogResponse])
|
|
def list_user_worklogs(user_id: int, limit: int = 50, db: Session = Depends(get_db)):
|
|
return db.query(WorkLog).filter(WorkLog.user_id == user_id).order_by(WorkLog.logged_date.desc()).limit(limit).all()
|