285 lines
8.3 KiB
Python
285 lines
8.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)
|
|
|
|
|
|
# ---- Counts ----
|
|
|
|
@router.get("/counts")
|
|
def list_counts(
|
|
colony_id: int = Query(...),
|
|
session: Session = Depends(get_session),
|
|
_: str = Depends(get_current_user),
|
|
) -> list[dict]:
|
|
rows = session.execute(text("""
|
|
SELECT * FROM varroa_counts WHERE colony_id = :colony_id ORDER BY date DESC
|
|
"""), {"colony_id": colony_id}).fetchall()
|
|
return [_row(r) for r in rows]
|
|
|
|
|
|
@router.post("/counts", status_code=201)
|
|
def create_count(
|
|
body: dict[str, Any],
|
|
session: Session = Depends(get_session),
|
|
_: str = Depends(get_current_user),
|
|
) -> dict:
|
|
row = session.execute(text("""
|
|
INSERT INTO varroa_counts (colony_id, date_from, date, method, count, notes, time)
|
|
VALUES (:colony_id, :date_from, :date, :method, :count, :notes, :time)
|
|
RETURNING *
|
|
"""), {
|
|
"colony_id": body.get("colony_id"),
|
|
"date_from": body.get("date_from"),
|
|
"date": body.get("date"),
|
|
"method": body.get("method"),
|
|
"count": body.get("count"),
|
|
"notes": body.get("notes"),
|
|
"time": body.get("time"),
|
|
}).fetchone()
|
|
session.commit()
|
|
return _row(row)
|
|
|
|
|
|
@router.get("/counts/{id}")
|
|
def get_count(
|
|
id: int,
|
|
session: Session = Depends(get_session),
|
|
_: str = Depends(get_current_user),
|
|
) -> dict:
|
|
row = session.execute(
|
|
text("SELECT * FROM varroa_counts WHERE id = :id"), {"id": id}
|
|
).fetchone()
|
|
if row is None:
|
|
raise HTTPException(status_code=404, detail="Not found")
|
|
return _row(row)
|
|
|
|
|
|
@router.put("/counts/{id}")
|
|
def update_count(
|
|
id: int,
|
|
body: dict[str, Any],
|
|
session: Session = Depends(get_session),
|
|
_: str = Depends(get_current_user),
|
|
) -> dict:
|
|
result = session.execute(text("""
|
|
UPDATE varroa_counts
|
|
SET date_from = :date_from, date = :date, method = :method, count = :count, notes = :notes
|
|
WHERE id = :id
|
|
RETURNING *
|
|
"""), {
|
|
"id": id,
|
|
"date_from": body.get("date_from"),
|
|
"date": body.get("date"),
|
|
"method": body.get("method"),
|
|
"count": body.get("count"),
|
|
"notes": body.get("notes"),
|
|
}).fetchone()
|
|
session.commit()
|
|
if result is None:
|
|
raise HTTPException(status_code=404, detail="Not found")
|
|
return _row(result)
|
|
|
|
|
|
@router.delete("/counts/{id}", status_code=204)
|
|
def delete_count(
|
|
id: int,
|
|
session: Session = Depends(get_session),
|
|
_: str = Depends(get_current_user),
|
|
):
|
|
session.execute(text("DELETE FROM varroa_counts WHERE id = :id"), {"id": id})
|
|
session.commit()
|
|
|
|
|
|
# ---- Treatments ----
|
|
|
|
@router.get("/treatments")
|
|
def list_treatments(
|
|
colony_id: int = Query(...),
|
|
session: Session = Depends(get_session),
|
|
_: str = Depends(get_current_user),
|
|
) -> list[dict]:
|
|
rows = session.execute(text("""
|
|
SELECT * FROM varroa_treatments WHERE colony_id = :colony_id ORDER BY date DESC
|
|
"""), {"colony_id": colony_id}).fetchall()
|
|
return [_row(r) for r in rows]
|
|
|
|
|
|
@router.post("/treatments", status_code=201)
|
|
def create_treatment(
|
|
body: dict[str, Any],
|
|
session: Session = Depends(get_session),
|
|
_: str = Depends(get_current_user),
|
|
) -> dict:
|
|
row = session.execute(text("""
|
|
INSERT INTO varroa_treatments (colony_id, date, product, method, dosage, notes, time)
|
|
VALUES (:colony_id, :date, :product, :method, :dosage, :notes, :time)
|
|
RETURNING *
|
|
"""), {
|
|
"colony_id": body.get("colony_id"),
|
|
"date": body.get("date"),
|
|
"product": body.get("product"),
|
|
"method": body.get("method"),
|
|
"dosage": body.get("dosage"),
|
|
"notes": body.get("notes"),
|
|
"time": body.get("time"),
|
|
}).fetchone()
|
|
session.commit()
|
|
return _row(row)
|
|
|
|
|
|
@router.get("/treatments/{id}")
|
|
def get_treatment(
|
|
id: int,
|
|
session: Session = Depends(get_session),
|
|
_: str = Depends(get_current_user),
|
|
) -> dict:
|
|
row = session.execute(
|
|
text("SELECT * FROM varroa_treatments WHERE id = :id"), {"id": id}
|
|
).fetchone()
|
|
if row is None:
|
|
raise HTTPException(status_code=404, detail="Not found")
|
|
return _row(row)
|
|
|
|
|
|
@router.put("/treatments/{id}")
|
|
def update_treatment(
|
|
id: int,
|
|
body: dict[str, Any],
|
|
session: Session = Depends(get_session),
|
|
_: str = Depends(get_current_user),
|
|
) -> dict:
|
|
result = session.execute(text("""
|
|
UPDATE varroa_treatments
|
|
SET date = :date, product = :product, method = :method, dosage = :dosage, notes = :notes
|
|
WHERE id = :id
|
|
RETURNING *
|
|
"""), {
|
|
"id": id,
|
|
"date": body.get("date"),
|
|
"product": body.get("product"),
|
|
"method": body.get("method"),
|
|
"dosage": body.get("dosage"),
|
|
"notes": body.get("notes"),
|
|
}).fetchone()
|
|
session.commit()
|
|
if result is None:
|
|
raise HTTPException(status_code=404, detail="Not found")
|
|
return _row(result)
|
|
|
|
|
|
@router.delete("/treatments/{id}", status_code=204)
|
|
def delete_treatment(
|
|
id: int,
|
|
session: Session = Depends(get_session),
|
|
_: str = Depends(get_current_user),
|
|
):
|
|
session.execute(text("DELETE FROM varroa_treatments WHERE id = :id"), {"id": id})
|
|
session.commit()
|
|
|
|
|
|
# ---- Controls ----
|
|
|
|
@router.get("/controls")
|
|
def list_controls(
|
|
colony_id: int | None = Query(default=None),
|
|
treatment_id: int | None = Query(default=None),
|
|
session: Session = Depends(get_session),
|
|
_: str = Depends(get_current_user),
|
|
) -> list[dict]:
|
|
if treatment_id is not None:
|
|
rows = session.execute(text("""
|
|
SELECT vc.*, vt.product, vt.date AS treatment_date
|
|
FROM varroa_controls vc
|
|
JOIN varroa_treatments vt ON vt.id = vc.treatment_id
|
|
WHERE vc.treatment_id = :treatment_id
|
|
ORDER BY vc.date DESC
|
|
"""), {"treatment_id": treatment_id}).fetchall()
|
|
elif colony_id is not None:
|
|
rows = session.execute(text("""
|
|
SELECT vc.*, vt.product, vt.date AS treatment_date
|
|
FROM varroa_controls vc
|
|
JOIN varroa_treatments vt ON vt.id = vc.treatment_id
|
|
WHERE vc.colony_id = :colony_id
|
|
ORDER BY vc.date DESC
|
|
"""), {"colony_id": colony_id}).fetchall()
|
|
else:
|
|
return []
|
|
return [_row(r) for r in rows]
|
|
|
|
|
|
@router.post("/controls", status_code=201)
|
|
def create_control(
|
|
body: dict[str, Any],
|
|
session: Session = Depends(get_session),
|
|
_: str = Depends(get_current_user),
|
|
) -> dict:
|
|
row = session.execute(text("""
|
|
INSERT INTO varroa_controls (treatment_id, colony_id, date, result, notes)
|
|
VALUES (:treatment_id, :colony_id, :date, :result, :notes)
|
|
RETURNING *
|
|
"""), {
|
|
"treatment_id": body.get("treatment_id"),
|
|
"colony_id": body.get("colony_id"),
|
|
"date": body.get("date"),
|
|
"result": body.get("result"),
|
|
"notes": body.get("notes"),
|
|
}).fetchone()
|
|
session.commit()
|
|
return _row(row)
|
|
|
|
|
|
@router.get("/controls/{id}")
|
|
def get_control(
|
|
id: int,
|
|
session: Session = Depends(get_session),
|
|
_: str = Depends(get_current_user),
|
|
) -> dict:
|
|
row = session.execute(
|
|
text("SELECT * FROM varroa_controls WHERE id = :id"), {"id": id}
|
|
).fetchone()
|
|
if row is None:
|
|
raise HTTPException(status_code=404, detail="Not found")
|
|
return _row(row)
|
|
|
|
|
|
@router.put("/controls/{id}")
|
|
def update_control(
|
|
id: int,
|
|
body: dict[str, Any],
|
|
session: Session = Depends(get_session),
|
|
_: str = Depends(get_current_user),
|
|
) -> dict:
|
|
result = session.execute(text("""
|
|
UPDATE varroa_controls SET date = :date, result = :result, notes = :notes
|
|
WHERE id = :id RETURNING *
|
|
"""), {
|
|
"id": id,
|
|
"date": body.get("date"),
|
|
"result": body.get("result"),
|
|
"notes": body.get("notes"),
|
|
}).fetchone()
|
|
session.commit()
|
|
if result is None:
|
|
raise HTTPException(status_code=404, detail="Not found")
|
|
return _row(result)
|
|
|
|
|
|
@router.delete("/controls/{id}", status_code=204)
|
|
def delete_control(
|
|
id: int,
|
|
session: Session = Depends(get_session),
|
|
_: str = Depends(get_current_user),
|
|
):
|
|
session.execute(text("DELETE FROM varroa_controls WHERE id = :id"), {"id": id})
|
|
session.commit()
|