150 lines
4.3 KiB
Python
150 lines
4.3 KiB
Python
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
from sqlalchemy import text
|
|
from sqlalchemy.orm import Session
|
|
from typing import Any
|
|
|
|
from app.database import get_session
|
|
from app.auth import get_current_user
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
def _row(r) -> dict:
|
|
return dict(r._mapping)
|
|
|
|
|
|
@router.get("")
|
|
def list_lists(
|
|
include_closed: bool = Query(default=False),
|
|
session: Session = Depends(get_session),
|
|
_: str = Depends(get_current_user),
|
|
) -> list[dict]:
|
|
where = "" if include_closed else "WHERE l.closed_at IS NULL"
|
|
order = "ORDER BY l.closed_at IS NOT NULL, l.created_at DESC" if include_closed else "ORDER BY l.created_at DESC"
|
|
rows = session.execute(text(f"""
|
|
SELECT
|
|
l.*,
|
|
COUNT(i.id) AS item_count,
|
|
COUNT(i.id) FILTER (WHERE i.checked) AS checked_count
|
|
FROM lk.lists l
|
|
LEFT JOIN lk.list_items i ON i.list_id = l.id
|
|
{where}
|
|
GROUP BY l.id
|
|
{order}
|
|
""")).fetchall()
|
|
return [_row(r) for r in rows]
|
|
|
|
|
|
@router.post("", status_code=201)
|
|
def create_list(
|
|
body: dict[str, Any],
|
|
session: Session = Depends(get_session),
|
|
_: str = Depends(get_current_user),
|
|
) -> dict:
|
|
row = session.execute(text("""
|
|
INSERT INTO lk.lists (name, type, t_shirt_size, notes)
|
|
VALUES (:name, :type, :t_shirt_size, :notes)
|
|
RETURNING *
|
|
"""), {
|
|
"name": body.get("name"),
|
|
"type": body.get("type"),
|
|
"t_shirt_size": body.get("t_shirt_size"),
|
|
"notes": body.get("notes"),
|
|
}).fetchone()
|
|
session.commit()
|
|
return _row(row)
|
|
|
|
|
|
@router.post("/{id}/close")
|
|
def close_list(
|
|
id: int,
|
|
session: Session = Depends(get_session),
|
|
_: str = Depends(get_current_user),
|
|
) -> dict:
|
|
result = session.execute(text("""
|
|
UPDATE lk.lists SET status = 'closed', closed_at = now()
|
|
WHERE id = :id RETURNING *
|
|
"""), {"id": id}).fetchone()
|
|
session.commit()
|
|
if result is None:
|
|
raise HTTPException(status_code=404, detail="Not found")
|
|
return _row(result)
|
|
|
|
|
|
@router.post("/{id}/reopen")
|
|
def reopen_list(
|
|
id: int,
|
|
session: Session = Depends(get_session),
|
|
_: str = Depends(get_current_user),
|
|
) -> dict:
|
|
result = session.execute(text("""
|
|
UPDATE lk.lists SET status = 'open', closed_at = NULL
|
|
WHERE id = :id RETURNING *
|
|
"""), {"id": id}).fetchone()
|
|
session.commit()
|
|
if result is None:
|
|
raise HTTPException(status_code=404, detail="Not found")
|
|
return _row(result)
|
|
|
|
|
|
@router.delete("/{id}", status_code=204)
|
|
def delete_list(
|
|
id: int,
|
|
session: Session = Depends(get_session),
|
|
_: str = Depends(get_current_user),
|
|
):
|
|
session.execute(text("DELETE FROM lk.lists WHERE id = :id"), {"id": id})
|
|
session.commit()
|
|
|
|
|
|
@router.get("/{id}/items")
|
|
def list_items(
|
|
id: int,
|
|
session: Session = Depends(get_session),
|
|
_: str = Depends(get_current_user),
|
|
) -> list[dict]:
|
|
rows = session.execute(text("""
|
|
SELECT * FROM lk.list_items WHERE list_id = :list_id ORDER BY position, created_at
|
|
"""), {"list_id": id}).fetchall()
|
|
return [_row(r) for r in rows]
|
|
|
|
|
|
@router.post("/{id}/items", status_code=201)
|
|
def create_item(
|
|
id: int,
|
|
body: dict[str, Any],
|
|
session: Session = Depends(get_session),
|
|
_: str = Depends(get_current_user),
|
|
) -> dict:
|
|
row = session.execute(text("""
|
|
INSERT INTO lk.list_items (list_id, text) VALUES (:list_id, :text) RETURNING *
|
|
"""), {"list_id": id, "text": body.get("text")}).fetchone()
|
|
session.commit()
|
|
return _row(row)
|
|
|
|
|
|
@router.patch("/items/{id}")
|
|
def update_item_checked(
|
|
id: int,
|
|
body: dict[str, Any],
|
|
session: Session = Depends(get_session),
|
|
_: str = Depends(get_current_user),
|
|
) -> dict:
|
|
result = session.execute(text("""
|
|
UPDATE lk.list_items SET checked = :checked WHERE id = :id RETURNING *
|
|
"""), {"id": id, "checked": bool(body.get("checked", False))}).fetchone()
|
|
session.commit()
|
|
if result is None:
|
|
raise HTTPException(status_code=404, detail="Not found")
|
|
return _row(result)
|
|
|
|
|
|
@router.delete("/items/{id}", status_code=204)
|
|
def delete_item(
|
|
id: int,
|
|
session: Session = Depends(get_session),
|
|
_: str = Depends(get_current_user),
|
|
):
|
|
session.execute(text("DELETE FROM lk.list_items WHERE id = :id"), {"id": id})
|
|
session.commit()
|