acazares
b68c4316ff
- Rearranged imports in multiple files for consistency and clarity. - Updated logging middleware to exclude specific paths from logging. - Enhanced security module by cleaning up token handling and improving tenant validation. - Added tenant and company scoped mixins for better database model management. - Implemented generic CRUD routes for tenant-scoped resources. - Improved error handling and response management in API routes. - Cleaned up login and logout processes to ensure proper session management. - Introduced mechanisms to clear local storage and cookies on tenant change. - Enhanced company store to detect tenant changes and clear data accordingly. - Added new DTO mixins for currency and value affect flags.
118 lines
3.9 KiB
Python
118 lines
3.9 KiB
Python
"""
|
|
Routes for PedimentoIncrementables CRUD operations
|
|
"""
|
|
|
|
from typing import Any, Dict, List
|
|
|
|
from core.database import get_core_db
|
|
from core.security import get_current_user, validate_access_to_resource
|
|
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
from sqlalchemy.orm import Session
|
|
|
|
from ..dtos.pedimento_incrementables import (
|
|
PedimentoIncrementablesCreate,
|
|
PedimentoIncrementablesResponse,
|
|
PedimentoIncrementablesUpdate,
|
|
)
|
|
from ..services.pedimento_incrementables import PedimentoIncrementablesService
|
|
|
|
router = APIRouter(prefix="/{pedimento_id}/incrementables")
|
|
|
|
|
|
@router.get("/", response_model=List[PedimentoIncrementablesResponse])
|
|
async def list_incrementables(
|
|
pedimento_id: int,
|
|
company_id: int = Query(..., description="Company ID"),
|
|
db: Session = Depends(get_core_db),
|
|
current_user: Dict[str, Any] = Depends(get_current_user),
|
|
):
|
|
"""Get all incrementables for a pedimento"""
|
|
tenant_id = validate_access_to_resource(db, company_id, current_user)
|
|
|
|
incrementables = PedimentoIncrementablesService.get_by_pedimento_id(
|
|
db, pedimento_id, tenant_id, company_id
|
|
)
|
|
return incrementables
|
|
|
|
|
|
@router.get("/{incrementable_id}", response_model=PedimentoIncrementablesResponse)
|
|
async def get_incrementable(
|
|
pedimento_id: int,
|
|
incrementable_id: int,
|
|
company_id: int = Query(..., description="Company ID"),
|
|
db: Session = Depends(get_core_db),
|
|
current_user: Dict[str, Any] = Depends(get_current_user),
|
|
):
|
|
"""Get a specific incrementable by ID"""
|
|
tenant_id = validate_access_to_resource(db, company_id, current_user)
|
|
|
|
incrementable = PedimentoIncrementablesService.get_by_id(
|
|
db, incrementable_id, pedimento_id, tenant_id, company_id
|
|
)
|
|
if not incrementable:
|
|
raise HTTPException(status_code=404, detail="Incrementable not found")
|
|
|
|
return incrementable
|
|
|
|
|
|
@router.post("/", response_model=PedimentoIncrementablesResponse, status_code=201)
|
|
async def create_incrementable(
|
|
pedimento_id: int,
|
|
data: PedimentoIncrementablesCreate,
|
|
company_id: int = Query(..., description="Company ID"),
|
|
db: Session = Depends(get_core_db),
|
|
current_user: Dict[str, Any] = Depends(get_current_user),
|
|
):
|
|
"""Create a new incrementable"""
|
|
tenant_id = validate_access_to_resource(db, company_id, current_user)
|
|
|
|
# Ensure pedimento_id matches
|
|
if data.pedimento_id != pedimento_id:
|
|
raise HTTPException(status_code=400, detail="Pedimento ID mismatch")
|
|
|
|
incrementable = PedimentoIncrementablesService.create(
|
|
db, data, tenant_id, company_id
|
|
)
|
|
return incrementable
|
|
|
|
|
|
@router.put("/{incrementable_id}", response_model=PedimentoIncrementablesResponse)
|
|
async def update_incrementable(
|
|
pedimento_id: int,
|
|
incrementable_id: int,
|
|
data: PedimentoIncrementablesUpdate,
|
|
company_id: int = Query(..., description="Company ID"),
|
|
db: Session = Depends(get_core_db),
|
|
current_user: Dict[str, Any] = Depends(get_current_user),
|
|
):
|
|
"""Update an incrementable"""
|
|
tenant_id = validate_access_to_resource(db, company_id, current_user)
|
|
|
|
incrementable = PedimentoIncrementablesService.update(
|
|
db, incrementable_id, pedimento_id, tenant_id, company_id, data
|
|
)
|
|
if not incrementable:
|
|
raise HTTPException(status_code=404, detail="Incrementable not found")
|
|
|
|
return incrementable
|
|
|
|
|
|
@router.delete("/{incrementable_id}", status_code=204)
|
|
async def delete_incrementable(
|
|
pedimento_id: int,
|
|
incrementable_id: int,
|
|
company_id: int = Query(..., description="Company ID"),
|
|
db: Session = Depends(get_core_db),
|
|
current_user: Dict[str, Any] = Depends(get_current_user),
|
|
):
|
|
"""Delete an incrementable"""
|
|
tenant_id = validate_access_to_resource(db, company_id, current_user)
|
|
|
|
success = PedimentoIncrementablesService.delete(
|
|
db, incrementable_id, pedimento_id, tenant_id, company_id
|
|
)
|
|
if not success:
|
|
raise HTTPException(status_code=404, detail="Incrementable not found")
|
|
|
|
return None
|