121 lines
3.8 KiB
Python
121 lines
3.8 KiB
Python
import sqlite3
|
|
from typing import List
|
|
|
|
from fastapi import APIRouter, HTTPException, status
|
|
|
|
from app.database import Database
|
|
from app.schemas import Order, OrderCreate, OrderUpdate
|
|
from app.temple_scope import TempleId, ensure_temple_exists
|
|
|
|
|
|
router = APIRouter(prefix="/orders", tags=["orders"])
|
|
|
|
|
|
def row_to_order(row: sqlite3.Row) -> Order:
|
|
return Order(
|
|
id=row["id"],
|
|
temple_id=row["temple_id"],
|
|
customer_name=row["customer_name"],
|
|
customer_phone=row["customer_phone"],
|
|
order_type=row["order_type"],
|
|
item_name=row["item_name"],
|
|
amount=float(row["amount"]),
|
|
status=row["status"],
|
|
created_at=row["created_at"],
|
|
)
|
|
|
|
|
|
@router.get("", response_model=List[Order])
|
|
def list_orders(db: Database, temple_id: TempleId) -> List[Order]:
|
|
ensure_temple_exists(db, temple_id)
|
|
rows = db.execute(
|
|
"""
|
|
SELECT id, temple_id, customer_name, customer_phone, order_type, item_name, amount, status, created_at
|
|
FROM orders
|
|
WHERE temple_id = ?
|
|
ORDER BY id DESC
|
|
""",
|
|
(temple_id,),
|
|
).fetchall()
|
|
return [row_to_order(row) for row in rows]
|
|
|
|
|
|
@router.post("", response_model=Order, status_code=status.HTTP_201_CREATED)
|
|
def create_order(payload: OrderCreate, db: Database, temple_id: TempleId) -> Order:
|
|
ensure_temple_exists(db, temple_id)
|
|
cursor = db.execute(
|
|
"""
|
|
INSERT INTO orders
|
|
(temple_id, customer_name, customer_phone, order_type, item_name, amount, status)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
(
|
|
temple_id,
|
|
payload.customer_name,
|
|
payload.customer_phone,
|
|
payload.order_type,
|
|
payload.item_name,
|
|
payload.amount,
|
|
payload.status,
|
|
),
|
|
)
|
|
db.commit()
|
|
return get_order(cursor.lastrowid, db, temple_id)
|
|
|
|
|
|
@router.get("/{order_id}", response_model=Order)
|
|
def get_order(order_id: int, db: Database, temple_id: TempleId) -> Order:
|
|
ensure_temple_exists(db, temple_id)
|
|
row = db.execute(
|
|
"""
|
|
SELECT id, temple_id, customer_name, customer_phone, order_type, item_name, amount, status, created_at
|
|
FROM orders
|
|
WHERE id = ? AND temple_id = ?
|
|
""",
|
|
(order_id, temple_id),
|
|
).fetchone()
|
|
if row is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Order not found")
|
|
return row_to_order(row)
|
|
|
|
|
|
@router.patch("/{order_id}", response_model=Order)
|
|
def update_order(order_id: int, payload: OrderUpdate, db: Database, temple_id: TempleId) -> Order:
|
|
order = get_order(order_id, db, temple_id)
|
|
updates = payload.model_dump(exclude_unset=True)
|
|
|
|
if not updates:
|
|
return order
|
|
|
|
db.execute(
|
|
"""
|
|
UPDATE orders
|
|
SET customer_name = ?, customer_phone = ?, order_type = ?, item_name = ?, amount = ?, status = ?
|
|
WHERE id = ? AND temple_id = ?
|
|
""",
|
|
(
|
|
updates.get("customer_name", order.customer_name),
|
|
updates.get("customer_phone", order.customer_phone),
|
|
updates.get("order_type", order.order_type),
|
|
updates.get("item_name", order.item_name),
|
|
updates.get("amount", order.amount),
|
|
updates.get("status", order.status),
|
|
order_id,
|
|
temple_id,
|
|
),
|
|
)
|
|
db.commit()
|
|
return get_order(order_id, db, temple_id)
|
|
|
|
|
|
@router.delete("/{order_id}", status_code=status.HTTP_204_NO_CONTENT)
|
|
def delete_order(order_id: int, db: Database, temple_id: TempleId) -> None:
|
|
ensure_temple_exists(db, temple_id)
|
|
cursor = db.execute(
|
|
"DELETE FROM orders WHERE id = ? AND temple_id = ?",
|
|
(order_id, temple_id),
|
|
)
|
|
db.commit()
|
|
if cursor.rowcount == 0:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Order not found")
|