142 lines
4.4 KiB
Python
142 lines
4.4 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_queens(
|
|
colony_id: int | None = Query(default=None),
|
|
session: Session = Depends(get_session),
|
|
_: str = Depends(get_current_user),
|
|
) -> list[dict]:
|
|
params: dict[str, Any] = {}
|
|
where = ""
|
|
if colony_id is not None:
|
|
where = "WHERE q.colony_id = :colony_id"
|
|
params["colony_id"] = colony_id
|
|
|
|
rows = session.execute(text(f"""
|
|
SELECT q.*, c.name AS colony_name, o.name AS origin_colony_name
|
|
FROM queens q
|
|
LEFT JOIN colonies c ON c.id = q.colony_id
|
|
LEFT JOIN colonies o ON o.id = q.origin_colony_id
|
|
{where}
|
|
ORDER BY q.introduced_at DESC
|
|
"""), params).fetchall()
|
|
return [_row(r) for r in rows]
|
|
|
|
|
|
@router.post("", status_code=201)
|
|
def create_queen(
|
|
body: dict[str, Any],
|
|
session: Session = Depends(get_session),
|
|
_: str = Depends(get_current_user),
|
|
) -> dict:
|
|
row = session.execute(text("""
|
|
INSERT INTO queens (colony_id, year, breed, origin_colony_id, marked_color, notes, introduced_at)
|
|
VALUES (:colony_id, :year, :breed, :origin_colony_id, :marked_color, :notes, :introduced_at)
|
|
RETURNING *
|
|
"""), {
|
|
"colony_id": body.get("colony_id"),
|
|
"year": body.get("year"),
|
|
"breed": body.get("breed"),
|
|
"origin_colony_id": body.get("origin_colony_id"),
|
|
"marked_color": body.get("marked_color"),
|
|
"notes": body.get("notes"),
|
|
"introduced_at": body.get("introduced_at"),
|
|
}).fetchone()
|
|
queen_id = row._mapping["id"]
|
|
colony_id = body.get("colony_id")
|
|
if colony_id is not None:
|
|
session.execute(text("""
|
|
UPDATE colonies SET current_queen_id = :queen_id WHERE id = :colony_id
|
|
"""), {"queen_id": queen_id, "colony_id": colony_id})
|
|
session.commit()
|
|
return _row(row)
|
|
|
|
|
|
@router.get("/{id}")
|
|
def get_queen(
|
|
id: int,
|
|
session: Session = Depends(get_session),
|
|
_: str = Depends(get_current_user),
|
|
) -> dict:
|
|
row = session.execute(
|
|
text("SELECT * FROM queens WHERE id = :id"), {"id": id}
|
|
).fetchone()
|
|
if row is None:
|
|
raise HTTPException(status_code=404, detail="Not found")
|
|
return _row(row)
|
|
|
|
|
|
@router.put("/{id}")
|
|
def update_queen(
|
|
id: int,
|
|
body: dict[str, Any],
|
|
session: Session = Depends(get_session),
|
|
_: str = Depends(get_current_user),
|
|
) -> dict:
|
|
result = session.execute(text("""
|
|
UPDATE queens SET
|
|
colony_id = :colony_id, year = :year, breed = :breed,
|
|
origin_colony_id = :origin_colony_id, marked_color = :marked_color,
|
|
notes = :notes, introduced_at = :introduced_at, superseded_at = :superseded_at
|
|
WHERE id = :id
|
|
RETURNING *
|
|
"""), {
|
|
"id": id,
|
|
"colony_id": body.get("colony_id"),
|
|
"year": body.get("year"),
|
|
"breed": body.get("breed"),
|
|
"origin_colony_id": body.get("origin_colony_id"),
|
|
"marked_color": body.get("marked_color"),
|
|
"notes": body.get("notes"),
|
|
"introduced_at": body.get("introduced_at"),
|
|
"superseded_at": body.get("superseded_at"),
|
|
}).fetchone()
|
|
session.commit()
|
|
if result is None:
|
|
raise HTTPException(status_code=404, detail="Not found")
|
|
return _row(result)
|
|
|
|
|
|
@router.post("/{id}/supersede")
|
|
def supersede_queen(
|
|
id: int,
|
|
session: Session = Depends(get_session),
|
|
_: str = Depends(get_current_user),
|
|
) -> dict:
|
|
result = session.execute(text("""
|
|
UPDATE queens SET superseded_at = CURRENT_DATE WHERE id = :id RETURNING *
|
|
"""), {"id": id}).fetchone()
|
|
if result is None:
|
|
raise HTTPException(status_code=404, detail="Not found")
|
|
colony_id = result._mapping.get("colony_id")
|
|
if colony_id is not None:
|
|
session.execute(text("""
|
|
UPDATE colonies SET current_queen_id = NULL
|
|
WHERE id = :colony_id AND current_queen_id = :queen_id
|
|
"""), {"colony_id": colony_id, "queen_id": id})
|
|
session.commit()
|
|
return _row(result)
|
|
|
|
|
|
@router.delete("/{id}", status_code=204)
|
|
def delete_queen(
|
|
id: int,
|
|
session: Session = Depends(get_session),
|
|
_: str = Depends(get_current_user),
|
|
):
|
|
session.execute(text("DELETE FROM queens WHERE id = :id"), {"id": id})
|
|
session.commit()
|