Spaces:
Build error
Build error
File size: 1,500 Bytes
5e52bd7 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 | from typing import List, Optional
from fastapi import APIRouter, Query, Depends
from sqlalchemy.orm import Session
from app.schemas.oc import PurchaseOrder
from app.database import get_db
from app.models.oc import OCModel
from app.services.mercado_publico_oc import get_ocs_by_date, get_oc_by_code
from app.services.sync import sync_purchase_orders_to_db
router = APIRouter()
@router.get("/purchase-orders", response_model=List[PurchaseOrder])
async def list_purchase_orders(
date: Optional[str] = None,
status: str = "todos",
db: Session = Depends(get_db)
):
"""
List purchase orders for a specific date (ddmmaaaa).
"""
if not date:
from datetime import datetime
date = datetime.now().strftime("%d%m%Y")
# Try to fetch current OC data from the live API
ocs = await get_ocs_by_date(date, status)
if ocs:
await sync_purchase_orders_to_db(db, date, status)
return ocs
# Fallback to cached DB entries when the API returns no results
db_results = db.query(OCModel).order_by(OCModel.date_creation.desc()).all()
return db_results
@router.post("/purchase-orders/sync")
async def sync_purchase_orders(
date: Optional[str] = None,
status: str = "todos",
db: Session = Depends(get_db)
):
return await sync_purchase_orders_to_db(db, date, status)
@router.get("/purchase-orders/{code}", response_model=Optional[PurchaseOrder])
async def get_purchase_order(code: str):
return await get_oc_by_code(code)
|