149 lines
4.8 KiB
Python
149 lines
4.8 KiB
Python
"""
|
|
POS Transactions API Endpoints
|
|
ATOMIC layer - Basic CRUD operations for POS transactions
|
|
"""
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Path, Query
|
|
from typing import Optional
|
|
from uuid import UUID
|
|
from datetime import datetime
|
|
from decimal import Decimal
|
|
import structlog
|
|
|
|
from app.core.database import get_db
|
|
from shared.auth.decorators import get_current_user_dep
|
|
from shared.auth.access_control import require_user_role
|
|
from shared.routing import RouteBuilder
|
|
from app.services.pos_transaction_service import POSTransactionService
|
|
from app.schemas.pos_transaction import (
|
|
POSTransactionResponse,
|
|
POSTransactionListResponse,
|
|
POSTransactionDashboardSummary
|
|
)
|
|
|
|
router = APIRouter()
|
|
logger = structlog.get_logger()
|
|
route_builder = RouteBuilder('pos')
|
|
|
|
|
|
@router.get(
|
|
route_builder.build_base_route("transactions"),
|
|
response_model=POSTransactionListResponse
|
|
)
|
|
@require_user_role(['viewer', 'member', 'admin', 'owner'])
|
|
async def list_pos_transactions(
|
|
tenant_id: UUID = Path(...),
|
|
pos_system: Optional[str] = Query(None),
|
|
start_date: Optional[datetime] = Query(None),
|
|
end_date: Optional[datetime] = Query(None),
|
|
status: Optional[str] = Query(None),
|
|
is_synced: Optional[bool] = Query(None),
|
|
limit: int = Query(50, ge=1, le=200),
|
|
offset: int = Query(0, ge=0),
|
|
current_user: dict = Depends(get_current_user_dep),
|
|
db=Depends(get_db)
|
|
):
|
|
"""List POS transactions for a tenant"""
|
|
try:
|
|
service = POSTransactionService()
|
|
|
|
transactions = await service.get_transactions_by_tenant(
|
|
tenant_id=tenant_id,
|
|
pos_system=pos_system,
|
|
start_date=start_date,
|
|
end_date=end_date,
|
|
status=status,
|
|
is_synced=is_synced,
|
|
skip=offset,
|
|
limit=limit
|
|
)
|
|
|
|
total = await service.count_transactions_by_tenant(
|
|
tenant_id=tenant_id,
|
|
pos_system=pos_system,
|
|
start_date=start_date,
|
|
end_date=end_date,
|
|
status=status,
|
|
is_synced=is_synced
|
|
)
|
|
|
|
# Get sync metrics for summary
|
|
sync_metrics = await service.get_sync_metrics(tenant_id)
|
|
|
|
# Calculate summary
|
|
total_amount = sum(float(t.total_amount) for t in transactions if t.status == "completed")
|
|
|
|
has_more = (offset + limit) < total
|
|
|
|
return POSTransactionListResponse(
|
|
transactions=transactions,
|
|
total=total,
|
|
has_more=has_more,
|
|
summary={
|
|
"total_amount": total_amount,
|
|
"transaction_count": len(transactions),
|
|
"sync_status": sync_metrics["sync_status"]
|
|
}
|
|
)
|
|
except Exception as e:
|
|
logger.error("Failed to list POS transactions", error=str(e), tenant_id=tenant_id)
|
|
raise HTTPException(status_code=500, detail=f"Failed to list transactions: {str(e)}")
|
|
|
|
|
|
@router.get(
|
|
route_builder.build_resource_detail_route("transactions", "transaction_id"),
|
|
response_model=POSTransactionResponse
|
|
)
|
|
@require_user_role(['viewer', 'member', 'admin', 'owner'])
|
|
async def get_pos_transaction(
|
|
tenant_id: UUID = Path(...),
|
|
transaction_id: UUID = Path(...),
|
|
current_user: dict = Depends(get_current_user_dep),
|
|
db=Depends(get_db)
|
|
):
|
|
"""Get a specific POS transaction"""
|
|
try:
|
|
service = POSTransactionService()
|
|
|
|
transaction = await service.get_transaction_with_items(
|
|
transaction_id=transaction_id,
|
|
tenant_id=tenant_id
|
|
)
|
|
|
|
if not transaction:
|
|
raise HTTPException(status_code=404, detail="Transaction not found")
|
|
|
|
return transaction
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.error("Failed to get POS transaction", error=str(e),
|
|
tenant_id=tenant_id, transaction_id=transaction_id)
|
|
raise HTTPException(status_code=500, detail=f"Failed to get transaction: {str(e)}")
|
|
|
|
|
|
@router.get(
|
|
route_builder.build_operations_route("transactions-dashboard"),
|
|
response_model=POSTransactionDashboardSummary
|
|
)
|
|
@require_user_role(['viewer', 'member', 'admin', 'owner'])
|
|
async def get_transactions_dashboard(
|
|
tenant_id: UUID = Path(...),
|
|
current_user: dict = Depends(get_current_user_dep),
|
|
db=Depends(get_db)
|
|
):
|
|
"""Get dashboard summary for POS transactions"""
|
|
try:
|
|
service = POSTransactionService()
|
|
|
|
summary = await service.get_dashboard_summary(tenant_id)
|
|
|
|
logger.info("Transactions dashboard retrieved",
|
|
tenant_id=str(tenant_id),
|
|
total_today=summary.total_transactions_today)
|
|
|
|
return summary
|
|
except Exception as e:
|
|
logger.error("Failed to get transactions dashboard", error=str(e), tenant_id=tenant_id)
|
|
raise HTTPException(status_code=500, detail=f"Failed to get dashboard: {str(e)}")
|