ADUANASOFT/plantillas-proyectos
13
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Refactor and enhance CRUD operations for Seal, Trailer, Transporter, Vehicle, and Customs Broker modules

Browse Source 
- Updated SealService to support tenant and company filtering with pagination and enhanced CRUD methods.
- Refactored TrailerService to include tenant and company support, added filtering capabilities, and improved CRUD methods.
- Introduced TenantCRUDRoutes for Trailer and Transporter routes to streamline API endpoint creation and management.
- Enhanced TransporterService with tenant and company filtering, pagination, and improved CRUD operations.
- Added Customs Broker module with DTOs, models, services, and routes for managing customs broker data.
- Implemented CRUD operations for Customs Broker, including personnel and VU management.
- Improved data validation and descriptions in DTOs for better API documentation.
This commit is contained in:
acazares
2025-11-11 18:20:39 -06:00
parent b68c4316ff
commit 962f43fe62
39 changed files with 1488 additions and 1366 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
backend/api/v1/common/dto_mixins.py
View File

@@ -1,17 +1,17 @@
from decimal import Decimal
from typing import Optional
from pydantic import BaseModel, Field
from pydantic import Field
class CurrencyMixin(BaseModel):
class CurrencyMixin:
"""Mixin for currency-related fields"""
currency: Optional[str] = Field(None, max_length=3, description="Currency")
currency_factor: Optional[Decimal] = Field(None, description="Currency factor")
class AffectValueMixin(BaseModel):
class AffectValueMixin:
"""Mixin for value affect flags"""
not_affect_usd_value: Optional[int] = Field(
@@ -22,7 +22,7 @@ class AffectValueMixin(BaseModel):
)
class UpdateFlagsMixin(BaseModel):
class UpdateFlagsMixin:
"""Mixin for update flags"""
update_vat: Optional[int] = Field(None, description="Update VAT")

24
backend/api/v1/common/tenant_crud_routes.py
View File

