529 lines
21 KiB
Python
529 lines
21 KiB
Python
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
|
|
from fastapi import APIRouter, Body, Depends, HTTPException, Path, Query
|
|
from pydantic import BaseModel
|
|
from sqlalchemy.orm import Session
|
|
|
|
# Type variables for generic types
|
|
ModelType = TypeVar("ModelType")
|
|
CreateSchemaType = TypeVar("CreateSchemaType", bound=BaseModel)
|
|
UpdateSchemaType = TypeVar("UpdateSchemaType", bound=BaseModel)
|
|
ResponseSchemaType = TypeVar("ResponseSchemaType", bound=BaseModel)
|
|
ServiceType = TypeVar("ServiceType")
|
|
|
|
|
|
class TenantCRUDRoutes(
|
|
Generic[CreateSchemaType, UpdateSchemaType, ResponseSchemaType, ServiceType]
|
|
):
|
|
"""
|
|
Generic CRUD routes factory for tenant-scoped resources
|
|
|
|
Supports both parent resources (with list/pagination) and child resources (nested under parent).
|
|
|
|
Usage examples:
|
|
|
|
1. Parent resource with list (e.g., /pedimentos):
|
|
router = TenantCRUDRoutes(
|
|
service=PedimentosService,
|
|
create_schema=PedimentosCreate,
|
|
update_schema=PedimentosUpdate,
|
|
response_schema=PedimentosResponse,
|
|
prefix="/pedimentos",
|
|
tags=["Pedimentos"],
|
|
resource_name="Pedimento",
|
|
id_name="pedimento_id",
|
|
enable_list=True,
|
|
).router
|
|
|
|
2. Child resource (e.g., /pedimentos/{pedimento_id}/config-additional):
|
|
router = TenantCRUDRoutes(
|
|
service=PedimentoConfigAdditionalService,
|
|
create_schema=PedimentoConfigAdditionalCreate,
|
|
update_schema=PedimentoConfigAdditionalUpdate,
|
|
response_schema=PedimentoConfigAdditionalResponse,
|
|
prefix="/{pedimento_id}/config-additional",
|
|
tags=["Pedimento Config Additional"],
|
|
resource_name="Config additional",
|
|
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__(
|
|
self,
|
|
service: Type[ServiceType],
|
|
create_schema: Type[CreateSchemaType],
|
|
update_schema: Type[UpdateSchemaType],
|
|
response_schema: Type[ResponseSchemaType],
|
|
prefix: str,
|
|
tags: list[str],
|
|
resource_name: str = "Resource",
|
|
# For parent resources (e.g., "pedimento_id")
|
|
id_name: Optional[str] = None,
|
|
id_type: Type = int, # Type of the ID (int, str, etc.)
|
|
parent_id_name: Optional[
|
|
str
|
|
] = None, # For child resources (e.g., "pedimento_id")
|
|
db_dependency: Callable = get_core_db,
|
|
auth_dependency: Callable = get_current_user,
|
|
validate_parent_match: bool = True, # Validate parent_id matches in create
|
|
enable_list: bool = False, # Enable GET list endpoint with pagination
|
|
enable_filters: bool = False, # Enable custom filters in list endpoint
|
|
default_page_size: int = 50,
|
|
max_page_size: int = 100,
|
|
# Permissions for each operation
|
|
list_permissions: Optional[list[str]] = None,
|
|
get_permissions: Optional[list[str]] = None,
|
|
create_permissions: Optional[list[str]] = None,
|
|
update_permissions: Optional[list[str]] = None,
|
|
delete_permissions: Optional[list[str]] = None,
|
|
require_all: bool = True, # If True, requires ALL permissions; if False, requires ANY
|
|
):
|
|
self.service = service
|
|
self.create_schema = create_schema
|
|
self.update_schema = update_schema
|
|
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
|
|
self.validate_parent_match = validate_parent_match
|
|
self.enable_list = enable_list
|
|
self.enable_filters = enable_filters
|
|
self.default_page_size = default_page_size
|
|
self.max_page_size = max_page_size
|
|
self.list_permissions = list_permissions
|
|
self.get_permissions = get_permissions
|
|
self.create_permissions = create_permissions
|
|
self.update_permissions = update_permissions
|
|
self.delete_permissions = delete_permissions
|
|
self.require_all = require_all
|
|
|
|
self.router = APIRouter(prefix=prefix, tags=tags)
|
|
self._register_routes()
|
|
|
|
def _register_routes(self):
|
|
"""Register all CRUD routes"""
|
|
|
|
# LIST route (optional, for parent resources)
|
|
if self.enable_list:
|
|
if self.enable_filters:
|
|
|
|
@self.router.get(
|
|
"/",
|
|
response_model=Dict[str, Any],
|
|
summary=f"List {self.resource_name}s",
|
|
description=f"Get paginated list of {self.resource_name}s with optional filters",
|
|
)
|
|
async def list_resources(
|
|
company_id: int = Query(..., description="Company ID"),
|
|
page: int = Query(1, ge=1, description="Page number"),
|
|
page_size: int = Query(
|
|
self.default_page_size,
|
|
ge=1,
|
|
le=self.max_page_size,
|
|
description="Page size",
|
|
),
|
|
status: Optional[str] = Query(None, description="Filter by status"),
|
|
operation_type: Optional[str] = Query(
|
|
None, description="Filter by operation type"
|
|
),
|
|
invoice_type: Optional[str] = Query(
|
|
None, description="Filter by invoice type"
|
|
),
|
|
db: Session = Depends(self.db_dependency),
|
|
current_user: Dict[str, Any] = Depends(self.auth_dependency),
|
|
):
|
|
tenant_id = validate_access_to_resource(
|
|
db,
|
|
company_id,
|
|
current_user,
|
|
self.list_permissions,
|
|
self.require_all,
|
|
)
|
|
|
|
skip = (page - 1) * page_size
|
|
filters = {}
|
|
if status:
|
|
filters["status"] = status
|
|
if operation_type:
|
|
filters["operation_type"] = operation_type
|
|
if invoice_type:
|
|
filters["invoice_type"] = invoice_type
|
|
|
|
items, total = self.service.get_all(
|
|
db, tenant_id, company_id, skip, page_size, filters
|
|
)
|
|
|
|
return {
|
|
"items": [
|
|
self.response_schema.model_validate(item) for item in items
|
|
],
|
|
"total": total,
|
|
"page": page,
|
|
"page_size": page_size,
|
|
}
|
|
|
|
else:
|
|
|
|
@self.router.get(
|
|
"/",
|
|
response_model=Dict[str, Any],
|
|
summary=f"List {self.resource_name}s",
|
|
description=f"Get paginated list of {self.resource_name}s",
|
|
)
|
|
async def list_resources(
|
|
company_id: int = Query(..., description="Company ID"),
|
|
page: int = Query(1, ge=1, description="Page number"),
|
|
page_size: int = Query(
|
|
self.default_page_size,
|
|
ge=1,
|
|
le=self.max_page_size,
|
|
description="Page size",
|
|
),
|
|
db: Session = Depends(self.db_dependency),
|
|
current_user: Dict[str, Any] = Depends(self.auth_dependency),
|
|
):
|
|
tenant_id = validate_access_to_resource(
|
|
db,
|
|
company_id,
|
|
current_user,
|
|
self.list_permissions,
|
|
self.require_all,
|
|
)
|
|
|
|
skip = (page - 1) * page_size
|
|
|
|
items, total = self.service.get_all(
|
|
db, tenant_id, company_id, skip, page_size, None
|
|
)
|
|
|
|
return {
|
|
"items": [
|
|
self.response_schema.model_validate(item) for item in items
|
|
],
|
|
"total": total,
|
|
"page": page,
|
|
"page_size": page_size,
|
|
}
|
|
|
|
# GET single resource route
|
|
# For parent resources: GET /{id}
|
|
# For child resources: GET / (parent_id comes from path)
|
|
if self.parent_id_name:
|
|
# Child resource - single GET without ID in path
|
|
@self.router.get(
|
|
"/",
|
|
response_model=self.response_schema,
|
|
summary=f"Get {self.resource_name}",
|
|
description=f"Get {self.resource_name} by {self.parent_id_name}",
|
|
)
|
|
async def get_resource(
|
|
company_id: int = Query(..., description="Company ID"),
|
|
db: Session = Depends(self.db_dependency),
|
|
current_user: Dict[str, Any] = Depends(self.auth_dependency),
|
|
**path_params,
|
|
):
|
|
|
|
tenant_id = validate_access_to_resource(
|
|
db, company_id, current_user, self.get_permissions, self.require_all
|
|
)
|
|
parent_id = path_params.get(self.parent_id_name)
|
|
|
|
# Try method with 4 params (pedimento_id, tenant_id, company_id)
|
|
if hasattr(self.service, "get_by_pedimento_id"):
|
|
resource = self.service.get_by_pedimento_id(
|
|
db, parent_id, tenant_id, company_id
|
|
)
|
|
# Fallback to method with 3 params
|
|
elif hasattr(self.service, "get_by_id"):
|
|
resource = self.service.get_by_id(
|
|
db, parent_id, tenant_id, company_id
|
|
)
|
|
else:
|
|
resource = self.service.get(db, parent_id, tenant_id, company_id)
|
|
|
|
if not resource:
|
|
raise HTTPException(
|
|
status_code=404, detail=f"{self.resource_name} not found"
|
|
)
|
|
return resource
|
|
|
|
else:
|
|
# Parent resource - GET by ID in path
|
|
@self.router.get(
|
|
f"/{{{self.id_name}}}",
|
|
response_model=self.response_schema,
|
|
summary=f"Get {self.resource_name} by ID",
|
|
description=f"Get a specific {self.resource_name} by {self.id_name}",
|
|
)
|
|
async def get_resource_by_id(
|
|
resource_id: Union[int, str] = Path(
|
|
..., alias=self.id_name, description=f"{self.resource_name} ID"
|
|
),
|
|
company_id: int = Query(..., description="Company ID"),
|
|
db: Session = Depends(self.db_dependency),
|
|
current_user: Dict[str, Any] = Depends(self.auth_dependency),
|
|
):
|
|
tenant_id = validate_access_to_resource(
|
|
db, company_id, current_user, self.get_permissions, self.require_all
|
|
)
|
|
|
|
resource = self.service.get_by_id(
|
|
db, resource_id, tenant_id, company_id
|
|
)
|
|
|
|
if not resource:
|
|
raise HTTPException(
|
|
status_code=404, detail=f"{self.resource_name} not found"
|
|
)
|
|
return resource
|
|
|
|
# POST route
|
|
if self.parent_id_name:
|
|
# Child resource - needs parent_id from path
|
|
|
|
# Create a closure to capture the schema type
|
|
create_schema = self.create_schema
|
|
|
|
@self.router.post(
|
|
"/",
|
|
response_model=self.response_schema,
|
|
status_code=201,
|
|
summary=f"Create {self.resource_name}",
|
|
description=f"Create a new {self.resource_name}",
|
|
)
|
|
async def create_child_resource(
|
|
company_id: int = Query(..., description="Company ID"),
|
|
data: create_schema = Body(...), # type: ignore
|
|
db: Session = Depends(self.db_dependency),
|
|
current_user: Dict[str, Any] = Depends(self.auth_dependency),
|
|
):
|
|
tenant_id = validate_access_to_resource(
|
|
db,
|
|
company_id,
|
|
current_user,
|
|
self.create_permissions,
|
|
self.require_all,
|
|
)
|
|
|
|
# For child resources, parent_id validation would go here
|
|
try:
|
|
resource = self.service.create(db, data, tenant_id, company_id)
|
|
return resource
|
|
except ValueError as e:
|
|
# Capturar errores de validación (como duplicados)
|
|
raise HTTPException(status_code=400, detail=str(e))
|
|
except Exception as e:
|
|
# Re-lanzar otros errores
|
|
raise
|
|
|
|
else:
|
|
# Parent resource - no parent_id needed
|
|
|
|
# Create a closure to capture the schema type
|
|
create_schema = self.create_schema
|
|
|
|
@self.router.post(
|
|
"/",
|
|
response_model=self.response_schema,
|
|
status_code=201,
|
|
summary=f"Create {self.resource_name}",
|
|
description=f"Create a new {self.resource_name}",
|
|
)
|
|
async def create_parent_resource(
|
|
company_id: int = Query(..., description="Company ID"),
|
|
data: create_schema = Body(...), # type: ignore
|
|
db: Session = Depends(self.db_dependency),
|
|
current_user: Dict[str, Any] = Depends(self.auth_dependency),
|
|
):
|
|
tenant_id = validate_access_to_resource(
|
|
db,
|
|
company_id,
|
|
current_user,
|
|
self.create_permissions,
|
|
self.require_all,
|
|
)
|
|
try:
|
|
resource = self.service.create(db, data, tenant_id, company_id)
|
|
return resource
|
|
except ValueError as e:
|
|
# Capturar errores de validación (como duplicados)
|
|
raise HTTPException(status_code=400, detail=str(e))
|
|
except Exception as e:
|
|
# Re-lanzar otros errores
|
|
raise
|
|
|
|
# PUT route
|
|
# For parent resources: PUT /{id}
|
|
# For child resources: PUT / (parent_id comes from path)
|
|
if self.parent_id_name:
|
|
# Child resource
|
|
|
|
# Create a closure to capture the schema type
|
|
update_schema = self.update_schema
|
|
|
|
@self.router.put(
|
|
"/",
|
|
response_model=self.response_schema,
|
|
summary=f"Update {self.resource_name}",
|
|
description=f"Update an existing {self.resource_name}",
|
|
)
|
|
async def update_resource(
|
|
company_id: int = Query(..., description="Company ID"),
|
|
data: update_schema = Body(...), # type: ignore
|
|
db: Session = Depends(self.db_dependency),
|
|
current_user: Dict[str, Any] = Depends(self.auth_dependency),
|
|
**path_params,
|
|
):
|
|
tenant_id = validate_access_to_resource(
|
|
db,
|
|
company_id,
|
|
current_user,
|
|
self.update_permissions,
|
|
self.require_all,
|
|
)
|
|
parent_id = path_params.get(self.parent_id_name)
|
|
|
|
try:
|
|
resource = self.service.update(
|
|
db, parent_id, tenant_id, data, company_id
|
|
)
|
|
except ValueError as e:
|
|
# Capturar errores de validación (como duplicados)
|
|
raise HTTPException(status_code=400, detail=str(e))
|
|
|
|
if not resource:
|
|
raise HTTPException(
|
|
status_code=404, detail=f"{self.resource_name} not found"
|
|
)
|
|
return resource
|
|
|
|
else:
|
|
# Parent resource
|
|
|
|
# Create a closure to capture the schema type
|
|
update_schema = self.update_schema
|
|
|
|
@self.router.put(
|
|
f"/{{{self.id_name}}}/",
|
|
response_model=self.response_schema,
|
|
summary=f"Update {self.resource_name}",
|
|
description=f"Update an existing {self.resource_name} by {self.id_name}",
|
|
)
|
|
async def update_resource_by_id(
|
|
resource_id: Union[int, str] = Path(
|
|
..., alias=self.id_name, description=f"{self.resource_name} ID"
|
|
),
|
|
company_id: int = Query(..., description="Company ID"),
|
|
data: update_schema = Body(...), # type: ignore
|
|
db: Session = Depends(self.db_dependency),
|
|
current_user: Dict[str, Any] = Depends(self.auth_dependency),
|
|
):
|
|
f"""Update {self.resource_name}"""
|
|
tenant_id = validate_access_to_resource(
|
|
db,
|
|
company_id,
|
|
current_user,
|
|
self.update_permissions,
|
|
self.require_all,
|
|
)
|
|
|
|
try:
|
|
resource = self.service.update(
|
|
db, resource_id, tenant_id, data, company_id
|
|
)
|
|
except ValueError as e:
|
|
# Capturar errores de validación (como duplicados)
|
|
raise HTTPException(status_code=400, detail=str(e))
|
|
|
|
if not resource:
|
|
raise HTTPException(
|
|
status_code=404, detail=f"{self.resource_name} not found"
|
|
)
|
|
return resource
|
|
|
|
# DELETE route
|
|
# For parent resources: DELETE /{id}
|
|
# For child resources: DELETE / (parent_id comes from path)
|
|
if self.parent_id_name:
|
|
# Child resource
|
|
@self.router.delete(
|
|
"/",
|
|
status_code=204,
|
|
summary=f"Delete {self.resource_name}",
|
|
description=f"Delete an existing {self.resource_name}",
|
|
)
|
|
async def delete_resource(
|
|
company_id: int = Query(..., description="Company ID"),
|
|
db: Session = Depends(self.db_dependency),
|
|
current_user: Dict[str, Any] = Depends(self.auth_dependency),
|
|
**path_params,
|
|
):
|
|
tenant_id = validate_access_to_resource(
|
|
db,
|
|
company_id,
|
|
current_user,
|
|
self.delete_permissions,
|
|
self.require_all,
|
|
)
|
|
parent_id = path_params.get(self.parent_id_name)
|
|
|
|
success = self.service.delete(db, parent_id, tenant_id, company_id)
|
|
|
|
if not success:
|
|
raise HTTPException(
|
|
status_code=404, detail=f"{self.resource_name} not found"
|
|
)
|
|
return None
|
|
|
|
else:
|
|
# Parent resource
|
|
@self.router.delete(
|
|
f"/{{{self.id_name}}}",
|
|
status_code=204,
|
|
summary=f"Delete {self.resource_name}",
|
|
description=f"Delete an existing {self.resource_name} by {self.id_name}",
|
|
)
|
|
async def delete_resource_by_id(
|
|
resource_id: Union[int, str] = Path(
|
|
..., alias=self.id_name, description=f"{self.resource_name} ID"
|
|
),
|
|
company_id: int = Query(..., description="Company ID"),
|
|
db: Session = Depends(self.db_dependency),
|
|
current_user: Dict[str, Any] = Depends(self.auth_dependency),
|
|
):
|
|
tenant_id = validate_access_to_resource(
|
|
db,
|
|
company_id,
|
|
current_user,
|
|
self.delete_permissions,
|
|
self.require_all,
|
|
)
|
|
|
|
success = self.service.delete(db, resource_id, tenant_id, company_id)
|
|
|
|
if not success:
|
|
raise HTTPException(
|
|
status_code=404, detail=f"{self.resource_name} not found"
|
|
)
|
|
return None
|