Add basic user CRUD endpoints

app/api/router.py

@@ -2,8 +2,10 @@
 
 from app.api.routes.health import router as health_router
 from app.api.routes.tickets import router as tickets_router
+from app.api.routes.users import router as users_router
 from app.core.config import settings
 
 api_router = APIRouter(prefix=settings.api_v1_prefix)
 api_router.include_router(health_router, tags=["health"])
 api_router.include_router(tickets_router, tags=["tickets"])
+api_router.include_router(users_router, tags=["users"])

app/api/routes/users.py

@@ -0,0 +1,80 @@
+from fastapi import APIRouter, Depends, status
+from sqlalchemy.orm import Session
+
+from app.core.errors import AppError
+from app.crud.users import create_user, get_user, get_user_by_email, get_users, update_user
+from app.db.session import get_db
+from app.models.user import User
+from app.schemas.user import UserCreate, UserResponse, UserUpdate
+
+router = APIRouter()
+
+
+def _validate_unique_email(db: Session, email: str, current_user: User | None = None) -> None:
+    existing_user = get_user_by_email(db, email)
+
+    if existing_user is not None and (
+        current_user is None or existing_user.id != current_user.id
+    ):
+        raise AppError("Email already exists.", status.HTTP_409_CONFLICT)
+
+
+@router.post(
+    "/users",
+    response_model=UserResponse,
+    status_code=status.HTTP_201_CREATED,
+    summary="Create user",
+)
+def create_user_endpoint(
+    user_in: UserCreate,
+    db: Session = Depends(get_db),
+) -> UserResponse:
+    _validate_unique_email(db, user_in.email)
+    return create_user(db, user_in)
+
+
+@router.get(
+    "/users",
+    response_model=list[UserResponse],
+    summary="List users",
+)
+def list_users_endpoint(db: Session = Depends(get_db)) -> list[UserResponse]:
+    return get_users(db)
+
+
+@router.get(
+    "/users/{user_id}",
+    response_model=UserResponse,
+    summary="Get user",
+)
+def get_user_endpoint(
+    user_id: int,
+    db: Session = Depends(get_db),
+) -> UserResponse:
+    user = get_user(db, user_id)
+
+    if user is None:
+        raise AppError("User not found.", status.HTTP_404_NOT_FOUND)
+
+    return user
+
+
+@router.patch(
+    "/users/{user_id}",
+    response_model=UserResponse,
+    summary="Update user",
+)
+def update_user_endpoint(
+    user_id: int,
+    user_in: UserUpdate,
+    db: Session = Depends(get_db),
+) -> UserResponse:
+    user = get_user(db, user_id)
+
+    if user is None:
+        raise AppError("User not found.", status.HTTP_404_NOT_FOUND)
+
+    if user_in.email is not None:
+        _validate_unique_email(db, user_in.email, current_user=user)
+
+    return update_user(db, user, user_in)

app/crud/users.py

@@ -0,0 +1,37 @@
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from app.models.user import User
+from app.schemas.user import UserCreate, UserUpdate
+
+
+def create_user(db: Session, user_in: UserCreate) -> User:
+    user = User(**user_in.model_dump())
+    db.add(user)
+    db.commit()
+    db.refresh(user)
+    return user
+
+
+def get_user(db: Session, user_id: int) -> User | None:
+    return db.get(User, user_id)
+
+
+def get_user_by_email(db: Session, email: str) -> User | None:
+    return db.scalar(select(User).where(User.email == email))
+
+
+def get_users(db: Session) -> list[User]:
+    return list(db.scalars(select(User)).all())
+
+
+def update_user(db: Session, user: User, user_in: UserUpdate) -> User:
+    update_data = user_in.model_dump(exclude_unset=True)
+
+    for field, value in update_data.items():
+        setattr(user, field, value)
+
+    db.add(user)
+    db.commit()
+    db.refresh(user)
+    return user

app/schemas/user.py

@@ -0,0 +1,29 @@
+from datetime import datetime
+
+from pydantic import BaseModel, ConfigDict
+
+
+class UserBase(BaseModel):
+    email: str
+    full_name: str
+    role: str = "requester"
+    is_active: bool = True
+
+
+class UserCreate(UserBase):
+    hashed_password: str
+
+
+class UserUpdate(BaseModel):
+    email: str | None = None
+    full_name: str | None = None
+    hashed_password: str | None = None
+    role: str | None = None
+    is_active: bool | None = None
+
+
+class UserResponse(UserBase):
+    model_config = ConfigDict(from_attributes=True)
+
+    id: int
+    created_at: datetime

