117 lines
3.4 KiB
Python
117 lines
3.4 KiB
Python
import sqlite3
|
|
from typing import List
|
|
|
|
from fastapi import APIRouter, HTTPException, status
|
|
|
|
from app.database import Database
|
|
from app.schemas import Temple, TempleCreate, TempleUpdate
|
|
|
|
|
|
router = APIRouter(prefix="/temples", tags=["temples"])
|
|
|
|
|
|
def row_to_temple(row: sqlite3.Row) -> Temple:
|
|
return Temple(
|
|
id=row["id"],
|
|
name=row["name"],
|
|
location=row["location"],
|
|
contact_phone=row["contact_phone"],
|
|
description=row["description"],
|
|
is_active=bool(row["is_active"]),
|
|
)
|
|
|
|
|
|
def raise_duplicate_temple_error() -> None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_409_CONFLICT,
|
|
detail="Temple name already exists",
|
|
)
|
|
|
|
|
|
@router.get("", response_model=List[Temple])
|
|
def list_temples(db: Database) -> List[Temple]:
|
|
rows = db.execute(
|
|
"""
|
|
SELECT id, name, location, contact_phone, description, is_active
|
|
FROM temples
|
|
ORDER BY id
|
|
"""
|
|
).fetchall()
|
|
return [row_to_temple(row) for row in rows]
|
|
|
|
|
|
@router.post("", response_model=Temple, status_code=status.HTTP_201_CREATED)
|
|
def create_temple(payload: TempleCreate, db: Database) -> Temple:
|
|
try:
|
|
cursor = db.execute(
|
|
"""
|
|
INSERT INTO temples (name, location, contact_phone, description, is_active)
|
|
VALUES (?, ?, ?, ?, ?)
|
|
""",
|
|
(
|
|
payload.name,
|
|
payload.location,
|
|
payload.contact_phone,
|
|
payload.description,
|
|
int(payload.is_active),
|
|
),
|
|
)
|
|
db.commit()
|
|
except sqlite3.IntegrityError:
|
|
raise_duplicate_temple_error()
|
|
|
|
return get_temple(cursor.lastrowid, db)
|
|
|
|
|
|
@router.get("/{temple_id}", response_model=Temple)
|
|
def get_temple(temple_id: int, db: Database) -> Temple:
|
|
row = db.execute(
|
|
"""
|
|
SELECT id, name, location, contact_phone, description, is_active
|
|
FROM temples
|
|
WHERE id = ?
|
|
""",
|
|
(temple_id,),
|
|
).fetchone()
|
|
if row is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Temple not found")
|
|
return row_to_temple(row)
|
|
|
|
|
|
@router.patch("/{temple_id}", response_model=Temple)
|
|
def update_temple(temple_id: int, payload: TempleUpdate, db: Database) -> Temple:
|
|
temple = get_temple(temple_id, db)
|
|
updates = payload.model_dump(exclude_unset=True)
|
|
|
|
if not updates:
|
|
return temple
|
|
|
|
name = updates.get("name", temple.name)
|
|
location = updates.get("location", temple.location)
|
|
contact_phone = updates.get("contact_phone", temple.contact_phone)
|
|
description = updates.get("description", temple.description)
|
|
is_active = updates.get("is_active", temple.is_active)
|
|
|
|
try:
|
|
db.execute(
|
|
"""
|
|
UPDATE temples
|
|
SET name = ?, location = ?, contact_phone = ?, description = ?, is_active = ?
|
|
WHERE id = ?
|
|
""",
|
|
(name, location, contact_phone, description, int(is_active), temple_id),
|
|
)
|
|
db.commit()
|
|
except sqlite3.IntegrityError:
|
|
raise_duplicate_temple_error()
|
|
|
|
return get_temple(temple_id, db)
|
|
|
|
|
|
@router.delete("/{temple_id}", status_code=status.HTTP_204_NO_CONTENT)
|
|
def delete_temple(temple_id: int, db: Database) -> None:
|
|
cursor = db.execute("DELETE FROM temples WHERE id = ?", (temple_id,))
|
|
db.commit()
|
|
if cursor.rowcount == 0:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Temple not found")
|