Add basic ticket CRUD endpoints

app/api/router.py

@@ -1,7 +1,9 @@
 from fastapi import APIRouter
 
 from app.api.routes.health import router as health_router
+from app.api.routes.tickets import router as tickets_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"])

app/api/routes/tickets.py

@@ -0,0 +1,81 @@
+from fastapi import APIRouter, Depends, status
+from sqlalchemy.orm import Session
+
+from app.crud.tickets import create_ticket, get_ticket, get_tickets, update_ticket
+from app.core.errors import AppError
+from app.db.session import get_db
+from app.models.user import User
+from app.schemas.ticket import TicketCreate, TicketResponse, TicketUpdate
+
+router = APIRouter()
+
+
+def _get_user_or_none(db: Session, user_id: int) -> User | None:
+    return db.get(User, user_id)
+
+
+def _validate_ticket_users(db: Session, requester_id: int | None, assignee_id: int | None) -> None:
+    if requester_id is not None and _get_user_or_none(db, requester_id) is None:
+        raise AppError("Requester not found.", status.HTTP_404_NOT_FOUND)
+
+    if assignee_id is not None and _get_user_or_none(db, assignee_id) is None:
+        raise AppError("Assignee not found.", status.HTTP_404_NOT_FOUND)
+
+
+@router.post(
+    "/tickets",
+    response_model=TicketResponse,
+    status_code=status.HTTP_201_CREATED,
+    summary="Create ticket",
+)
+def create_ticket_endpoint(
+    ticket_in: TicketCreate,
+    db: Session = Depends(get_db),
+) -> TicketResponse:
+    _validate_ticket_users(db, ticket_in.requester_id, ticket_in.assignee_id)
+    return create_ticket(db, ticket_in)
+
+
+@router.get(
+    "/tickets",
+    response_model=list[TicketResponse],
+    summary="List tickets",
+)
+def list_tickets_endpoint(db: Session = Depends(get_db)) -> list[TicketResponse]:
+    return get_tickets(db)
+
+
+@router.get(
+    "/tickets/{ticket_id}",
+    response_model=TicketResponse,
+    summary="Get ticket",
+)
+def get_ticket_endpoint(
+    ticket_id: int,
+    db: Session = Depends(get_db),
+) -> TicketResponse:
+    ticket = get_ticket(db, ticket_id)
+
+    if ticket is None:
+        raise AppError("Ticket not found.", status.HTTP_404_NOT_FOUND)
+
+    return ticket
+
+
+@router.patch(
+    "/tickets/{ticket_id}",
+    response_model=TicketResponse,
+    summary="Update ticket",
+)
+def update_ticket_endpoint(
+    ticket_id: int,
+    ticket_in: TicketUpdate,
+    db: Session = Depends(get_db),
+) -> TicketResponse:
+    ticket = get_ticket(db, ticket_id)
+
+    if ticket is None:
+        raise AppError("Ticket not found.", status.HTTP_404_NOT_FOUND)
+
+    _validate_ticket_users(db, ticket_in.requester_id, ticket_in.assignee_id)
+    return update_ticket(db, ticket, ticket_in)

app/crud/__init__.py

@@ -0,0 +1 @@
+"""Database operations for application resources."""

app/crud/tickets.py

@@ -0,0 +1,33 @@
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from app.models.ticket import Ticket
+from app.schemas.ticket import TicketCreate, TicketUpdate
+
+
+def create_ticket(db: Session, ticket_in: TicketCreate) -> Ticket:
+    ticket = Ticket(**ticket_in.model_dump())
+    db.add(ticket)
+    db.commit()
+    db.refresh(ticket)
+    return ticket
+
+
+def get_ticket(db: Session, ticket_id: int) -> Ticket | None:
+    return db.get(Ticket, ticket_id)
+
+
+def get_tickets(db: Session) -> list[Ticket]:
+    return list(db.scalars(select(Ticket)).all())
+
+
+def update_ticket(db: Session, ticket: Ticket, ticket_in: TicketUpdate) -> Ticket:
+    update_data = ticket_in.model_dump(exclude_unset=True)
+
+    for field, value in update_data.items():
+        setattr(ticket, field, value)
+
+    db.add(ticket)
+    db.commit()
+    db.refresh(ticket)
+    return ticket

app/schemas/ticket.py