tests/test_users.py

@@ -0,0 +1,152 @@
+from collections.abc import Generator
+
+import pytest
+from fastapi.testclient import TestClient
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session, sessionmaker
+from sqlalchemy.pool import StaticPool
+
+from app.db.base import Base
+from app.db.session import get_db
+from app.main import app
+import app.models as models  # noqa: F401
+
+
+engine = create_engine(
+    "sqlite://",
+    connect_args={"check_same_thread": False},
+    poolclass=StaticPool,
+)
+
+TestingSessionLocal = sessionmaker(
+    autocommit=False,
+    autoflush=False,
+    bind=engine,
+)
+
+
+@pytest.fixture()
+def db_session() -> Generator[Session, None, None]:
+    Base.metadata.create_all(bind=engine)
+    db = TestingSessionLocal()
+
+    try:
+        yield db
+    finally:
+        db.close()
+        Base.metadata.drop_all(bind=engine)
+
+
+@pytest.fixture()
+def client(db_session: Session) -> Generator[TestClient, None, None]:
+    def override_get_db() -> Generator[Session, None, None]:
+        yield db_session
+
+    app.dependency_overrides[get_db] = override_get_db
+
+    with TestClient(app) as test_client:
+        yield test_client
+
+    app.dependency_overrides.clear()
+
+
+def create_user_payload(email: str = "user@example.com") -> dict[str, object]:
+    return {
+        "email": email,
+        "full_name": "Test User",
+        "hashed_password": "not-a-real-password-hash",
+    }
+
+
+def create_user(client: TestClient, email: str = "user@example.com") -> dict[str, object]:
+    response = client.post("/api/v1/users", json=create_user_payload(email=email))
+    assert response.status_code == 201
+    return response.json()
+
+
+def test_create_user_returns_created_user_without_hashed_password(client: TestClient) -> None:
+    response = client.post("/api/v1/users", json=create_user_payload())
+
+    assert response.status_code == 201
+
+    data = response.json()
+    assert data["id"] == 1
+    assert data["email"] == "user@example.com"
+    assert data["full_name"] == "Test User"
+    assert data["role"] == "requester"
+    assert data["is_active"] is True
+    assert "created_at" in data
+    assert "hashed_password" not in data
+
+
+def test_list_users_returns_created_users(client: TestClient) -> None:
+    create_user(client)
+
+    response = client.get("/api/v1/users")
+
+    assert response.status_code == 200
+
+    data = response.json()
+    assert len(data) == 1
+    assert data[0]["email"] == "user@example.com"
+    assert "hashed_password" not in data[0]
+
+
+def test_get_user_returns_existing_user(client: TestClient) -> None:
+    user = create_user(client)
+
+    response = client.get(f"/api/v1/users/{user['id']}")
+
+    assert response.status_code == 200
+    assert response.json()["id"] == user["id"]
+    assert response.json()["email"] == "user@example.com"
+
+
+def test_get_user_returns_not_found_for_missing_user(client: TestClient) -> None:
+    response = client.get("/api/v1/users/999")
+
+    assert response.status_code == 404
+    assert response.json()["error"]["message"] == "User not found."
+
+
+def test_update_user_partially_updates_existing_user(client: TestClient) -> None:
+    user = create_user(client)
+
+    response = client.patch(
+        f"/api/v1/users/{user['id']}",
+        json={
+            "full_name": "Updated User",
+            "is_active": False,
+        },
+    )
+
+    assert response.status_code == 200
+
+    data = response.json()
+    assert data["id"] == user["id"]
+    assert data["email"] == "user@example.com"
+    assert data["full_name"] == "Updated User"
+    assert data["is_active"] is False
+    assert "hashed_password" not in data
+
+
+def test_create_user_returns_conflict_for_duplicate_email(client: TestClient) -> None:
+    create_user(client)
+
+    response = client.post("/api/v1/users", json=create_user_payload())
+
+    assert response.status_code == 409
+    assert response.json()["error"]["message"] == "Email already exists."
+
+
+def test_update_user_returns_conflict_for_duplicate_email(client: TestClient) -> None:
+    create_user(client, email="first@example.com")
+    second_user = create_user(client, email="second@example.com")
+
+    response = client.patch(
+        f"/api/v1/users/{second_user['id']}",
+        json={"email": "first@example.com"},
+    )
+
+    assert response.status_code == 409
+    assert response.json()["error"]["message"] == "Email already exists."