@@ -1,4 +1,4 @@
from typing import Any, Callable, Dict, Generic, Optional, Type, TypeVar
from typing import Any, Callable, Dict, Generic, Optional, Type, TypeVar, Union
from core.database import get_core_db
from core.security import get_current_user, validate_access_to_resource
@@ -49,6 +49,20 @@ class TenantCRUDRoutes(
parent_id_name="pedimento_id",
enable_list=False,
).router
3. Parent resource with string ID (e.g., /vehicles with vehicle_key):
router = TenantCRUDRoutes(
service=VehicleService,
create_schema=VehicleCreate,
update_schema=VehicleUpdate,
response_schema=VehicleResponse,
prefix="/vehicles",
tags=["Vehicles"],
resource_name="Vehicle",
id_name="vehicle_key",
id_type=str, # Specify string type for vehicle_key
enable_list=True,
).router
"""
def __init__(
@@ -61,6 +75,7 @@ class TenantCRUDRoutes(
tags: list[str],
resource_name: str = "Resource",
id_name: Optional[str] = None, # For parent resources (e.g., "pedimento_id")
id_type: Type = int, # Type of the ID (int, str, etc.)
parent_id_name: Optional[
str
] = None, # For child resources (e.g., "pedimento_id")
@@ -78,6 +93,7 @@ class TenantCRUDRoutes(
self.response_schema = response_schema
self.resource_name = resource_name
self.id_name = id_name or parent_id_name or "id"
self.id_type = id_type
self.parent_id_name = parent_id_name
self.db_dependency = db_dependency
self.auth_dependency = auth_dependency
@@ -210,7 +226,7 @@ class TenantCRUDRoutes(
f"/{{{self.id_name}}}", response_model=self.response_schema
)
async def get_resource_by_id(
resource_id: int = Path(..., alias=self.id_name),
resource_id: Union[int, str] = Path(..., alias=self.id_name),
company_id: int = Query(..., description="Company ID"),
db: Session = Depends(self.db_dependency),
current_user: Dict[str, Any] = Depends(self.auth_dependency),
@@ -287,7 +303,7 @@ class TenantCRUDRoutes(
)
async def update_resource_by_id(
data: UpdateSchemaType,
resource_id: int = Path(..., alias=self.id_name),
resource_id: Union[int, str] = Path(..., alias=self.id_name),
company_id: int = Query(..., description="Company ID"),
db: Session = Depends(self.db_dependency),
current_user: Dict[str, Any] = Depends(self.auth_dependency),
@@ -333,7 +349,7 @@ class TenantCRUDRoutes(
# Parent resource
@self.router.delete(f"/{{{self.id_name}}}", status_code=204)
async def delete_resource_by_id(
resource_id: int = Path(..., alias=self.id_name),
resource_id: Union[int, str] = Path(..., alias=self.id_name),
company_id: int = Query(..., description="Company ID"),
db: Session = Depends(self.db_dependency),
current_user: Dict[str, Any] = Depends(self.auth_dependency),

2
backend/api/v1/modules/a76/classes/routes.py
View File

@@ -19,7 +19,7 @@ from .dto import (
)
from .service import ClassService
router = APIRouter(prefix="/classes", tags=["Classes"])
router = APIRouter(prefix="/classes")
@router.get("/", response_model=ClassListDTO)

5
backend/api/v1/modules/a76/client_and_provider/dto.py
View File

@@ -93,7 +93,6 @@ class ClientProviderProgramsDTO(BaseModel):
class ClientProviderCreateDTO(BaseModel):
"""DTO para crear cliente/proveedor"""
client_id: str = Field(..., max_length=8, description="Client ID")
type_nat_foreign: Optional[str] = Field(
None, max_length=1, description="Type national/foreign"
)
@@ -180,7 +179,7 @@ class ClientProviderUpdateDTO(BaseModel):
class ClientProviderResponseDTO(BaseModel):
"""DTO para respuesta de cliente/proveedor"""
client_id: str
id: int
type_nat_foreign: Optional[str] = None
name: Optional[str] = None
short_name: Optional[str] = None
@@ -196,6 +195,8 @@ class ClientProviderResponseDTO(BaseModel):
incoterm: Optional[str] = None
is_national_provider: Optional[str] = None
enabled_disabled: Optional[int] = None
tenant_id: int
company_id: int
# Nested DTOs
address: Optional[ClientProviderAddressDTO] = None

366
backend/api/v1/modules/a76/client_and_provider/routes.py
View File

@@ -5,348 +5,154 @@ Endpoints API para gestión de clientes y proveedores
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 core.security import get_current_user, validate_access_to_resource
from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy.orm import Session
from api.v1.common.tenant_crud_routes import TenantCRUDRoutes
from .dto import (
ClientProviderBasicDTO,
ClientProviderCreateDTO,
ClientProviderListDTO,
ClientProviderResponseDTO,
ClientProviderUpdateDTO,
)
from .service import ClientProviderService
from .models import ClientProvider
# Create base CRUD router using TenantCRUDRoutes factory
base_router = TenantCRUDRoutes(
service=ClientProviderService,
create_schema=ClientProviderCreateDTO,
update_schema=ClientProviderUpdateDTO,
response_schema=ClientProviderResponseDTO,
prefix="/clients-providers",
tags=[],
resource_name="Client/Provider",
id_name="id", # Using numeric ID
enable_list=True, # Enable GET /clients-providers with pagination
enable_filters=True, # Enable filtering
default_page_size=50,
max_page_size=100,
).router
# Create main router to add custom endpoints
router = APIRouter(prefix="/clients-providers")
@router.post(
"/", response_model=ClientProviderResponseDTO, status_code=status.HTTP_201_CREATED
)
async def create_client_provider(
client_data: ClientProviderCreateDTO,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
"""
Create a new client or provider 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"
)
# Ensure the client_data is associated with the correct tenant and company
if client_data.tenant_id != tenant_id or client_data.company_id != company_id:
raise HTTPException(
status_code=400, detail="Mismatch in tenant or company association"
)
service = ClientProviderService(db)
return service.create_client_provider(client_data)
@router.get("/", response_model=ClientProviderListDTO)
async def list_clients_providers(
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"
),
search: Optional[str] = Query(None, description="Search text for name, RFC, or ID"),
client_or_provider: Optional[str] = Query(
None, regex="^[CP]$", description="Filter by type: C=Client, P=Provider"
),
enabled_only: bool = Query(False, description="Show only enabled records"),
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
"""
List clients and providers 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 = ClientProviderService(db)
return service.list_clients_providers(
skip, limit, search, client_or_provider, enabled_only
)
# Include base CRUD routes
router.include_router(base_router, prefix="")
# Custom endpoints
@router.get("/clients", response_model=List[ClientProviderBasicDTO])
async def get_clients_only(
company_id: int = Query(..., description="Company ID"),
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 only clients (client_or_provider = 'C')
"""
# Validate access to the tenant and company
tenant_id = current_user.get("tenant_id")
company_id = current_user.get("company_id")
"""Get only clients (client_or_provider = 'C')"""
tenant_id = validate_access_to_resource(db, company_id, current_user)
if not tenant_id or not company_id:
raise HTTPException(
status_code=403, detail="Access denied: Tenant or Company not found"
clients = (
db.query(ClientProvider)
.filter(
ClientProvider.tenant_id == tenant_id,
ClientProvider.company_id == company_id,
ClientProvider.client_or_provider == "C"
)
service = ClientProviderService(db)
return service.get_clients_only(skip, limit)
.offset(skip)
.limit(limit)
.all()
)
return [ClientProviderBasicDTO.model_validate(c) for c in clients]
@router.get("/providers", response_model=List[ClientProviderBasicDTO])
async def get_providers_only(
company_id: int = Query(..., description="Company ID"),
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 only providers (client_or_provider = 'P')
"""
# Validate access to the tenant and company
tenant_id = current_user.get("tenant_id")
company_id = current_user.get("company_id")
"""Get only providers (client_or_provider = 'P')"""
tenant_id = validate_access_to_resource(db, company_id, current_user)
if not tenant_id or not company_id:
raise HTTPException(
status_code=403, detail="Access denied: Tenant or Company not found"
providers = (
db.query(ClientProvider)
.filter(
ClientProvider.tenant_id == tenant_id,
ClientProvider.company_id == company_id,
ClientProvider.client_or_provider == "P"
)
service = ClientProviderService(db)
return service.get_providers_only(skip, limit)
.offset(skip)
.limit(limit)
.all()
)
return [ClientProviderBasicDTO.model_validate(p) for p in providers]
@router.get("/search/rfc/{rfc}", response_model=List[ClientProviderBasicDTO])
async def search_by_rfc(
rfc: str,
company_id: int = Query(..., description="Company ID"),
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
"""
Search clients/providers by RFC
"""
# Validate access to the tenant and company
tenant_id = current_user.get("tenant_id")
company_id = current_user.get("company_id")
"""Search clients/providers by RFC"""
tenant_id = validate_access_to_resource(db, company_id, current_user)
if not tenant_id or not company_id:
raise HTTPException(
status_code=403, detail="Access denied: Tenant or Company not found"
clients = (
db.query(ClientProvider)
.filter(
ClientProvider.tenant_id == tenant_id,
ClientProvider.company_id == company_id,
ClientProvider.rfc.ilike(f"%{rfc}%")
)
service = ClientProviderService(db)
return service.search_by_rfc(rfc)
@router.get("/{client_id}", response_model=ClientProviderResponseDTO)
async def get_client_provider(
client_id: str,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
"""
Get client/provider by ID with all related 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 = ClientProviderService(db)
client = service.get_client_provider(client_id)
if not client:
raise HTTPException(
status_code=404, detail=f"Client/Provider with ID '{client_id}' not found"
)
return client
@router.put("/{client_id}", response_model=ClientProviderResponseDTO)
async def update_client_provider(
client_id: str,
client_data: ClientProviderUpdateDTO,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
"""
Update client/provider 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 = ClientProviderService(db)
client = service.update_client_provider(client_id, client_data)
if not client:
raise HTTPException(
status_code=404, detail=f"Client/Provider with ID '{client_id}' not found"
)
return client
@router.delete("/{client_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_client_provider(
client_id: str,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
"""
Delete client/provider from the system
Note: This will completely remove the client/provider and all related data.
"""
# 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 = ClientProviderService(db)
if not service.delete_client_provider(client_id):
raise HTTPException(
status_code=404, detail=f"Client/Provider with ID '{client_id}' not found"
.all()
)
return [ClientProviderBasicDTO.model_validate(c) for c in clients]
@router.patch("/{client_id}/toggle-status", response_model=ClientProviderResponseDTO)
async def toggle_client_provider_status(
client_id: str,
client_id: int,
company_id: int = Query(..., description="Company ID"),
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
"""
Toggle client/provider enabled/disabled status
"""
# Validate access to the tenant and company
tenant_id = current_user.get("tenant_id")
company_id = current_user.get("company_id")
"""Toggle client/provider enabled/disabled status"""
tenant_id = validate_access_to_resource(db, company_id, current_user)
if not tenant_id or not company_id:
raise HTTPException(
status_code=403, detail="Access denied: Tenant or Company not found"
)
service = ClientProviderService(db)
client = service.toggle_status(client_id)
client = ClientProviderService.get_by_id(db, client_id, tenant_id, company_id)
if not client:
raise HTTPException(
status_code=404, detail=f"Client/Provider with ID '{client_id}' not found"
)
raise HTTPException(status_code=404, detail="Client/Provider not found")
# Toggle status (1 = enabled, 0 = disabled)
client.enabled_disabled = 1 if client.enabled_disabled == 0 else 0
try:
db.commit()
db.refresh(client)
return client
# Endpoints específicos para información detallada
@router.get("/{client_id}/address", response_model=dict)
async def get_client_provider_address(
client_id: str,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
"""
Get only address information for a client/provider
"""
# 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 = ClientProviderService(db)
client = service.get_client_provider(client_id)
if not client:
raise HTTPException(
status_code=404, detail=f"Client/Provider with ID '{client_id}' not found"
)
return {"client_id": client.client_id, "address": client.address}
@router.get("/{client_id}/programs", response_model=dict)
async def get_client_provider_programs(
client_id: str,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
"""
Get only programs information for a client/provider
"""
# 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 = ClientProviderService(db)
client = service.get_client_provider(client_id)
if not client:
raise HTTPException(
status_code=404, detail=f"Client/Provider with ID '{client_id}' not found"
)
return {"client_id": client.client_id, "programs": client.programs}
except Exception as e:
db.rollback()
raise HTTPException(status_code=500, detail="Error updating status")
@router.get("/{client_id}/basic", response_model=ClientProviderBasicDTO)
async def get_client_provider_basic_info(
client_id: str,
client_id: int,
company_id: int = Query(..., description="Company ID"),
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
"""
Get basic information for a client/provider (without address and programs)
"""
# Validate access to the tenant and company
tenant_id = current_user.get("tenant_id")
company_id = current_user.get("company_id")
"""Get basic information for a client/provider (without address and programs)"""
tenant_id = validate_access_to_resource(db, company_id, current_user)
if not tenant_id or not company_id:
raise HTTPException(
status_code=403, detail="Access denied: Tenant or Company not found"
)
service = ClientProviderService(db)
client = service.get_client_provider(client_id)
client = ClientProviderService.get_by_id(db, client_id, tenant_id, company_id)
if not client:
raise HTTPException(
status_code=404, detail=f"Client/Provider with ID '{client_id}' not found"
)
raise HTTPException(status_code=404, detail="Client/Provider not found")
return ClientProviderBasicDTO(
client_id=client.client_id,
name=client.name,
short_name=client.short_name,
rfc=client.rfc,
client_or_provider=client.client_or_provider,
enabled_disabled=client.enabled_disabled,
)
return ClientProviderBasicDTO.model_validate(client)

256
backend/api/v1/modules/a76/client_and_provider/service.py
View File

@@ -3,7 +3,7 @@ Capa de servicio para lógica de negocio de clientes y proveedores
"""
import logging
from typing import List, Optional
from typing import List, Optional, Tuple, Dict, Any
from fastapi import HTTPException
from sqlalchemy import or_
@@ -28,77 +28,237 @@ class ClientProviderService:
def __init__(self, db: Session):
self.db = db
def create_client_provider(
self, client_data: ClientProviderCreateDTO
) -> ClientProviderResponseDTO:
"""
Crea un nuevo cliente/proveedor en el sistema
# Métodos para TenantCRUDRoutes
@staticmethod
def get_all(
db: Session,
tenant_id: int,
company_id: int,
skip: int = 0,
limit: int = 50,
filters: Optional[Dict[str, Any]] = None,
) -> Tuple[List[ClientProvider], int]:
"""Get all clients/providers for a tenant/company with pagination"""
query = db.query(ClientProvider).filter(
ClientProvider.tenant_id == tenant_id,
ClientProvider.company_id == company_id,
)
Args:
client_data: Datos del cliente/proveedor a crear
# Apply filters if provided
if filters:
if filters.get("search"):
search_pattern = f"%{filters['search']}%"
query = query.filter(
or_(
ClientProvider.name.ilike(search_pattern),
ClientProvider.short_name.ilike(search_pattern),
ClientProvider.rfc.ilike(search_pattern),
)
)
if filters.get("client_or_provider"):
query = query.filter(
ClientProvider.client_or_provider == filters["client_or_provider"]
)
if filters.get("status"):
enabled = 1 if filters["status"] == "enabled" else 0
query = query.filter(ClientProvider.enabled_disabled == enabled)
Returns:
ClientProviderResponseDTO con información del cliente/proveedor creado
total = query.count()
clients = query.options(
joinedload(ClientProvider.address),
joinedload(ClientProvider.programs)
).offset(skip).limit(limit).all()
Raises:
HTTPException: Si el cliente ya existe o error en la creación
"""
try:
# Verificar que no exista el cliente
existing = (
self.db.query(ClientProvider)
.filter(ClientProvider.client_id == client_data.client_id)
return clients, total
@staticmethod
def get_by_id(
db: Session, client_id: int, tenant_id: int, company_id: int
) -> Optional[ClientProvider]:
"""Get client/provider by ID"""
return (
db.query(ClientProvider)
.options(
joinedload(ClientProvider.address),
joinedload(ClientProvider.programs)
)
.filter(
ClientProvider.id == client_id,
ClientProvider.tenant_id == tenant_id,
ClientProvider.company_id == company_id,
)
.first()
)
if existing:
raise HTTPException(
status_code=400,
detail=f"Client with ID '{client_data.client_id}' already exists",
)
# Crear cliente/proveedor principal
@staticmethod
def create(
db: Session,
client_data: ClientProviderCreateDTO,
tenant_id: int,
company_id: int,
) -> ClientProvider:
"""Create a new client/provider"""
try:
# Create main client/provider
data_dict = client_data.model_dump(exclude={"address", "programs"})
db_client = ClientProvider(
client_id=client_data.client_id,
type_nat_foreign=client_data.type_nat_foreign,
name=client_data.name,
short_name=client_data.short_name,
rfc=client_data.rfc,
curp=client_data.curp,
client_or_provider=client_data.client_or_provider,
linking=client_data.linking,
transform_subassembly=client_data.transform_subassembly,
extra_information=client_data.extra_information,
web_key=client_data.web_key,
responsible=client_data.responsible,
position=client_data.position,
incoterm=client_data.incoterm,
is_national_provider=client_data.is_national_provider,
enabled_disabled=client_data.enabled_disabled,
**data_dict,
tenant_id=tenant_id,
company_id=company_id
)
self.db.add(db_client)
self.db.flush() # Para obtener el ID antes del commit
db.add(db_client)
db.flush()
# Crear dirección si se proporciona
# Create address if provided
if client_data.address:
db_address = ClientProviderAddress(
client_id=client_data.client_id,
client_id=db_client.id,
tenant_id=tenant_id,
**client_data.address.model_dump(exclude_unset=True),
)
db.add(db_address)
# Create programs if provided
if client_data.programs:
db_programs = ClientProviderPrograms(
client_id=db_client.id,
tenant_id=tenant_id,
**client_data.programs.model_dump(exclude_unset=True),
)
db.add(db_programs)
db.commit()
db.refresh(db_client)
return db_client
except IntegrityError as e:
db.rollback()
logger.error(f"IntegrityError creating client/provider: {str(e)}")
raise HTTPException(
status_code=400, detail="Client/Provider already exists"
)
except Exception as e:
db.rollback()
logger.error(f"Error creating client/provider: {str(e)}")
raise HTTPException(
status_code=500, detail="Error creating client/provider"
)
@staticmethod
def update(
db: Session,
client_id: int,
tenant_id: int,
company_id: int,
client_data: ClientProviderUpdateDTO,
) -> Optional[ClientProvider]:
"""Update a client/provider"""
client = ClientProviderService.get_by_id(db, client_id, tenant_id, company_id)
if not client:
return None
try:
# Update main fields
update_data = client_data.model_dump(
exclude_unset=True, exclude={"address", "programs"}
)
for field, value in update_data.items():
setattr(client, field, value)
# Update address
if client_data.address:
if client.address:
address_data = client_data.address.model_dump(exclude_unset=True)
for field, value in address_data.items():
setattr(client.address, field, value)
else:
db_address = ClientProviderAddress(
client_id=client.id,
tenant_id=tenant_id,
**client_data.address.model_dump(exclude_unset=True),
)
db.add(db_address)
# Update programs
if client_data.programs:
if client.programs:
programs_data = client_data.programs.model_dump(exclude_unset=True)
for field, value in programs_data.items():
setattr(client.programs, field, value)
else:
db_programs = ClientProviderPrograms(
client_id=client.id,
tenant_id=tenant_id,
**client_data.programs.model_dump(exclude_unset=True),
)
db.add(db_programs)
db.commit()
db.refresh(client)
return client
except Exception as e:
db.rollback()
logger.error(f"Error updating client/provider {client_id}: {str(e)}")
raise HTTPException(
status_code=500, detail="Error updating client/provider"
)
@staticmethod
def delete(
db: Session, client_id: int, tenant_id: int, company_id: int
) -> bool:
"""Delete a client/provider"""
client = ClientProviderService.get_by_id(db, client_id, tenant_id, company_id)
if not client:
return False
try:
db.delete(client)
db.commit()
return True
except Exception as e:
db.rollback()
logger.error(f"Error deleting client/provider {client_id}: {str(e)}")
raise HTTPException(
status_code=500, detail="Error deleting client/provider"
)
# Legacy methods for custom endpoints
def create_client_provider_legacy(
self, client_data: ClientProviderCreateDTO
) -> ClientProviderResponseDTO:
"""Legacy method for creating client/provider"""
try:
# Create main client/provider
data_dict = client_data.model_dump(exclude={"address", "programs"})
db_client = ClientProvider(**data_dict)
self.db.add(db_client)
self.db.flush()
# Create address if provided
if client_data.address:
db_address = ClientProviderAddress(
client_id=db_client.client_id,
**client_data.address.model_dump(exclude_unset=True),
)
self.db.add(db_address)
# Crear programas si se proporciona
# Create programs if provided
if client_data.programs:
db_programs = ClientProviderPrograms(
client_id=client_data.client_id,
client_id=db_client.client_id,
**client_data.programs.model_dump(exclude_unset=True),
)
self.db.add(db_programs)
self.db.commit()
self.db.refresh(db_client)
return self._get_client_with_relations(client_data.client_id)
return self._get_client_with_relations(db_client.client_id)
except IntegrityError as e:
self.db.rollback()

2
backend/api/v1/modules/a76/company/dto.py
View File

@@ -12,8 +12,6 @@ from pydantic import BaseModel, Field
class CompanyCreateDTO(BaseModel):
"""DTO para crear una empresa"""
id: str = Field(default="EMP", max_length=3, description="Company ID")
consecutive: bool = Field(default=True, description="Unique record control")
name: Optional[str] = Field(None, max_length=255, description="Company name")
rfc: Optional[str] = Field(None, max_length=30, description="Company RFC")
main_activity: Optional[str] = Field(

266
backend/api/v1/modules/a76/company/routes.py
View File

@@ -1,119 +1,143 @@
"""
Endpoints API para gestión de empresa
Rutas para gestión de empresa
"""
from typing import Optional
from typing import List
from core.database import get_core_db
from core.security import get_current_user, get_tenant_from_token
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from core.database import get_core_db
from core.security import get_current_user, validate_access_to_resource
from ....common.tenant_crud_routes import TenantCRUDRoutes
from .dto import CompanyCreateDTO, CompanyResponseDTO, CompanyUpdateDTO
from .models import Company
from .service import CompanyService
router = APIRouter(prefix="/company")
# Base CRUD routes using TenantCRUDRoutes
base_router = TenantCRUDRoutes(
service=CompanyService,
create_schema=CompanyCreateDTO,
update_schema=CompanyUpdateDTO,
response_schema=CompanyResponseDTO,
prefix="/company",
tags=[],
id_name="id",
enable_list=True,
enable_filters=True,
).router
# Main router that includes base CRUD
router = APIRouter()
router.include_router(base_router)
@router.post(
"/", response_model=CompanyResponseDTO, status_code=status.HTTP_201_CREATED
# Custom endpoints
@router.get(
"/my-companies",
response_model=List[CompanyResponseDTO],
summary="Get all companies for current tenant",
)
async def create_company(
company_data: CompanyCreateDTO,
async def get_my_companies(
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
"""
Create a new company in the system
Only one company can exist per system due to the unique consecutive field.
"""
service = CompanyService(db)
return service.create_company(company_data)
@router.get("/", response_model=Optional[CompanyResponseDTO])
async def get_company(
db: Session = Depends(get_core_db), current_user: dict = Depends(get_current_user)
):
"""
Get the registered company information
Returns the unique company in the system or None if it doesn't exist.
"""
service = CompanyService(db)
company = service.get_company()
if not company:
raise HTTPException(status_code=404, detail="No company found")
return company
@router.get("/my-companies", response_model=list[CompanyResponseDTO])
async def get_my_companies(
db: Session = Depends(get_core_db), current_user: dict = Depends(get_current_user)
):
"""
Get all companies that belong to the user's tenant
Returns a list of companies associated with the tenant_id from the user's token
"""
tenant_id = get_tenant_from_token(current_user)
"""Get all companies that belong to the current user's tenant"""
tenant_id = current_user.get("tenant_id")
if not tenant_id:
raise HTTPException(status_code=400, detail="Tenant ID not found in token")
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Tenant ID not found in user data",
)
service = CompanyService(db)
companies = service.get_companies_by_tenant(tenant_id)
return companies
return [CompanyResponseDTO.model_validate(company) for company in companies]
@router.get("/status/exists", response_model=dict)
@router.get(
"/status/exists",
response_model=dict,
summary="Check if company exists for tenant",
)
async def check_company_exists(
db: Session = Depends(get_core_db), current_user: dict = Depends(get_current_user)
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
"""
Check if a company is registered in the system
"""
"""Check if a company exists for the current tenant"""
tenant_id = current_user.get("tenant_id")
if not tenant_id:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Tenant ID not found in user data",
)
service = CompanyService(db)
exists = service.exists_company()
return {
"exists": exists,
"message": "Company found" if exists else "No company registered",
}
exists = service.exists_company(tenant_id)
return {"exists": exists}
# Specific endpoints for important fields
@router.get("/info/basic", response_model=dict)
async def get_company_basic_info(
db: Session = Depends(get_core_db), current_user: dict = Depends(get_current_user)
@router.get(
"/info/basic/{company_id}",
response_model=dict,
summary="Get basic company info",
)
async def get_basic_info(
company_id: int,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
"""
Get basic company information (name, RFC, main activity)
"""
service = CompanyService(db)
company = service.get_company()
"""Get basic information about a company"""
tenant_id = current_user.get("tenant_id")
company_id_from_user = current_user.get("company_id")
# Validate access
validate_access_to_resource(
db, tenant_id, company_id_from_user, Company, company_id, "id"
)
company = CompanyService.get_by_id(db, company_id, tenant_id, company_id_from_user)
if not company:
raise HTTPException(status_code=404, detail="No company found")
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Company not found",
)
return {
"id": company.id,
"name": company.name,
"rfc": company.rfc,
"main_activity": company.main_activity,
"logo": company.logo,
"program": company.program,
}
@router.get("/info/responsible", response_model=dict)
async def get_company_responsible_info(
db: Session = Depends(get_core_db), current_user: dict = Depends(get_current_user)
@router.get(
"/info/responsible/{company_id}",
response_model=dict,
summary="Get responsible person info",
)
async def get_responsible_info(
company_id: int,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
"""
Get company responsible person information
"""
service = CompanyService(db)
company = service.get_company()
"""Get responsible person information for a company"""
tenant_id = current_user.get("tenant_id")
company_id_from_user = current_user.get("company_id")
# Validate access
validate_access_to_resource(
db, tenant_id, company_id_from_user, Company, company_id, "id"
)
company = CompanyService.get_by_id(db, company_id, tenant_id, company_id_from_user)
if not company:
raise HTTPException(status_code=404, detail="No company found")
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Company not found",
)
return {
"responsible": company.responsible,
@@ -125,77 +149,35 @@ async def get_company_responsible_info(
}
@router.get("/info/program", response_model=dict)
async def get_company_program_info(
db: Session = Depends(get_core_db), current_user: dict = Depends(get_current_user)
@router.get(
"/info/program/{company_id}",
response_model=dict,
summary="Get program information",
)
async def get_program_info(
company_id: int,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
"""
Get company program information
"""
service = CompanyService(db)
company = service.get_company()
"""Get program information for a company"""
tenant_id = current_user.get("tenant_id")
company_id_from_user = current_user.get("company_id")
# Validate access
validate_access_to_resource(
db, tenant_id, company_id_from_user, Company, company_id, "id"
)
company = CompanyService.get_by_id(db, company_id, tenant_id, company_id_from_user)
if not company:
raise HTTPException(status_code=404, detail="No company found")
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Company not found",
)
return {
"program": company.program,
"program_number": company.program_number,
"prosec": company.prosec,
"prosec_authorization": company.prosec_authorization,
"manufacturer_id": company.manufacturer_id,
}
@router.get("/{company_id}", response_model=CompanyResponseDTO)
async def get_company_by_id(
company_id: str,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
"""
Get company by specific ID
"""
service = CompanyService(db)
company = service.get_company_by_id(company_id)
if not company:
raise HTTPException(
status_code=404, detail=f"Company with ID '{company_id}' not found"
)
return company
@router.put("/{company_id}", response_model=CompanyResponseDTO)
async def update_company(
company_id: str,
company_data: CompanyUpdateDTO,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
"""
Update company information
"""
service = CompanyService(db)
company = service.update_company(company_id, company_data)
if not company:
raise HTTPException(
status_code=404, detail=f"Company with ID '{company_id}' not found"
)
return company
@router.delete("/{company_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_company(
company_id: str,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
"""
Delete company from the system
Note: This will completely remove the company from the system.
"""
service = CompanyService(db)
if not service.delete_company(company_id):
raise HTTPException(
status_code=404, detail=f"Company with ID '{company_id}' not found"
)

234
backend/api/v1/modules/a76/company/service.py
View File

@@ -3,7 +3,7 @@ Capa de servicio para lógica de negocio de empresa
"""
import logging
from typing import List, Optional
from typing import List, Optional, Tuple, Dict, Any
from fastapi import HTTPException
from sqlalchemy.exc import IntegrityError
@@ -21,187 +21,141 @@ class CompanyService:
def __init__(self, db: Session):
self.db = db
def create_company(self, company_data: CompanyCreateDTO) -> CompanyResponseDTO:
"""
Crea una nueva empresa en el sistema
# Métodos para TenantCRUDRoutes
@staticmethod
def get_all(
db: Session,
tenant_id: int,
company_id: int,
skip: int = 0,
limit: int = 50,
filters: Optional[Dict[str, Any]] = None,
) -> Tuple[List[Company], int]:
"""Get all companies for a tenant with pagination"""
query = db.query(Company).filter(Company.tenant_id == tenant_id)
Args:
company_data: Datos de la empresa a crear
# Apply filters if provided
if filters:
if filters.get("name"):
query = query.filter(
Company.name.ilike(f"%{filters['name']}%")
)
if filters.get("rfc"):
query = query.filter(
Company.rfc.ilike(f"%{filters['rfc']}%")
)
Returns:
CompanyResponseDTO con información de la empresa creada
total = query.count()
companies = query.offset(skip).limit(limit).all()
Raises:
HTTPException: Si ya existe una empresa o error en la creación
"""
return companies, total
@staticmethod
def get_by_id(
db: Session, company_id: int, tenant_id: int, company_id_unused: int
) -> Optional[Company]:
"""Get company by ID"""
return (
db.query(Company)
.filter(
Company.id == company_id,
Company.tenant_id == tenant_id,
)
.first()
)
@staticmethod
def create(
db: Session,
company_data: CompanyCreateDTO,
tenant_id: int,
company_id: int,
) -> Company:
"""Create a new company"""
try:
# Verificar que no exista ya una empresa (solo puede haber una por el consecutivo único)
existing = self.db.query(Company).filter(Company.id).first()
if existing:
raise HTTPException(
status_code=400,
detail="A company is already registered in the system",
)
# Crear empresa
db_company = Company(
id=company_data.id,
consecutive=company_data.consecutive,
name=company_data.name,
rfc=company_data.rfc,
main_activity=company_data.main_activity,
program=company_data.program,
program_number=company_data.program_number,
prosec=company_data.prosec,
prosec_authorization=company_data.prosec_authorization,
manufacturer_id=company_data.manufacturer_id,
broker_company=company_data.broker_company,
responsible=company_data.responsible,
responsible_name=company_data.responsible_name,
responsible_last_name=company_data.responsible_last_name,
responsible_mother_last_name=company_data.responsible_mother_last_name,
responsible_rfc=company_data.responsible_rfc,
position=company_data.position,
logo=company_data.logo,
has_express_line=company_data.has_express_line,
order_format_type=company_data.order_format_type,
previous_code=company_data.previous_code,
is_service_company=company_data.is_service_company,
client_name=company_data.client_name,
subassembly_mode=company_data.subassembly_mode,
curp=company_data.curp,
inter_db_name=company_data.inter_db_name,
ctpat_svi=company_data.ctpat_svi,
trusted_exporter_number=company_data.trusted_exporter_number,
prevalidator_key=company_data.prevalidator_key,
seventh_amendment=company_data.seventh_amendment,
**company_data.model_dump(exclude_unset=True),
tenant_id=tenant_id
)
self.db.add(db_company)
self.db.commit()
self.db.refresh(db_company)
db.add(db_company)
db.commit()
db.refresh(db_company)
return CompanyResponseDTO.model_validate(db_company)
return db_company
except IntegrityError as e:
self.db.rollback()
db.rollback()
logger.error(f"IntegrityError creating company: {str(e)}")
raise HTTPException(
status_code=400,
detail="Integrity error: A company already exists in the system",
detail="Company already exists",
)
except HTTPException:
raise
except Exception as e:
self.db.rollback()
db.rollback()
logger.error(f"Error creating company: {str(e)}")
raise HTTPException(status_code=500, detail="Error creating company")
def get_company(self) -> Optional[CompanyResponseDTO]:
"""
Obtiene la empresa (solo puede haber una)
Returns:
CompanyResponseDTO o None si no existe
"""
company = self.db.query(Company).filter(Company.id).first()
if not company:
return None
return CompanyResponseDTO.model_validate(company)
def get_company_by_id(self, company_id: str) -> Optional[CompanyResponseDTO]:
"""
Obtiene una empresa por ID
Args:
company_id: ID de la empresa
Returns:
CompanyResponseDTO o None si no existe
"""
company = self.db.query(Company).filter(Company.id == company_id).first()
if not company:
return None
return CompanyResponseDTO.model_validate(company)
def update_company(
self, company_id: str, company_data: CompanyUpdateDTO
) -> Optional[CompanyResponseDTO]:
"""
Actualiza una empresa
Args:
company_id: ID de la empresa a actualizar
company_data: Datos a actualizar
Returns:
CompanyResponseDTO actualizada o None si no existe
"""
company = self.db.query(Company).filter(Company.id == company_id).first()
@staticmethod
def update(
db: Session,
company_id: int,
tenant_id: int,
company_id_unused: int,
company_data: CompanyUpdateDTO,
) -> Optional[Company]:
"""Update a company"""
company = CompanyService.get_by_id(db, company_id, tenant_id, company_id_unused)
if not company:
return None
# Actualizar solo campos proporcionados
# Update only provided fields
update_data = company_data.model_dump(exclude_unset=True)
for field, value in update_data.items():
setattr(company, field, value)
try:
self.db.commit()
self.db.refresh(company)
return CompanyResponseDTO.model_validate(company)
db.commit()
db.refresh(company)
return company
except Exception as e:
self.db.rollback()
db.rollback()
logger.error(f"Error updating company {company_id}: {str(e)}")
raise HTTPException(status_code=500, detail="Error updating company")
def delete_company(self, company_id: str) -> bool:
"""
Elimina una empresa
Args:
company_id: ID de la empresa a eliminar
Returns:
True si se eliminó, False si no existe
"""
company = self.db.query(Company).filter(Company.id == company_id).first()
@staticmethod
def delete(
db: Session, company_id: int, tenant_id: int, company_id_unused: int
) -> bool:
"""Delete a company"""
company = CompanyService.get_by_id(db, company_id, tenant_id, company_id_unused)
if not company:
return False
try:
self.db.delete(company)
self.db.commit()
db.delete(company)
db.commit()
return True
except Exception as e:
self.db.rollback()
db.rollback()
logger.error(f"Error deleting company {company_id}: {str(e)}")
raise HTTPException(status_code=500, detail="Error deleting company")
def exists_company(self) -> bool:
"""
Verifica si existe una empresa registrada
Returns:
True si existe una empresa, False en caso contrario
"""
return self.db.query(Company).filter(Company.id).first() is not None
def get_companies_by_tenant(self, tenant_id: int) -> List[CompanyResponseDTO]:
"""
Obtiene todas las compañías que pertenecen a un tenant específico
Args:
tenant_id: ID del tenant
Returns:
Lista de CompanyResponseDTO
"""
companies = (
# Custom methods
def get_companies_by_tenant(self, tenant_id: int) -> List[Company]:
"""Get all companies for a tenant"""
return (
self.db.query(Company)
.filter(Company.tenant_id == tenant_id)
.order_by(Company.name)
.all()
)
return [CompanyResponseDTO.model_validate(company) for company in companies]
def exists_company(self, tenant_id: int) -> bool:
"""Check if a company exists for a tenant"""
return (
self.db.query(Company)
.filter(Company.tenant_id == tenant_id)
.first()
is not None
)

2
backend/api/v1/modules/a76/country_rule_oct/routes.py
View File

@@ -8,7 +8,7 @@ from sqlalchemy.orm import Session
from .dto import CountryRuleOctCreateDTO, CountryRuleOctResponseDTO
from .services import CountryRuleOctService
router = APIRouter(prefix="/country-rule-oct", tags=["CountryRuleOct"])
router = APIRouter(prefix="/country-rule-oct")
@router.get("/", response_model=List[CountryRuleOctResponseDTO])

0
backend/api/v1/modules/a76/aduanal/dto.py → backend/api/v1/modules/a76/customs_brokers/dto.py
View File

0
backend/api/v1/modules/a76/aduanal/models.py → backend/api/v1/modules/a76/customs_brokers/models.py
View File

0
backend/api/v1/modules/a76/aduanal/routes.py → backend/api/v1/modules/a76/customs_brokers/routes.py
View File

0
backend/api/v1/modules/a76/aduanal/services.py → backend/api/v1/modules/a76/customs_brokers/services.py
View File

2
backend/api/v1/modules/a76/drivers/routes.py
View File

@@ -8,7 +8,7 @@ from sqlalchemy.orm import Session
from .dto import DriverCreateDTO, DriverResponseDTO
from .services import DriverService
router = APIRouter(prefix="/drivers", tags=["Drivers"])
router = APIRouter(prefix="/drivers")
@router.get("/", response_model=List[DriverResponseDTO])

21
backend/api/v1/modules/a76/exchange_rate/dto.py
View File

@@ -1,19 +1,30 @@
from typing import Optional
from pydantic import BaseModel
from pydantic import BaseModel, Field
class ExchangeRateBaseDTO(BaseModel):
date: int
value: Optional[float]
local_currency: Optional[str]
foreign_currency: Optional[str]
date: int = Field(..., description="Exchange rate date")
value: Optional[float] = Field(None, description="Exchange rate value")
local_currency: Optional[str] = Field(None, max_length=7, description="Local currency code")
foreign_currency: Optional[str] = Field(None, max_length=7, description="Foreign currency code")
class ExchangeRateCreateDTO(ExchangeRateBaseDTO):
"""Schema for creating an exchange rate"""
pass
class ExchangeRateUpdateDTO(ExchangeRateBaseDTO):
"""Schema for updating an exchange rate"""
date: Optional[int] = Field(None, description="Exchange rate date")
class ExchangeRateResponseDTO(ExchangeRateBaseDTO):
"""Schema for exchange rate response"""
id: int
company_id: int
tenant_id: int
class Config:
from_attributes = True

114
backend/api/v1/modules/a76/exchange_rate/routes.py
View File

@@ -1,100 +1,20 @@
from typing import List
from api.v1.common.tenant_crud_routes import TenantCRUDRoutes
from core.database import get_core_db
from core.security import get_current_user
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from .dto import ExchangeRateCreateDTO, ExchangeRateResponseDTO
from .dto import ExchangeRateCreateDTO, ExchangeRateResponseDTO, ExchangeRateUpdateDTO
from .services import ExchangeRateService
router = APIRouter(prefix="/exchange-rate", tags=["ExchangeRate"])
@router.get("/", response_model=List[ExchangeRateResponseDTO])
async def list_exchange_rates(
db: Session = Depends(get_core_db), current_user: dict = Depends(get_current_user)
):
"""
List all ExchangeRate entries.
"""
# 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"
)
return db.query(ExchangeRateService).all()
@router.get("/{date}", response_model=ExchangeRateResponseDTO)
async def read_exchange_rate(
date: int,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
"""
Get a specific ExchangeRate by its date.
"""
# 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"
)
exchange_rate = ExchangeRateService.get_exchange_rate_by_date(db, date)
if not exchange_rate:
raise HTTPException(status_code=404, detail="ExchangeRate not found")
return exchange_rate
@router.post(
"/", response_model=ExchangeRateResponseDTO, status_code=status.HTTP_201_CREATED
)
async def create_exchange_rate(
exchange_rate_data: ExchangeRateCreateDTO,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
"""
Create a new ExchangeRate entry.
"""
# 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"
)
return ExchangeRateService.create_exchange_rate(db, exchange_rate_data)
@router.delete("/{date}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_exchange_rate(
date: int,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
"""
Delete an ExchangeRate by its date.
"""
# 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"
)
exchange_rate = ExchangeRateService.delete_exchange_rate(db, date)
if not exchange_rate:
raise HTTPException(status_code=404, detail="ExchangeRate not found")
# Create router using TenantCRUDRoutes factory
router = TenantCRUDRoutes(
service=ExchangeRateService,
create_schema=ExchangeRateCreateDTO,
update_schema=ExchangeRateUpdateDTO,
response_schema=ExchangeRateResponseDTO,
prefix="/exchange-rate",
tags=[],
resource_name="Exchange Rate",
id_name="id", # Using numeric ID
enable_list=True, # Enable GET /exchange-rate with pagination
enable_filters=True, # Enable filtering by date, local_currency, foreign_currency
default_page_size=50,
max_page_size=100,
).router

101
backend/api/v1/modules/a76/exchange_rate/services.py
View File

@@ -1,31 +1,112 @@
from typing import Optional, Tuple, List, Dict, Any
from sqlalchemy.orm import Session
from . import dto, models
class ExchangeRateService:
"""Service for ExchangeRate CRUD operations with tenant support"""
@staticmethod
def get_exchange_rate_by_date(db: Session, date: int):
def get_all(
db: Session,
tenant_id: int,
company_id: int,
skip: int = 0,
limit: int = 50,
filters: Optional[Dict[str, Any]] = None,
) -> Tuple[List[models.ExchangeRate], int]:
"""Get all exchange rates for a tenant/company with pagination"""
query = db.query(models.ExchangeRate).filter(
models.ExchangeRate.tenant_id == tenant_id,
models.ExchangeRate.company_id == company_id,
)
# Apply filters if provided
if filters:
if filters.get("date"):
query = query.filter(models.ExchangeRate.date == filters["date"])
if filters.get("local_currency"):
query = query.filter(
models.ExchangeRate.local_currency == filters["local_currency"]
)
if filters.get("foreign_currency"):
query = query.filter(
models.ExchangeRate.foreign_currency == filters["foreign_currency"]
)
total = query.count()
exchange_rates = query.order_by(models.ExchangeRate.date.desc()).offset(skip).limit(limit).all()
return exchange_rates, total
@staticmethod
def get_by_id(
db: Session, exchange_rate_id: int, tenant_id: int, company_id: int
) -> Optional[models.ExchangeRate]:
"""Get exchange rate by ID"""
return (
db.query(models.ExchangeRate)
.filter(models.ExchangeRate.date == date)
.filter(
models.ExchangeRate.id == exchange_rate_id,
models.ExchangeRate.tenant_id == tenant_id,
models.ExchangeRate.company_id == company_id,
)
.first()
)
@staticmethod
def create_exchange_rate(
db: Session, exchange_rate_data: dto.ExchangeRateCreateDTO
):
new_exchange_rate = models.ExchangeRate(**exchange_rate_data.dict())
def create(
db: Session,
exchange_rate_data: dto.ExchangeRateCreateDTO,
tenant_id: int,
company_id: int,
) -> models.ExchangeRate:
"""Create a new exchange rate"""
new_exchange_rate = models.ExchangeRate(
**exchange_rate_data.model_dump(), tenant_id=tenant_id, company_id=company_id
)
db.add(new_exchange_rate)
db.commit()
db.refresh(new_exchange_rate)
return new_exchange_rate
@staticmethod
def delete_exchange_rate(db: Session, date: int):
exchange_rate = ExchangeRateService.get_exchange_rate_by_date(db, date)
if exchange_rate:
def update(
db: Session,
exchange_rate_id: int,
tenant_id: int,
company_id: int,
exchange_rate_data: dto.ExchangeRateUpdateDTO,
) -> Optional[models.ExchangeRate]:
"""Update an exchange rate"""
exchange_rate = ExchangeRateService.get_by_id(
db, exchange_rate_id, tenant_id, company_id
)
if not exchange_rate:
return None
# Update fields
update_data = exchange_rate_data.model_dump(exclude_unset=True)
for field, value in update_data.items():
setattr(exchange_rate, field, value)
db.commit()
db.refresh(exchange_rate)
return exchange_rate
@staticmethod
def delete(
db: Session, exchange_rate_id: int, tenant_id: int, company_id: int
) -> bool:
"""Delete an exchange rate"""
exchange_rate = ExchangeRateService.get_by_id(
db, exchange_rate_id, tenant_id, company_id
)
if not exchange_rate:
return False
db.delete(exchange_rate)
db.commit()
return exchange_rate
return True

2
backend/api/v1/modules/a76/fraction_rule_octave/routes.py
View File

@@ -8,7 +8,7 @@ from sqlalchemy.orm import Session
from .dto import FractionRuleOctaveCreateDTO, FractionRuleOctaveResponseDTO
from .services import FractionRuleOctaveService
router = APIRouter(prefix="/fraction_rule_octave", tags=["FractionRuleOctave"])
router = APIRouter(prefix="/fraction_rule_octave")
@router.get("/", response_model=List[FractionRuleOctaveResponseDTO])

46
backend/api/v1/modules/a76/package/dto.py
View File

@@ -1,40 +1,40 @@
"""
DTOs for GBultos.
DTOs for Packages (GBultos).
"""
from typing import Optional
from pydantic import BaseModel
from pydantic import BaseModel, Field
class GBultoBaseDTO(BaseModel):
CODE: str
DESCRIPTION: Optional[str]
DESCRIPTIONI: Optional[str]
WEIGHT_UNIT: Optional[float]
PLURALS: Optional[str]
PLURAL_IN: Optional[str]
CODE_ACE: Optional[str]
CODE_AAMEX: Optional[str]
class PackageBaseDTO(BaseModel):
key: str = Field(..., description="Package key (primary identifier)", max_length=5)
description_es: Optional[str] = Field(None, description="Description in Spanish", max_length=40)
description_en: Optional[str] = Field(None, description="Description in English", max_length=40)
weight_unit: Optional[float] = Field(None, description="Weight unit")
plurals: Optional[str] = Field(None, max_length=4)
plural_in: Optional[str] = Field(None, max_length=4)
code_ace: Optional[str] = Field(None, max_length=4)
code_aamex: Optional[str] = Field(None, max_length=9)
class GBultoCreateDTO(GBultoBaseDTO):
class PackageCreateDTO(PackageBaseDTO):
"""Schema for creating a package"""
pass
class GBultoUpdateDTO(BaseModel):
DESCRIPTION: Optional[str]
DESCRIPTIONI: Optional[str]
WEIGHT_UNIT: Optional[float]
PLURALS: Optional[str]
PLURAL_IN: Optional[str]
CODE_ACE: Optional[str]
CODE_AAMEX: Optional[str]
class PackageUpdateDTO(PackageBaseDTO):
"""Schema for updating a package"""
key: Optional[str] = Field(None, description="Package key (cannot be modified)", max_length=5)
class GBultoResponseDTO(GBultoBaseDTO):
CREATED_AT: Optional[str]
UPDATED_AT: Optional[str]
class PackageResponseDTO(PackageBaseDTO):
"""Schema for package response"""
id: int
company_id: int
tenant_id: int
created_at: Optional[str] = None
updated_at: Optional[str] = None
class Config:
from_attributes = True

145
backend/api/v1/modules/a76/package/routes.py
View File

@@ -1,127 +1,24 @@
from typing import List
"""
Routes for managing Packages (GBultos).
"""
from core.database import get_core_db
from core.security import get_current_user
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from api.v1.common.tenant_crud_routes import TenantCRUDRoutes
from .dto import GBultoCreateDTO, GBultoResponseDTO, GBultoUpdateDTO
from .models import Package
from .services import GBultoService
from .dto import PackageCreateDTO, PackageResponseDTO, PackageUpdateDTO
from .services import PackageService
router = APIRouter(prefix="/bultos", tags=["GBultos"])
@router.get("/", response_model=List[GBultoResponseDTO])
async def list_bultos(
skip: int = 0,
limit: int = 100,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
"""
List all GBultos with 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"
)
return db.query(Package).offset(skip).limit(limit).all()
@router.get("/{code}", response_model=GBultoResponseDTO)
async def read_bulto(
code: str,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
"""
Get a specific Package by its CODE.
"""
# 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"
)
bulto = GBultoService.get_bulto_by_code(db, code)
if not bulto:
raise HTTPException(status_code=404, detail="Package not found")
return bulto
@router.post("/", response_model=GBultoResponseDTO, status_code=status.HTTP_201_CREATED)
async def create_gbulto(
bulto_data: GBultoCreateDTO,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
"""
Create a new Package.
"""
# 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"
)
return GBultoService.create_gbulto(db, bulto_data)
@router.put("/{code}", response_model=GBultoResponseDTO)
async def update_bulto(
code: str,
bulto_data: GBultoUpdateDTO,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
"""
Update an existing Package.
"""
# 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"
)
bulto = GBultoService.update_bulto(db, code, bulto_data)
if not bulto:
raise HTTPException(status_code=404, detail="Package not found")
return bulto
@router.delete("/{code}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_bulto(
code: str,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
"""
Delete a Package by its CODE.
"""
# 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"
)
bulto = GBultoService.delete_bulto(db, code)
if not bulto:
raise HTTPException(status_code=404, detail="Package not found")
# Create router using TenantCRUDRoutes factory
router = TenantCRUDRoutes(
service=PackageService,
create_schema=PackageCreateDTO,
update_schema=PackageUpdateDTO,
response_schema=PackageResponseDTO,
prefix="/package",
tags=[],
resource_name="Package",
id_name="id", # Using numeric ID
enable_list=True, # Enable GET /package with pagination
enable_filters=True, # Enable filtering by key and description_es
default_page_size=50,
max_page_size=100,
).router

117
backend/api/v1/modules/a76/package/services.py
View File

@@ -1,39 +1,110 @@
"""
Service layer for Packages (GBultos).
"""
from typing import Optional, Tuple, List, Dict, Any
from sqlalchemy.orm import Session
from . import dto, models
class GBultoService:
"""
Service layer for GBultos.
"""
class PackageService:
"""Service for Package CRUD operations with tenant support"""
@staticmethod
def get_bulto_by_code(db: Session, code: str):
return db.query(models.Package).filter(models.Package.CODE == code).first()
def get_all(
db: Session,
tenant_id: int,
company_id: int,
skip: int = 0,
limit: int = 50,
filters: Optional[Dict[str, Any]] = None,
) -> Tuple[List[models.Package], int]:
"""Get all packages for a tenant/company with pagination"""
query = db.query(models.Package).filter(
models.Package.tenant_id == tenant_id,
models.Package.company_id == company_id,
)
# Apply filters if provided
if filters:
if filters.get("key"):
query = query.filter(
models.Package.key.ilike(f"%{filters['key']}%")
)
if filters.get("description_es"):
query = query.filter(
models.Package.description_es.ilike(f"%{filters['description_es']}%")
)
total = query.count()
packages = query.offset(skip).limit(limit).all()
return packages, total
@staticmethod
def create_gbulto(db: Session, gbulto_data: dto.GBultoCreateDTO):
new_gbulto = models.Package(**gbulto_data.dict())
db.add(new_gbulto)
def get_by_id(
db: Session, package_id: int, tenant_id: int, company_id: int
) -> Optional[models.Package]:
"""Get package by ID"""
return (
db.query(models.Package)
.filter(
models.Package.id == package_id,
models.Package.tenant_id == tenant_id,
models.Package.company_id == company_id,
)
.first()
)
@staticmethod
def create(
db: Session,
package_data: dto.PackageCreateDTO,
tenant_id: int,
company_id: int,
) -> models.Package:
"""Create a new package"""
new_package = models.Package(
**package_data.model_dump(), tenant_id=tenant_id, company_id=company_id
)
db.add(new_package)
db.commit()
db.refresh(new_gbulto)
return new_gbulto
db.refresh(new_package)
return new_package
@staticmethod
def update_bulto(db: Session, code: str, bulto_data: dto.GBultoUpdateDTO):
bulto = GBultoService.get_bulto_by_code(db, code)
if bulto:
for key, value in bulto_data.dict(exclude_unset=True).items():
setattr(bulto, key, value)
def update(
db: Session,
package_id: int,
tenant_id: int,
company_id: int,
package_data: dto.PackageUpdateDTO,
) -> Optional[models.Package]:
"""Update a package"""
package = PackageService.get_by_id(db, package_id, tenant_id, company_id)
if not package:
return None
# Update fields (excluding key if it's meant to be immutable)
update_data = package_data.model_dump(exclude_unset=True, exclude={"key"})
for field, value in update_data.items():
setattr(package, field, value)
db.commit()
db.refresh(bulto)
return bulto
db.refresh(package)
return package
@staticmethod
def delete_bulto(db: Session, code: str):
bulto = GBultoService.get_bulto_by_code(db, code)
if bulto:
db.delete(bulto)
def delete(
db: Session, package_id: int, tenant_id: int, company_id: int
) -> bool:
"""Delete a package"""
package = PackageService.get_by_id(db, package_id, tenant_id, company_id)
if not package:
return False
db.delete(package)
db.commit()
return bulto
return True

2
backend/api/v1/modules/a76/pedmientos/dtos/pedimento_config_updates.py
View File

@@ -3,7 +3,7 @@ from typing import Optional
from pydantic import BaseModel, ConfigDict, Field
from backend.api.v1.common.dto_mixins import UpdateFlagsMixin
from api.v1.common.dto_mixins import UpdateFlagsMixin
class PedimentoConfigUpdatesBase(BaseModel, UpdateFlagsMixin):

23
backend/api/v1/modules/a76/permission_rule_oct/dto.py
View File

@@ -1,20 +1,31 @@
from typing import Optional
from pydantic import BaseModel
from pydantic import BaseModel, Field
class PermissionRuleOctBaseDTO(BaseModel):
permission: str
start_date: Optional[int]
end_date: Optional[int]
sector: Optional[str]
system: Optional[str]
permission: str = Field(..., description="Permission identifier", max_length=20)
start_date: Optional[int] = Field(None, description="Start date")
end_date: Optional[int] = Field(None, description="End date")
sector: Optional[str] = Field(None, max_length=8, description="Sector")
system: Optional[str] = Field(None, max_length=5, description="System")
class PermissionRuleOctCreateDTO(PermissionRuleOctBaseDTO):
"""Schema for creating a permission rule OCT"""
pass
class PermissionRuleOctUpdateDTO(PermissionRuleOctBaseDTO):
"""Schema for updating a permission rule OCT"""
permission: Optional[str] = Field(None, description="Permission identifier", max_length=20)
class PermissionRuleOctResponseDTO(PermissionRuleOctBaseDTO):
"""Schema for permission rule OCT response"""
id: int
company_id: int
tenant_id: int
class Config:
from_attributes = True

120
backend/api/v1/modules/a76/permission_rule_oct/routes.py
View File

@@ -1,102 +1,24 @@
from typing import List
from api.v1.common.tenant_crud_routes import TenantCRUDRoutes
from core.database import get_core_db
from core.security import get_current_user
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from .dto import PermissionRuleOctCreateDTO, PermissionRuleOctResponseDTO
from .dto import (
PermissionRuleOctCreateDTO,
PermissionRuleOctResponseDTO,
PermissionRuleOctUpdateDTO,
)
from .services import PermissionRuleOctService
router = APIRouter(prefix="/permission-rule-oct", tags=["PermissionRuleOct"])
@router.get("/", response_model=List[PermissionRuleOctResponseDTO])
async def list_permissions(
db: Session = Depends(get_core_db), current_user: dict = Depends(get_current_user)
):
"""
List all PermissionRuleOct entries.
"""
# 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"
)
return db.query(PermissionRuleOctService).all()
@router.get("/{permission}", response_model=PermissionRuleOctResponseDTO)
async def read_permission(
permission: str,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
"""
Get a specific PermissionRuleOct by its permission.
"""
# 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"
)
permission = PermissionRuleOctService.get_permission_by_id(db, permission)
if not permission:
raise HTTPException(status_code=404, detail="PermissionRuleOct not found")
return permission
@router.post(
"/",
response_model=PermissionRuleOctResponseDTO,
status_code=status.HTTP_201_CREATED,
)
async def create_permission(
permission_data: PermissionRuleOctCreateDTO,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
"""
Create a new PermissionRuleOct entry.
"""
# 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"
)
return PermissionRuleOctService.create_permission(db, permission_data)
@router.delete("/{permission}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_permission(
permission: str,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
"""
Delete a PermissionRuleOct by its permission.
"""
# 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"
)
permission = PermissionRuleOctService.delete_permission(db, permission)
if not permission:
raise HTTPException(status_code=404, detail="PermissionRuleOct not found")
# Create router using TenantCRUDRoutes factory
router = TenantCRUDRoutes(
service=PermissionRuleOctService,
create_schema=PermissionRuleOctCreateDTO,
update_schema=PermissionRuleOctUpdateDTO,
response_schema=PermissionRuleOctResponseDTO,
prefix="/permission-rule-oct",
tags=[],
resource_name="Permission Rule OCT",
id_name="id", # Using numeric ID
enable_list=True, # Enable GET /permission-rule-oct with pagination
enable_filters=True, # Enable filtering by permission, sector, system
default_page_size=50,
max_page_size=100,
).router

101
backend/api/v1/modules/a76/permission_rule_oct/services.py
View File

@@ -1,29 +1,114 @@
from typing import Optional, Tuple, List, Dict, Any
from sqlalchemy.orm import Session
from . import dto, models
class PermissionRuleOctService:
"""Service for PermissionRuleOct CRUD operations with tenant support"""
@staticmethod
def get_permission_by_id(db: Session, permission: str):
def get_all(
db: Session,
tenant_id: int,
company_id: int,
skip: int = 0,
limit: int = 50,
filters: Optional[Dict[str, Any]] = None,
) -> Tuple[List[models.PermissionRuleOct], int]:
"""Get all permission rules OCT for a tenant/company with pagination"""
query = db.query(models.PermissionRuleOct).filter(
models.PermissionRuleOct.tenant_id == tenant_id,
models.PermissionRuleOct.company_id == company_id,
)
# Apply filters if provided
if filters:
if filters.get("permission"):
query = query.filter(
models.PermissionRuleOct.permission.ilike(f"%{filters['permission']}%")
)
if filters.get("sector"):
query = query.filter(
models.PermissionRuleOct.sector == filters["sector"]
)
if filters.get("system"):
query = query.filter(
models.PermissionRuleOct.system == filters["system"]
)
total = query.count()
permissions = query.offset(skip).limit(limit).all()
return permissions, total
@staticmethod
def get_by_id(
db: Session, permission_id: int, tenant_id: int, company_id: int
) -> Optional[models.PermissionRuleOct]:
"""Get permission rule OCT by ID"""
return (
db.query(models.PermissionRuleOct)
.filter(models.PermissionRuleOct.permission == permission)
.filter(
models.PermissionRuleOct.id == permission_id,
models.PermissionRuleOct.tenant_id == tenant_id,
models.PermissionRuleOct.company_id == company_id,
)
.first()
)
@staticmethod
def create_permission(db: Session, permission_data: dto.PermissionRuleOctCreateDTO):
new_permission = models.PermissionRuleOct(**permission_data.dict())
def create(
db: Session,
permission_data: dto.PermissionRuleOctCreateDTO,
tenant_id: int,
company_id: int,
) -> models.PermissionRuleOct:
"""Create a new permission rule OCT"""
new_permission = models.PermissionRuleOct(
**permission_data.model_dump(), tenant_id=tenant_id, company_id=company_id
)
db.add(new_permission)
db.commit()
db.refresh(new_permission)
return new_permission
@staticmethod
def delete_permission(db: Session, permission: str):
permission = PermissionRuleOctService.get_permission_by_id(db, permission)
if permission:
def update(
db: Session,
permission_id: int,
tenant_id: int,
company_id: int,
permission_data: dto.PermissionRuleOctUpdateDTO,
) -> Optional[models.PermissionRuleOct]:
"""Update a permission rule OCT"""
permission = PermissionRuleOctService.get_by_id(
db, permission_id, tenant_id, company_id
)
if not permission:
return None
# Update fields
update_data = permission_data.model_dump(exclude_unset=True)
for field, value in update_data.items():
setattr(permission, field, value)
db.commit()
db.refresh(permission)
return permission
@staticmethod
def delete(
db: Session, permission_id: int, tenant_id: int, company_id: int
) -> bool:
"""Delete a permission rule OCT"""
permission = PermissionRuleOctService.get_by_id(
db, permission_id, tenant_id, company_id
)
if not permission:
return False
db.delete(permission)
db.commit()
return permission
return True

17
backend/api/v1/modules/a76/seal/dto.py
View File

@@ -2,17 +2,30 @@
DTOs for Seal.
"""
from pydantic import BaseModel
from typing import Optional
from pydantic import BaseModel, Field
class SealBaseDTO(BaseModel):
seal: str
seal: str = Field(..., description="Seal identifier", max_length=15)
class SealCreateDTO(SealBaseDTO):
"""Schema for creating a seal"""
pass
class SealUpdateDTO(SealBaseDTO):
"""Schema for updating a seal"""
seal: Optional[str] = Field(None, description="Seal identifier", max_length=15)
class SealResponseDTO(SealBaseDTO):
"""Schema for seal response"""
id: int
company_id: int
tenant_id: int
class Config:
from_attributes = True

112
backend/api/v1/modules/a76/seal/routes.py
View File

@@ -2,101 +2,23 @@
Routes for managing Seal entries.
"""
from typing import List
from api.v1.common.tenant_crud_routes import TenantCRUDRoutes
from core.database import get_core_db
from core.security import get_current_user
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from .dto import SealCreateDTO, SealResponseDTO
from .dto import SealCreateDTO, SealResponseDTO, SealUpdateDTO
from .services import SealService
router = APIRouter(prefix="/seals", tags=["Seal"])
@router.get("/", response_model=List[SealResponseDTO])
async def list_seals(
db: Session = Depends(get_core_db), current_user: dict = Depends(get_current_user)
):
"""
List all Seal entries.
"""
# 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"
)
return db.query(SealService).all()
@router.get("/{seal}", response_model=SealResponseDTO)
async def read_seal(
seal: str,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
"""
Get a specific Seal by its seal.
"""
# 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"
)
seal = SealService.get_seal_by_id(db, seal)
if not seal:
raise HTTPException(status_code=404, detail="Seal not found")
return seal
@router.post("/", response_model=SealResponseDTO, status_code=status.HTTP_201_CREATED)
async def create_seal(
seal_data: SealCreateDTO,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
"""
Create a new Seal entry.
"""
# 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"
)
return SealService.create_seal(db, seal_data)
@router.delete("/{seal}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_seal(
seal: str,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
"""
Delete a Seal by its seal.
"""
# 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"
)
seal = SealService.delete_seal(db, seal)
if not seal:
raise HTTPException(status_code=404, detail="Seal not found")
# Create router using TenantCRUDRoutes factory
router = TenantCRUDRoutes(
service=SealService,
create_schema=SealCreateDTO,
update_schema=SealUpdateDTO,
response_schema=SealResponseDTO,
prefix="/seals",
tags=[],
resource_name="Seal",
id_name="id", # Using numeric ID
enable_list=True, # Enable GET /seals with pagination
enable_filters=True, # Enable filtering by seal
default_page_size=50,
max_page_size=100,
).router

103
backend/api/v1/modules/a76/seal/services.py
View File

@@ -1,29 +1,106 @@
from sqlalchemy.orm import Session
from . import dto, models
"""
Service layer for Seal.
"""
from typing import Optional, Tuple, List, Dict, Any
from sqlalchemy.orm import Session
from . import dto, models
class SealService:
@staticmethod
def get_seal_by_id(db: Session, seal: str):
return db.query(models.Seal).filter(models.Seal.seal == seal).first()
"""Service for Seal CRUD operations with tenant support"""
@staticmethod
def create_seal(db: Session, seal_data: dto.SealCreateDTO):
new_seal = models.Seal(**seal_data.dict())
def get_all(
db: Session,
tenant_id: int,
company_id: int,
skip: int = 0,
limit: int = 50,
filters: Optional[Dict[str, Any]] = None,
) -> Tuple[List[models.Seal], int]:
"""Get all seals for a tenant/company with pagination"""
query = db.query(models.Seal).filter(
models.Seal.tenant_id == tenant_id,
models.Seal.company_id == company_id,
)
# Apply filters if provided
if filters:
if filters.get("seal"):
query = query.filter(
models.Seal.seal.ilike(f"%{filters['seal']}%")
)
total = query.count()
seals = query.offset(skip).limit(limit).all()
return seals, total
@staticmethod
def get_by_id(
db: Session, seal_id: int, tenant_id: int, company_id: int
) -> Optional[models.Seal]:
"""Get seal by ID"""
return (
db.query(models.Seal)
.filter(
models.Seal.id == seal_id,
models.Seal.tenant_id == tenant_id,
models.Seal.company_id == company_id,
)
.first()
)
@staticmethod
def create(
db: Session,
seal_data: dto.SealCreateDTO,
tenant_id: int,
company_id: int,
) -> models.Seal:
"""Create a new seal"""
new_seal = models.Seal(
**seal_data.model_dump(), tenant_id=tenant_id, company_id=company_id
)
db.add(new_seal)
db.commit()
db.refresh(new_seal)
return new_seal
@staticmethod
def delete_seal(db: Session, seal: str):
seal = SealService.get_seal_by_id(db, seal)
if seal:
def update(
db: Session,
seal_id: int,
tenant_id: int,
company_id: int,
seal_data: dto.SealUpdateDTO,
) -> Optional[models.Seal]:
"""Update a seal"""
seal = SealService.get_by_id(db, seal_id, tenant_id, company_id)
if not seal:
return None
# Update fields
update_data = seal_data.model_dump(exclude_unset=True)
for field, value in update_data.items():
setattr(seal, field, value)
db.commit()
db.refresh(seal)
return seal
@staticmethod
def delete(
db: Session, seal_id: int, tenant_id: int, company_id: int
) -> bool:
"""Delete a seal"""
seal = SealService.get_by_id(db, seal_id, tenant_id, company_id)
if not seal:
return False
db.delete(seal)
db.commit()
return seal
return True

32
backend/api/v1/modules/a76/trailers/dto.py
View File

@@ -1,26 +1,34 @@
from typing import Optional
from pydantic import BaseModel
from pydantic import BaseModel, Field
class TrailerBaseDTO(BaseModel):
trailer_number: str
ace_trailer_number: Optional[str]
trailer_type_key: Optional[str]
seal: Optional[str]
entity_code: Optional[str]
plate_number: Optional[str]
state: Optional[str]
country: Optional[str]
container_key: Optional[str]
company_id: str
tenant_id: str
trailer_number: str = Field(..., description="Trailer number (primary identifier)")
ace_trailer_number: Optional[str] = None
trailer_type_key: Optional[str] = None
seal: Optional[str] = None
entity_code: Optional[str] = None
plate_number: Optional[str] = None
state: Optional[str] = None
country: Optional[str] = None
container_key: Optional[str] = None
class TrailerCreateDTO(TrailerBaseDTO):
"""Schema for creating a trailer"""
pass
class TrailerUpdateDTO(TrailerBaseDTO):
"""Schema for updating a trailer"""
trailer_number: Optional[str] = Field(None, description="Trailer number (cannot be modified)")
class TrailerResponseDTO(TrailerBaseDTO):
"""Schema for trailer response"""
company_id: int
tenant_id: int
class Config:
from_attributes = True

48
backend/api/v1/modules/a76/trailers/routes.py
View File

@@ -1,30 +1,22 @@
from core.database import get_core_db
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from api.v1.common.tenant_crud_routes import TenantCRUDRoutes
from . import dto, services
from .dto import TrailerCreateDTO, TrailerResponseDTO, TrailerUpdateDTO
from .services import TrailerService
router = APIRouter()
@router.get("/trailers/{trailer_number}", response_model=dto.TrailerResponseDTO)
def get_trailer(trailer_number: str, db: Session = Depends(get_core_db)):
trailer = services.TrailerService.get_trailer_by_number(db, trailer_number)
if not trailer:
raise HTTPException(status_code=404, detail="Trailer not found")
return trailer
@router.post("/trailers", response_model=dto.TrailerResponseDTO)
def create_trailer(
trailer_data: dto.TrailerCreateDTO, db: Session = Depends(get_core_db)
):
return services.TrailerService.create_trailer(db, trailer_data)
@router.delete("/trailers/{trailer_number}", response_model=dto.TrailerResponseDTO)
def delete_trailer(trailer_number: str, db: Session = Depends(get_core_db)):
trailer = services.TrailerService.delete_trailer(db, trailer_number)
if not trailer:
raise HTTPException(status_code=404, detail="Trailer not found")
return trailer
# Create router using TenantCRUDRoutes factory
# Note: trailer_number is a string (not int) and is used as the primary key
router = TenantCRUDRoutes(
service=TrailerService,
create_schema=TrailerCreateDTO,
update_schema=TrailerUpdateDTO,
response_schema=TrailerResponseDTO,
prefix="/trailers",
tags=[],
resource_name="Trailer",
id_name="trailer_number", # Using trailer_number instead of numeric ID
id_type=str, # Specify that the ID is a string
enable_list=True, # Enable GET /trailers with pagination
enable_filters=True, # Enable filtering by plate_number and trailer_type_key
default_page_size=50,
max_page_size=100,
).router

95
backend/api/v1/modules/a76/trailers/services.py
View File

@@ -1,29 +1,108 @@
from typing import Optional, Tuple, List, Dict, Any
from sqlalchemy.orm import Session
from . import dto, models
class TrailerService:
"""Service for Trailer CRUD operations with tenant support"""
@staticmethod
def get_trailer_by_number(db: Session, trailer_number: str):
def get_all(
db: Session,
tenant_id: int,
company_id: int,
skip: int = 0,
limit: int = 50,
filters: Optional[Dict[str, Any]] = None,
) -> Tuple[List[models.Trailer], int]:
"""Get all trailers for a tenant/company with pagination"""
query = db.query(models.Trailer).filter(
models.Trailer.tenant_id == tenant_id,
models.Trailer.company_id == company_id,
)
# Apply filters if provided
if filters:
if filters.get("plate_number"):
query = query.filter(
models.Trailer.plate_number.ilike(f"%{filters['plate_number']}%")
)
if filters.get("trailer_type_key"):
query = query.filter(
models.Trailer.trailer_type_key == filters["trailer_type_key"]
)
total = query.count()
trailers = query.offset(skip).limit(limit).all()
return trailers, total
@staticmethod
def get_by_id(
db: Session, trailer_number: str, tenant_id: int, company_id: int
) -> Optional[models.Trailer]:
"""Get trailer by trailer_number"""
return (
db.query(models.Trailer)
.filter(models.Trailer.trailer_number == trailer_number)
.filter(
models.Trailer.trailer_number == trailer_number,
models.Trailer.tenant_id == tenant_id,
models.Trailer.company_id == company_id,
)
.first()
)
@staticmethod
def create_trailer(db: Session, trailer_data: dto.TrailerCreateDTO):
new_trailer = models.Trailer(**trailer_data.dict())
def create(
db: Session,
trailer_data: dto.TrailerCreateDTO,
tenant_id: int,
company_id: int,
) -> models.Trailer:
"""Create a new trailer"""
new_trailer = models.Trailer(
**trailer_data.model_dump(), tenant_id=tenant_id, company_id=company_id
)
db.add(new_trailer)
db.commit()
db.refresh(new_trailer)
return new_trailer
@staticmethod
def delete_trailer(db: Session, trailer_number: str):
trailer = TrailerService.get_trailer_by_number(db, trailer_number)
if trailer:
def update(
db: Session,
trailer_number: str,
tenant_id: int,
company_id: int,
trailer_data: dto.TrailerUpdateDTO,
) -> Optional[models.Trailer]:
"""Update a trailer"""
trailer = TrailerService.get_by_id(db, trailer_number, tenant_id, company_id)
if not trailer:
return None
# Update fields (excluding trailer_number as it's the primary key)
update_data = trailer_data.model_dump(
exclude_unset=True, exclude={"trailer_number"}
)
for field, value in update_data.items():
setattr(trailer, field, value)
db.commit()
db.refresh(trailer)
return trailer
@staticmethod
def delete(
db: Session, trailer_number: str, tenant_id: int, company_id: int
) -> bool:
"""Delete a trailer"""
trailer = TrailerService.get_by_id(db, trailer_number, tenant_id, company_id)
if not trailer:
return False
db.delete(trailer)
db.commit()
return trailer
return True

52
backend/api/v1/modules/a76/transporters/dto.py
View File

@@ -1,36 +1,44 @@
from typing import Optional
from pydantic import BaseModel
from pydantic import BaseModel, Field
class TransporterBaseDTO(BaseModel):
transporter_key: str
name: Optional[str]
short_name: Optional[str]
responsible: Optional[str]
rfc: Optional[str]
streets: Optional[str]
postal_code: Optional[str]
city: Optional[str]
state: Optional[str]
country: Optional[str]
loader_code: Optional[str]
caat_code: Optional[str]
transport_code: Optional[str]
transport_interface_type: Optional[str]
ftp_server: Optional[str]
ftp_user: Optional[str]
ftp_password: Optional[str]
ftp_directory: Optional[str]
filler_code: Optional[str]
company_id: str
tenant_id: str
transporter_key: str = Field(..., description="Transporter key (primary identifier)")
name: Optional[str] = None
short_name: Optional[str] = None
responsible: Optional[str] = None
rfc: Optional[str] = None
streets: Optional[str] = None
postal_code: Optional[str] = None
city: Optional[str] = None
state: Optional[str] = None
country: Optional[str] = None
loader_code: Optional[str] = None
caat_code: Optional[str] = None
transport_code: Optional[str] = None
transport_interface_type: Optional[str] = None
ftp_server: Optional[str] = None
ftp_user: Optional[str] = None
ftp_password: Optional[str] = None
ftp_directory: Optional[str] = None
filler_code: Optional[str] = None
class TransporterCreateDTO(TransporterBaseDTO):
"""Schema for creating a transporter"""
pass
class TransporterUpdateDTO(TransporterBaseDTO):
"""Schema for updating a transporter"""
transporter_key: Optional[str] = Field(None, description="Transporter key (cannot be modified)")
class TransporterResponseDTO(TransporterBaseDTO):
"""Schema for transporter response"""
company_id: int
tenant_id: int
class Config:
from_attributes = True

68
backend/api/v1/modules/a76/transporters/routes.py
View File

@@ -1,52 +1,22 @@
from typing import List
from api.v1.common.tenant_crud_routes import TenantCRUDRoutes
from core.database import get_core_db
from core.security import get_current_user
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from .dto import TransporterCreateDTO, TransporterResponseDTO
from .dto import TransporterCreateDTO, TransporterResponseDTO, TransporterUpdateDTO
from .services import TransporterService
router = APIRouter(prefix="/transporters", tags=["Transporters"])
@router.get("/", response_model=List[TransporterResponseDTO])
async def list_transporters(
db: Session = Depends(get_core_db), current_user: dict = Depends(get_current_user)
):
return db.query(TransporterService).all()
@router.get("/{transporter_key}", response_model=TransporterResponseDTO)
async def read_transporter(
transporter_key: str,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
transporter = TransporterService.get_transporter_by_key(db, transporter_key)
if not transporter:
raise HTTPException(status_code=404, detail="Transporter not found")
return transporter
@router.post(
"/", response_model=TransporterResponseDTO, status_code=status.HTTP_201_CREATED
)
async def create_transporter(
transporter_data: TransporterCreateDTO,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
return TransporterService.create_transporter(db, transporter_data)
@router.delete("/{transporter_key}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_transporter(
transporter_key: str,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
transporter = TransporterService.delete_transporter(db, transporter_key)
if not transporter:
raise HTTPException(status_code=404, detail="Transporter not found")
# Create router using TenantCRUDRoutes factory
# Note: transporter_key is a string (not int) and is used as the primary key
router = TenantCRUDRoutes(
service=TransporterService,
create_schema=TransporterCreateDTO,
update_schema=TransporterUpdateDTO,
response_schema=TransporterResponseDTO,
prefix="/transporters",
tags=[],
resource_name="Transporter",
id_name="transporter_key", # Using transporter_key instead of numeric ID
id_type=str, # Specify that the ID is a string
enable_list=True, # Enable GET /transporters with pagination
enable_filters=True, # Enable filtering by name and rfc
default_page_size=50,
max_page_size=100,
).router

99
backend/api/v1/modules/a76/transporters/services.py
View File

@@ -1,29 +1,112 @@
from typing import Optional, Tuple, List, Dict, Any
from sqlalchemy.orm import Session
from . import dto, models
class TransporterService:
"""Service for Transporter CRUD operations with tenant support"""
@staticmethod
def get_transporter_by_key(db: Session, transporter_key: str):
def get_all(
db: Session,
tenant_id: int,
company_id: int,
skip: int = 0,
limit: int = 50,
filters: Optional[Dict[str, Any]] = None,
) -> Tuple[List[models.Transporter], int]:
"""Get all transporters for a tenant/company with pagination"""
query = db.query(models.Transporter).filter(
models.Transporter.tenant_id == tenant_id,
models.Transporter.company_id == company_id,
)
# Apply filters if provided
if filters:
if filters.get("name"):
query = query.filter(
models.Transporter.name.ilike(f"%{filters['name']}%")
)
if filters.get("rfc"):
query = query.filter(
models.Transporter.rfc.ilike(f"%{filters['rfc']}%")
)
total = query.count()
transporters = query.offset(skip).limit(limit).all()
return transporters, total
@staticmethod
def get_by_id(
db: Session, transporter_key: str, tenant_id: int, company_id: int
) -> Optional[models.Transporter]:
"""Get transporter by transporter_key"""
return (
db.query(models.Transporter)
.filter(models.Transporter.transporter_key == transporter_key)
.filter(
models.Transporter.transporter_key == transporter_key,
models.Transporter.tenant_id == tenant_id,
models.Transporter.company_id == company_id,
)
.first()
)
@staticmethod
def create_transporter(db: Session, transporter_data: dto.TransporterCreateDTO):
new_transporter = models.Transporter(**transporter_data.dict())
def create(
db: Session,
transporter_data: dto.TransporterCreateDTO,
tenant_id: int,
company_id: int,
) -> models.Transporter:
"""Create a new transporter"""
new_transporter = models.Transporter(
**transporter_data.model_dump(), tenant_id=tenant_id, company_id=company_id
)
db.add(new_transporter)
db.commit()
db.refresh(new_transporter)
return new_transporter
@staticmethod
def delete_transporter(db: Session, transporter_key: str):
transporter = TransporterService.get_transporter_by_key(db, transporter_key)
if transporter:
def update(
db: Session,
transporter_key: str,
tenant_id: int,
company_id: int,
transporter_data: dto.TransporterUpdateDTO,
) -> Optional[models.Transporter]:
"""Update a transporter"""
transporter = TransporterService.get_by_id(
db, transporter_key, tenant_id, company_id
)
if not transporter:
return None
# Update fields (excluding transporter_key as it's the primary key)
update_data = transporter_data.model_dump(
exclude_unset=True, exclude={"transporter_key"}
)
for field, value in update_data.items():
setattr(transporter, field, value)
db.commit()
db.refresh(transporter)
return transporter
@staticmethod
def delete(
db: Session, transporter_key: str, tenant_id: int, company_id: int
) -> bool:
"""Delete a transporter"""
transporter = TransporterService.get_by_id(
db, transporter_key, tenant_id, company_id
)
if not transporter:
return False
db.delete(transporter)
db.commit()
return transporter
return True

2
backend/api/v1/modules/a76/user_tenant/routes.py
View File

@@ -19,7 +19,7 @@ from .dto import (
)
from .service import UserTenantService
router = APIRouter(prefix="/user-tenants", tags=["User-Tenant Relations"])
router = APIRouter(prefix="/user-tenants")
@router.post("/add", response_model=UserTenantResponseDTO)

66
backend/api/v1/modules/a76/vehicles/dto.py
View File

@@ -1,43 +1,51 @@
from typing import Optional
from pydantic import BaseModel
from pydantic import BaseModel, Field
class VehicleBaseDTO(BaseModel):
vehicle_key: str
ace_vehicle_key: Optional[str]
transporter_key: Optional[str]
transport_identifier: Optional[str]
transport_type: Optional[str]
entity_code: Optional[str]
transponder_number: Optional[str]
dot_number: Optional[str]
plate_number: Optional[str]
city: Optional[str]
state: Optional[str]
country: Optional[str]
seal: Optional[str]
insurance_company_name: Optional[str]
insurance_number: Optional[str]
insurance_amount: Optional[float]
insurance_date: Optional[int]
box_number: Optional[str]
brand: Optional[str]
year: Optional[str]
series: Optional[str]
description: Optional[str]
engine_number: Optional[str]
sct_permission: Optional[str]
color: Optional[str]
container_key: Optional[str]
company_id: str
tenant_id: str
vehicle_key: str = Field(..., description="Vehicle key (primary identifier)")
ace_vehicle_key: Optional[str] = None
transporter_key: Optional[str] = None
transport_identifier: Optional[str] = None
transport_type: Optional[str] = None
entity_code: Optional[str] = None
transponder_number: Optional[str] = None
dot_number: Optional[str] = None
plate_number: Optional[str] = None
city: Optional[str] = None
state: Optional[str] = None
country: Optional[str] = None
seal: Optional[str] = None
insurance_company_name: Optional[str] = None
insurance_number: Optional[str] = None
insurance_amount: Optional[float] = None
insurance_date: Optional[int] = None
box_number: Optional[str] = None
brand: Optional[str] = None
year: Optional[str] = None
series: Optional[str] = None
description: Optional[str] = None
engine_number: Optional[str] = None
sct_permission: Optional[str] = None
color: Optional[str] = None
container_key: Optional[str] = None
class VehicleCreateDTO(VehicleBaseDTO):
"""Schema for creating a vehicle"""
pass
class VehicleUpdateDTO(VehicleBaseDTO):
"""Schema for updating a vehicle"""
vehicle_key: Optional[str] = Field(None, description="Vehicle key (cannot be modified)")
class VehicleResponseDTO(VehicleBaseDTO):
"""Schema for vehicle response"""
company_id: int
tenant_id: int
class Config:
from_attributes = True

68
backend/api/v1/modules/a76/vehicles/routes.py
View File

@@ -1,52 +1,22 @@
from typing import List
from api.v1.common.tenant_crud_routes import TenantCRUDRoutes
from core.database import get_core_db
from core.security import get_current_user
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from .dto import VehicleCreateDTO, VehicleResponseDTO
from .dto import VehicleCreateDTO, VehicleResponseDTO, VehicleUpdateDTO
from .services import VehicleService
router = APIRouter(prefix="/vehicles", tags=["Vehicles"])
@router.get("/", response_model=List[VehicleResponseDTO])
async def list_vehicles(
db: Session = Depends(get_core_db), current_user: dict = Depends(get_current_user)
):
return db.query(VehicleService).all()
@router.get("/{vehicle_key}", response_model=VehicleResponseDTO)
async def read_vehicle(
vehicle_key: str,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
vehicle = VehicleService.get_vehicle_by_key(db, vehicle_key)
if not vehicle:
raise HTTPException(status_code=404, detail="Vehicle not found")
return vehicle
@router.post(
"/", response_model=VehicleResponseDTO, status_code=status.HTTP_201_CREATED
)
async def create_vehicle(
vehicle_data: VehicleCreateDTO,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
return VehicleService.create_vehicle(db, vehicle_data)
@router.delete("/{vehicle_key}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_vehicle(
vehicle_key: str,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
vehicle = VehicleService.delete_vehicle(db, vehicle_key)
if not vehicle:
raise HTTPException(status_code=404, detail="Vehicle not found")
# Create router using TenantCRUDRoutes factory
# Note: vehicle_key is a string (not int) and is used as the primary key
router = TenantCRUDRoutes(
service=VehicleService,
create_schema=VehicleCreateDTO,
update_schema=VehicleUpdateDTO,
response_schema=VehicleResponseDTO,
prefix="/vehicles",
tags=[],
resource_name="Vehicle",
id_name="vehicle_key", # Using vehicle_key instead of numeric ID
id_type=str, # Specify that the ID is a string
enable_list=True, # Enable GET /vehicles with pagination
enable_filters=True, # Enable filtering by plate_number and transport_type
default_page_size=50,
max_page_size=100,
).router

93
backend/api/v1/modules/a76/vehicles/services.py
View File

@@ -1,29 +1,106 @@
from typing import Optional, Tuple, List, Dict, Any
from sqlalchemy.orm import Session
from . import dto, models
class VehicleService:
"""Service for Vehicle CRUD operations with tenant support"""
@staticmethod
def get_vehicle_by_key(db: Session, vehicle_key: str):
def get_all(
db: Session,
tenant_id: int,
company_id: int,
skip: int = 0,
limit: int = 50,
filters: Optional[Dict[str, Any]] = None,
) -> Tuple[List[models.Vehicle], int]:
"""Get all vehicles for a tenant/company with pagination"""
query = db.query(models.Vehicle).filter(
models.Vehicle.tenant_id == tenant_id,
models.Vehicle.company_id == company_id,
)
# Apply filters if provided
if filters:
if filters.get("plate_number"):
query = query.filter(
models.Vehicle.plate_number.ilike(f"%{filters['plate_number']}%")
)
if filters.get("transport_type"):
query = query.filter(
models.Vehicle.transport_type == filters["transport_type"]
)
total = query.count()
vehicles = query.offset(skip).limit(limit).all()
return vehicles, total
@staticmethod
def get_by_id(
db: Session, vehicle_key: str, tenant_id: int, company_id: int
) -> Optional[models.Vehicle]:
"""Get vehicle by vehicle_key"""
return (
db.query(models.Vehicle)
.filter(models.Vehicle.vehicle_key == vehicle_key)
.filter(
models.Vehicle.vehicle_key == vehicle_key,
models.Vehicle.tenant_id == tenant_id,
models.Vehicle.company_id == company_id,
)
.first()
)
@staticmethod
def create_vehicle(db: Session, vehicle_data: dto.VehicleCreateDTO):
new_vehicle = models.Vehicle(**vehicle_data.dict())
def create(
db: Session,
vehicle_data: dto.VehicleCreateDTO,
tenant_id: int,
company_id: int,
) -> models.Vehicle:
"""Create a new vehicle"""
new_vehicle = models.Vehicle(
**vehicle_data.model_dump(), tenant_id=tenant_id, company_id=company_id
)
db.add(new_vehicle)
db.commit()
db.refresh(new_vehicle)
return new_vehicle
@staticmethod
def delete_vehicle(db: Session, vehicle_key: str):
vehicle = VehicleService.get_vehicle_by_key(db, vehicle_key)
if vehicle:
def update(
db: Session,
vehicle_key: str,
tenant_id: int,
company_id: int,
vehicle_data: dto.VehicleUpdateDTO,
) -> Optional[models.Vehicle]:
"""Update a vehicle"""
vehicle = VehicleService.get_by_id(db, vehicle_key, tenant_id, company_id)
if not vehicle:
return None
# Update fields (excluding vehicle_key as it's the primary key)
update_data = vehicle_data.model_dump(exclude_unset=True, exclude={"vehicle_key"})
for field, value in update_data.items():
setattr(vehicle, field, value)
db.commit()
db.refresh(vehicle)
return vehicle
@staticmethod
def delete(
db: Session, vehicle_key: str, tenant_id: int, company_id: int
) -> bool:
"""Delete a vehicle"""
vehicle = VehicleService.get_by_id(db, vehicle_key, tenant_id, company_id)
if not vehicle:
return False
db.delete(vehicle)
db.commit()
return vehicle
return True