52 lines
2.0 KiB
Python
52 lines
2.0 KiB
Python
"""Quote events — audit timeline. Manual creation is restricted to event_type=note."""
|
|
from fastapi import APIRouter, Depends, Query
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.db.engine import get_db
|
|
from app.models.procurement import Quote, QuoteEvent
|
|
from app.schemas.procurement import NoteEventIn
|
|
from app.services.expand_service import event_to_dict
|
|
from app.utils.errors import NotFoundError, ValidationError
|
|
from app.utils.pagination import paginate, sort_query
|
|
|
|
router = APIRouter(prefix="/quotes/{quote_id}/events", tags=["quote-events"])
|
|
|
|
|
|
@router.get("")
|
|
async def list_events(
|
|
quote_id,
|
|
page: int = Query(1, ge=1),
|
|
page_size: int = Query(20, ge=1, le=100),
|
|
event_type: str | None = None,
|
|
sort: str | None = None,
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
quote = await db.get(Quote, quote_id)
|
|
if quote is None:
|
|
raise NotFoundError("پیشنهاد (استعلام) یافت نشد")
|
|
stmt = select(QuoteEvent).where(QuoteEvent.quote_id == quote_id)
|
|
if event_type:
|
|
stmt = stmt.where(QuoteEvent.event_type == event_type)
|
|
stmt = sort_query(stmt, QuoteEvent, sort, default="occurred_at") # contract default: ASC
|
|
items, meta = await paginate(stmt, page, page_size, db)
|
|
return {"data": [event_to_dict(e) for e in items], "meta": meta}
|
|
|
|
|
|
@router.post("", status_code=201)
|
|
async def create_event(quote_id, payload: NoteEventIn, db: AsyncSession = Depends(get_db)):
|
|
quote = await db.get(Quote, quote_id)
|
|
if quote is None:
|
|
raise NotFoundError("پیشنهاد (استعلام) یافت نشد")
|
|
if payload.event_type != "note":
|
|
raise ValidationError("فقط رویدادهای note به صورت دستی ثبت میشوند", field="event_type")
|
|
event = QuoteEvent(
|
|
quote_id=quote.id,
|
|
event_type="note",
|
|
payload=payload.payload or {},
|
|
actor=payload.actor,
|
|
)
|
|
db.add(event)
|
|
await db.flush()
|
|
return {"data": event_to_dict(event)}
|