acazares
dcd42583d9
- Updated the "Nueva Sección" button in customs_sections, customs_warehouses, incoterms, invoice_types, material_types, payment_methods, pedimento_codes, pedimento_regimens, sectors, states, transport_modes, and transport_types pages to use the Plus icon from Lucide. - Updated the "Actualizar" button in the same pages to use the RefreshCw icon from Lucide. - Added a new edit dialog component for customs brokers with a comprehensive form for editing broker details, including validation and loading states.
394 lines
12 KiB
Python
394 lines
12 KiB
Python
"""
|
|
Endpoints API para gestión de partes/componentes
|
|
"""
|
|
|
|
from typing import List, Optional
|
|
|
|
from core.database import get_core_db
|
|
from core.security import get_current_user
|
|
from fastapi import APIRouter, Depends, HTTPException, Query, status
|
|
from sqlalchemy.orm import Session
|
|
|
|
from .dto import (
|
|
PartBasicDTO,
|
|
PartCreateDTO,
|
|
PartListDTO,
|
|
PartResponseDTO,
|
|
PartSearchDTO,
|
|
PartUpdateDTO,
|
|
)
|
|
from .service import PartService
|
|
|
|
router = APIRouter(prefix="/parts")
|
|
|
|
|
|
@router.post("/", response_model=PartResponseDTO, status_code=status.HTTP_201_CREATED)
|
|
async def create_part(
|
|
part_data: PartCreateDTO,
|
|
db: Session = Depends(get_core_db),
|
|
current_user: dict = Depends(get_current_user),
|
|
):
|
|
"""
|
|
Create a new part in the system
|
|
"""
|
|
# Validate access to the tenant and company
|
|
tenant_id = current_user.get("tenant_id")
|
|
company_id = current_user.get("company_id")
|
|
|
|
if not tenant_id or not company_id:
|
|
raise HTTPException(
|
|
status_code=403, detail="Access denied: Tenant or Company not found"
|
|
)
|
|
|
|
service = PartService(db)
|
|
return service.create_part(part_data)
|
|
|
|
|
|
@router.get("/", response_model=PartListDTO)
|
|
async def list_parts(
|
|
skip: int = Query(0, ge=0, description="Number of records to skip"),
|
|
limit: int = Query(
|
|
100, ge=1, le=1000, description="Maximum number of records to return"
|
|
),
|
|
client_id: Optional[int] = Query(None, description="Filter by client key"),
|
|
part_number: Optional[str] = Query(None, description="Search by part number"),
|
|
description: Optional[str] = Query(None, description="Search in descriptions"),
|
|
fraction: Optional[str] = Query(None, description="Filter by tariff fraction"),
|
|
supplier: Optional[str] = Query(None, description="Filter by supplier"),
|
|
enabled_only: bool = Query(False, description="Show only enabled parts"),
|
|
db: Session = Depends(get_core_db),
|
|
current_user: dict = Depends(get_current_user),
|
|
):
|
|
"""
|
|
List parts with optional filters and pagination
|
|
"""
|
|
# Validate access to the tenant and company
|
|
tenant_id = current_user.get("tenant_id")
|
|
company_id = current_user.get("company_id")
|
|
|
|
if not tenant_id or not company_id:
|
|
raise HTTPException(
|
|
status_code=403, detail="Access denied: Tenant or Company not found"
|
|
)
|
|
|
|
service = PartService(db)
|
|
search_params = PartSearchDTO(
|
|
client_id=client_id,
|
|
part_number=part_number,
|
|
description=description,
|
|
fraction=fraction,
|
|
supplier=supplier,
|
|
enabled_only=enabled_only,
|
|
)
|
|
return service.list_parts(skip, limit, search_params)
|
|
|
|
|
|
@router.get("/client/{client_id}", response_model=List[PartBasicDTO])
|
|
async def get_parts_by_client(
|
|
client_id: int,
|
|
skip: int = Query(0, ge=0),
|
|
limit: int = Query(100, ge=1, le=1000),
|
|
db: Session = Depends(get_core_db),
|
|
current_user: dict = Depends(get_current_user),
|
|
):
|
|
"""
|
|
Get all parts for a specific client
|
|
"""
|
|
# Validate access to the tenant and company
|
|
tenant_id = current_user.get("tenant_id")
|
|
company_id = current_user.get("company_id")
|
|
|
|
if not tenant_id or not company_id:
|
|
raise HTTPException(
|
|
status_code=403, detail="Access denied: Tenant or Company not found"
|
|
)
|
|
|
|
service = PartService(db)
|
|
return service.search_by_client(client_id, skip, limit)
|
|
|
|
|
|
@router.get("/search/fraction/{fraction}", response_model=List[PartBasicDTO])
|
|
async def search_by_fraction(
|
|
fraction: str,
|
|
db: Session = Depends(get_core_db),
|
|
current_user: dict = Depends(get_current_user),
|
|
):
|
|
"""
|
|
Search parts by tariff fraction
|
|
"""
|
|
# Validate access to the tenant and company
|
|
tenant_id = current_user.get("tenant_id")
|
|
company_id = current_user.get("company_id")
|
|
|
|
if not tenant_id or not company_id:
|
|
raise HTTPException(
|
|
status_code=403, detail="Access denied: Tenant or Company not found"
|
|
)
|
|
|
|
service = PartService(db)
|
|
return service.search_by_fraction(fraction)
|
|
|
|
|
|
@router.get("/search/supplier/{supplier}", response_model=List[PartBasicDTO])
|
|
async def search_by_supplier(
|
|
supplier: str,
|
|
db: Session = Depends(get_core_db),
|
|
current_user: dict = Depends(get_current_user),
|
|
):
|
|
"""
|
|
Search parts by supplier
|
|
"""
|
|
# Validate access to the tenant and company
|
|
tenant_id = current_user.get("tenant_id")
|
|
company_id = current_user.get("company_id")
|
|
|
|
if not tenant_id or not company_id:
|
|
raise HTTPException(
|
|
status_code=403, detail="Access denied: Tenant or Company not found"
|
|
)
|
|
|
|
service = PartService(db)
|
|
return service.search_by_supplier(supplier)
|
|
|
|
|
|
@router.get("/search/country/{country_code}", response_model=List[PartBasicDTO])
|
|
async def get_parts_by_country(
|
|
country_code: str,
|
|
db: Session = Depends(get_core_db),
|
|
current_user: dict = Depends(get_current_user),
|
|
):
|
|
"""
|
|
Get parts by country of origin
|
|
"""
|
|
# Validate access to the tenant and company
|
|
tenant_id = current_user.get("tenant_id")
|
|
company_id = current_user.get("company_id")
|
|
|
|
if not tenant_id or not company_id:
|
|
raise HTTPException(
|
|
status_code=403, detail="Access denied: Tenant or Company not found"
|
|
)
|
|
|
|
service = PartService(db)
|
|
return service.get_parts_by_country(country_code)
|
|
|
|
|
|
@router.get("/statistics", response_model=dict)
|
|
async def get_parts_statistics(
|
|
db: Session = Depends(get_core_db), current_user: dict = Depends(get_current_user)
|
|
):
|
|
"""
|
|
Get basic parts statistics
|
|
"""
|
|
# Validate access to the tenant and company
|
|
tenant_id = current_user.get("tenant_id")
|
|
company_id = current_user.get("company_id")
|
|
|
|
if not tenant_id or not company_id:
|
|
raise HTTPException(
|
|
status_code=403, detail="Access denied: Tenant or Company not found"
|
|
)
|
|
|
|
service = PartService(db)
|
|
return service.get_parts_statistics()
|
|
|
|
|
|
@router.get("/{client_id}/{part_number}", response_model=PartResponseDTO)
|
|
async def get_part(
|
|
client_id: int,
|
|
part_number: str,
|
|
db: Session = Depends(get_core_db),
|
|
current_user: dict = Depends(get_current_user),
|
|
):
|
|
"""
|
|
Get part by composite key (client_id + part_number)
|
|
"""
|
|
# Validate access to the tenant and company
|
|
tenant_id = current_user.get("tenant_id")
|
|
company_id = current_user.get("company_id")
|
|
|
|
if not tenant_id or not company_id:
|
|
raise HTTPException(
|
|
status_code=403, detail="Access denied: Tenant or Company not found"
|
|
)
|
|
|
|
service = PartService(db)
|
|
part = service.get_part(client_id, part_number)
|
|
if not part:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail=f"Part with client_id '{client_id}' and part_number '{part_number}' not found",
|
|
)
|
|
return part
|
|
|
|
|
|
@router.put("/{client_id}/{part_number}", response_model=PartResponseDTO)
|
|
async def update_part(
|
|
client_id: int,
|
|
part_number: str,
|
|
part_data: PartUpdateDTO,
|
|
db: Session = Depends(get_core_db),
|
|
current_user: dict = Depends(get_current_user),
|
|
):
|
|
"""
|
|
Update part information
|
|
"""
|
|
# Validate access to the tenant and company
|
|
tenant_id = current_user.get("tenant_id")
|
|
company_id = current_user.get("company_id")
|
|
|
|
if not tenant_id or not company_id:
|
|
raise HTTPException(
|
|
status_code=403, detail="Access denied: Tenant or Company not found"
|
|
)
|
|
|
|
service = PartService(db)
|
|
part = service.update_part(client_id, part_number, part_data)
|
|
if not part:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail=f"Part with client_id '{client_id}' and part_number '{part_number}' not found",
|
|
)
|
|
return part
|
|
|
|
|
|
@router.delete("/{client_id}/{part_number}", status_code=status.HTTP_204_NO_CONTENT)
|
|
async def delete_part(
|
|
client_id: int,
|
|
part_number: str,
|
|
db: Session = Depends(get_core_db),
|
|
current_user: dict = Depends(get_current_user),
|
|
):
|
|
"""
|
|
Delete part from the system
|
|
|
|
Note: This will completely remove the part from the system.
|
|
"""
|
|
# Validate access to the tenant and company
|
|
tenant_id = current_user.get("tenant_id")
|
|
company_id = current_user.get("company_id")
|
|
|
|
if not tenant_id or not company_id:
|
|
raise HTTPException(
|
|
status_code=403, detail="Access denied: Tenant or Company not found"
|
|
)
|
|
|
|
service = PartService(db)
|
|
if not service.delete_part(client_id, part_number):
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail=f"Part with client_id '{client_id}' and part_number '{part_number}' not found",
|
|
)
|
|
|
|
|
|
@router.patch(
|
|
"/{client_id}/{part_number}/toggle-status", response_model=PartResponseDTO
|
|
)
|
|
async def toggle_part_status(
|
|
client_id: int,
|
|
part_number: str,
|
|
db: Session = Depends(get_core_db),
|
|
current_user: dict = Depends(get_current_user),
|
|
):
|
|
"""
|
|
Toggle part enabled/disabled status
|
|
"""
|
|
# Validate access to the tenant and company
|
|
tenant_id = current_user.get("tenant_id")
|
|
company_id = current_user.get("company_id")
|
|
|
|
if not tenant_id or not company_id:
|
|
raise HTTPException(
|
|
status_code=403, detail="Access denied: Tenant or Company not found"
|
|
)
|
|
|
|
service = PartService(db)
|
|
part = service.toggle_status(client_id, part_number)
|
|
if not part:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail=f"Part with client_id '{client_id}' and part_number '{part_number}' not found",
|
|
)
|
|
return part
|
|
|
|
|
|
# Endpoints específicos para información detallada
|
|
@router.get("/{client_id}/{part_number}/basic", response_model=PartBasicDTO)
|
|
async def get_part_basic_info(
|
|
client_id: int,
|
|
part_number: str,
|
|
db: Session = Depends(get_core_db),
|
|
current_user: dict = Depends(get_current_user),
|
|
):
|
|
"""
|
|
Get basic information for a part
|
|
"""
|
|
# Validate access to the tenant and company
|
|
tenant_id = current_user.get("tenant_id")
|
|
company_id = current_user.get("company_id")
|
|
|
|
if not tenant_id or not company_id:
|
|
raise HTTPException(
|
|
status_code=403, detail="Access denied: Tenant or Company not found"
|
|
)
|
|
|
|
service = PartService(db)
|
|
part = service.get_part(client_id, part_number)
|
|
if not part:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail=f"Part with client_id '{client_id}' and part_number '{part_number}' not found",
|
|
)
|
|
|
|
return PartBasicDTO(
|
|
client_id=part.client_id,
|
|
part_number=part.part_number,
|
|
description_spanish=part.description_spanish,
|
|
description_english=part.description_english,
|
|
part_class=part.part_class,
|
|
unit_cost=part.unit_cost,
|
|
currency_key=part.currency_key,
|
|
is_active=part.is_active,
|
|
)
|
|
|
|
|
|
@router.get("/{client_id}/{part_number}/regulatory", response_model=dict)
|
|
async def get_part_regulatory_info(
|
|
client_id: int,
|
|
part_number: str,
|
|
db: Session = Depends(get_core_db),
|
|
current_user: dict = Depends(get_current_user),
|
|
):
|
|
"""
|
|
Get regulatory information for a part (FDA, FCC, ECCN, etc.)
|
|
"""
|
|
# Validate access to the tenant and company
|
|
tenant_id = current_user.get("tenant_id")
|
|
company_id = current_user.get("company_id")
|
|
|
|
if not tenant_id or not company_id:
|
|
raise HTTPException(
|
|
status_code=403, detail="Access denied: Tenant or Company not found"
|
|
)
|
|
|
|
service = PartService(db)
|
|
part = service.get_part(client_id, part_number)
|
|
if not part:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail=f"Part with client_id '{client_id}' and part_number '{part_number}' not found",
|
|
)
|
|
|
|
return {
|
|
"client_id": part.client_id,
|
|
"part_number": part.part_number,
|
|
"fraction": part.fraction,
|
|
"us_fraction": part.us_fraction,
|
|
"fda_key": part.fda_key,
|
|
"fcc_key": part.fcc_key,
|
|
"license_code": part.license_code,
|
|
"eccn": part.eccn,
|
|
"export_code": part.export_code,
|
|
"exclusion_symbol": part.exclusion_symbol,
|
|
}
|