@@ -0,0 +1,33 @@
+from datetime import datetime
+
+from pydantic import BaseModel, ConfigDict
+
+
+class TicketBase(BaseModel):
+    title: str
+    description: str | None = None
+    status: str = "open"
+    priority: str = "medium"
+    requester_id: int
+    assignee_id: int | None = None
+
+
+class TicketCreate(TicketBase):
+    pass
+
+
+class TicketUpdate(BaseModel):
+    title: str | None = None
+    description: str | None = None
+    status: str | None = None
+    priority: str | None = None
+    requester_id: int | None = None
+    assignee_id: int | None = None
+
+
+class TicketResponse(TicketBase):
+    model_config = ConfigDict(from_attributes=True)
+
+    id: int
+    created_at: datetime
+    updated_at: datetime | None = None

tests/test_tickets.py

@@ -0,0 +1,176 @@
+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
+from app.models.user import User
+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(db: Session, email: str = "requester@example.com") -> User:
+    user = User(
+        email=email,
+        full_name="Test User",
+        hashed_password="not-a-real-password-hash",
+    )
+    db.add(user)
+    db.commit()
+    db.refresh(user)
+    return user
+
+
+def create_ticket_payload(requester_id: int, assignee_id: int | None = None) -> dict[str, object]:
+    return {
+        "title": "Cannot access dashboard",
+        "description": "The dashboard returns an error after login.",
+        "priority": "high",
+        "requester_id": requester_id,
+        "assignee_id": assignee_id,
+    }
+
+
+def test_create_ticket_returns_created_ticket(client: TestClient, db_session: Session) -> None:
+    requester = create_user(db_session)
+
+    response = client.post(
+        "/api/v1/tickets",
+        json=create_ticket_payload(requester.id),
+    )
+
+    assert response.status_code == 201
+
+    data = response.json()
+    assert data["id"] == 1
+    assert data["title"] == "Cannot access dashboard"
+    assert data["description"] == "The dashboard returns an error after login."
+    assert data["status"] == "open"
+    assert data["priority"] == "high"
+    assert data["requester_id"] == requester.id
+    assert data["assignee_id"] is None
+    assert "created_at" in data
+
+
+def test_list_tickets_returns_created_tickets(client: TestClient, db_session: Session) -> None:
+    requester = create_user(db_session)
+    client.post("/api/v1/tickets", json=create_ticket_payload(requester.id))
+
+    response = client.get("/api/v1/tickets")
+
+    assert response.status_code == 200
+
+    data = response.json()
+    assert len(data) == 1
+    assert data[0]["title"] == "Cannot access dashboard"
+
+
+def test_get_ticket_returns_existing_ticket(client: TestClient, db_session: Session) -> None:
+    requester = create_user(db_session)
+    create_response = client.post("/api/v1/tickets", json=create_ticket_payload(requester.id))
+    ticket_id = create_response.json()["id"]
+
+    response = client.get(f"/api/v1/tickets/{ticket_id}")
+
+    assert response.status_code == 200
+    assert response.json()["id"] == ticket_id
+
+
+def test_get_ticket_returns_not_found_for_missing_ticket(client: TestClient) -> None:
+    response = client.get("/api/v1/tickets/999")
+
+    assert response.status_code == 404
+    assert response.json()["error"]["message"] == "Ticket not found."
+
+
+def test_update_ticket_partially_updates_existing_ticket(
+    client: TestClient,
+    db_session: Session,
+) -> None:
+    requester = create_user(db_session)
+    assignee = create_user(db_session, email="assignee@example.com")
+    create_response = client.post("/api/v1/tickets", json=create_ticket_payload(requester.id))
+    ticket_id = create_response.json()["id"]
+
+    response = client.patch(
+        f"/api/v1/tickets/{ticket_id}",
+        json={
+            "status": "in_progress",
+            "assignee_id": assignee.id,
+        },
+    )
+
+    assert response.status_code == 200
+
+    data = response.json()
+    assert data["id"] == ticket_id
+    assert data["status"] == "in_progress"
+    assert data["assignee_id"] == assignee.id
+    assert data["title"] == "Cannot access dashboard"
+
+
+def test_create_ticket_returns_not_found_for_missing_requester(client: TestClient) -> None:
+    response = client.post(
+        "/api/v1/tickets",
+        json=create_ticket_payload(requester_id=999),
+    )
+
+    assert response.status_code == 404
+    assert response.json()["error"]["message"] == "Requester not found."
+
+
+def test_create_ticket_returns_not_found_for_missing_assignee(
+    client: TestClient,
+    db_session: Session,
+) -> None:
+    requester = create_user(db_session)
+
+    response = client.post(
+        "/api/v1/tickets",
+        json=create_ticket_payload(requester_id=requester.id, assignee_id=999),
+    )
+
+    assert response.status_code == 404
+    assert response.json()["error"]["message"] == "Assignee not found."