100 lines
2.6 KiB
Python
100 lines
2.6 KiB
Python
from fastapi import APIRouter, Depends, HTTPException
|
|
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_apiaries(
|
|
session: Session = Depends(get_session),
|
|
_: str = Depends(get_current_user),
|
|
) -> list[dict]:
|
|
rows = session.execute(text("""
|
|
SELECT a.*, COUNT(c.id) AS colony_count
|
|
FROM apiaries a
|
|
LEFT JOIN colonies c ON c.apiary_id = a.id AND c.closed_at IS NULL
|
|
GROUP BY a.id
|
|
ORDER BY a.name
|
|
""")).fetchall()
|
|
return [_row(r) for r in rows]
|
|
|
|
|
|
@router.get("/{id}")
|
|
def get_apiary(
|
|
id: int,
|
|
session: Session = Depends(get_session),
|
|
_: str = Depends(get_current_user),
|
|
) -> dict:
|
|
row = session.execute(
|
|
text("SELECT * FROM apiaries WHERE id = :id"), {"id": id}
|
|
).fetchone()
|
|
if row is None:
|
|
raise HTTPException(status_code=404, detail="Not found")
|
|
return _row(row)
|
|
|
|
|
|
@router.post("", status_code=201)
|
|
def create_apiary(
|
|
body: dict[str, Any],
|
|
session: Session = Depends(get_session),
|
|
_: str = Depends(get_current_user),
|
|
) -> dict:
|
|
row = session.execute(text("""
|
|
INSERT INTO apiaries (name, location_text, lat, lng, notes)
|
|
VALUES (:name, :location_text, :lat, :lng, :notes)
|
|
RETURNING *
|
|
"""), {
|
|
"name": body.get("name"),
|
|
"location_text": body.get("location_text"),
|
|
"lat": body.get("lat"),
|
|
"lng": body.get("lng"),
|
|
"notes": body.get("notes"),
|
|
}).fetchone()
|
|
session.commit()
|
|
return _row(row)
|
|
|
|
|
|
@router.put("/{id}")
|
|
def update_apiary(
|
|
id: int,
|
|
body: dict[str, Any],
|
|
session: Session = Depends(get_session),
|
|
_: str = Depends(get_current_user),
|
|
) -> dict:
|
|
result = session.execute(text("""
|
|
UPDATE apiaries
|
|
SET name = :name, location_text = :location_text, lat = :lat, lng = :lng, notes = :notes
|
|
WHERE id = :id
|
|
RETURNING *
|
|
"""), {
|
|
"id": id,
|
|
"name": body.get("name"),
|
|
"location_text": body.get("location_text"),
|
|
"lat": body.get("lat"),
|
|
"lng": body.get("lng"),
|
|
"notes": body.get("notes"),
|
|
}).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_apiary(
|
|
id: int,
|
|
session: Session = Depends(get_session),
|
|
_: str = Depends(get_current_user),
|
|
):
|
|
session.execute(text("DELETE FROM apiaries WHERE id = :id"), {"id": id})
|
|
session.commit()
|