chore: baseline plantilla-proyectos como base del CRM
This commit is contained in:
Aduanasoft
651
backend/api/v1/common/tenant_crud_routes.py
Normal file
651
backend/api/v1/common/tenant_crud_routes.py
Normal file
@@ -0,0 +1,651 @@
|
||||
from typing import Any, Callable, Dict, Generic, Optional, Type, TypeVar, Union
|
||||
import logging
|
||||
import inspect
|
||||
|
||||
from core.database import get_core_db
|
||||
from core.security import get_current_user, is_hub_admin, resolve_tenant_id_required, validate_access_to_resource, get_active_system
|
||||
from fastapi import APIRouter, Body, Depends, HTTPException, Path, Query, Request
|
||||
|
||||
from api.v1.common.catalog_validation_errors import CatalogValidationError
|
||||
from pydantic import BaseModel
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# 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 = 2000,
|
||||
# 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(
|
||||
request: Request,
|
||||
company_id: int = Query(..., description="Company ID"),
|
||||
all_companies: bool = Query(False, description="Whether to search in all companies of the tenant"),
|
||||
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",
|
||||
),
|
||||
sort_by: Optional[str] = Query(None, description="Column to sort by"),
|
||||
sort_order: Optional[str] = Query("asc", pattern="^(asc|desc)$", description="Sort order (asc or desc)"),
|
||||
db: Session = Depends(self.db_dependency),
|
||||
current_user: Dict[str, Any] = Depends(self.auth_dependency),
|
||||
):
|
||||
if all_companies:
|
||||
# Hub admin: tenant_id=None → el servicio devuelve todas las empresas
|
||||
tenant_id = resolve_tenant_id_required(current_user, db=db)
|
||||
target_company_id = None
|
||||
else:
|
||||
tenant_id = validate_access_to_resource(
|
||||
db,
|
||||
company_id,
|
||||
current_user,
|
||||
self.list_permissions,
|
||||
self.require_all,
|
||||
)
|
||||
target_company_id = company_id
|
||||
|
||||
skip = (page - 1) * page_size
|
||||
|
||||
# Extraer todos los parámetros de búsqueda dinámicamente
|
||||
# Excluimos los parámetros estándar de paginación y control
|
||||
standard_params = {"company_id", "all_companies", "page", "page_size", "sort_by", "sort_order"}
|
||||
filters = {
|
||||
k: v
|
||||
for k, v in request.query_params.items()
|
||||
if k not in standard_params and v is not None and v != ""
|
||||
}
|
||||
|
||||
# Inyectar active_system (header/cookie) si no viene por query param
|
||||
active_system = get_active_system(request)
|
||||
if active_system and "system" not in filters:
|
||||
filters["system"] = active_system
|
||||
|
||||
# Determine what parameters the service method accepts
|
||||
sig = inspect.signature(self.service.get_all)
|
||||
kwargs = {}
|
||||
if "sort_by" in sig.parameters:
|
||||
kwargs["sort_by"] = sort_by
|
||||
if "sort_order" in sig.parameters:
|
||||
kwargs["sort_order"] = sort_order
|
||||
|
||||
try:
|
||||
items, total = self.service.get_all(
|
||||
db, tenant_id, target_company_id, skip, page_size, filters, **kwargs
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error(f"Error in {self.resource_name} list service: {str(e)}", exc_info=True)
|
||||
raise HTTPException(
|
||||
status_code=500,
|
||||
detail=f"Error listing {self.resource_name}s: {str(e)}"
|
||||
)
|
||||
|
||||
try:
|
||||
return {
|
||||
"items": [
|
||||
self.response_schema.model_validate(item) for item in items
|
||||
],
|
||||
"total": total,
|
||||
"page": page,
|
||||
"page_size": page_size,
|
||||
}
|
||||
except Exception as e:
|
||||
logger.error(f"Error validating {self.resource_name} response schema: {str(e)}", exc_info=True)
|
||||
raise HTTPException(
|
||||
status_code=500,
|
||||
detail=f"Data validation error in {self.resource_name}"
|
||||
)
|
||||
|
||||
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"),
|
||||
all_companies: bool = Query(False, description="Whether to search in all companies of the tenant"),
|
||||
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",
|
||||
),
|
||||
sort_by: Optional[str] = Query(None, description="Column to sort by"),
|
||||
sort_order: Optional[str] = Query("asc", pattern="^(asc|desc)$", description="Sort order (asc or desc)"),
|
||||
db: Session = Depends(self.db_dependency),
|
||||
current_user: Dict[str, Any] = Depends(self.auth_dependency),
|
||||
):
|
||||
if all_companies:
|
||||
# Hub admin: tenant_id=None → el servicio devuelve todas las empresas
|
||||
tenant_id = resolve_tenant_id_required(current_user, db=db)
|
||||
target_company_id = None
|
||||
else:
|
||||
tenant_id = validate_access_to_resource(
|
||||
db,
|
||||
company_id,
|
||||
current_user,
|
||||
self.list_permissions,
|
||||
self.require_all,
|
||||
)
|
||||
target_company_id = company_id
|
||||
|
||||
skip = (page - 1) * page_size
|
||||
|
||||
# Determine what parameters the service method accepts
|
||||
sig = inspect.signature(self.service.get_all)
|
||||
kwargs = {}
|
||||
if "sort_by" in sig.parameters:
|
||||
kwargs["sort_by"] = sort_by
|
||||
if "sort_order" in sig.parameters:
|
||||
kwargs["sort_order"] = sort_order
|
||||
|
||||
try:
|
||||
items, total = self.service.get_all(
|
||||
db, tenant_id, target_company_id, skip, page_size, None, **kwargs
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error(f"Error in {self.resource_name} list service: {str(e)}", exc_info=True)
|
||||
raise HTTPException(
|
||||
status_code=500,
|
||||
detail=f"Error listing {self.resource_name}s: {str(e)}"
|
||||
)
|
||||
|
||||
try:
|
||||
return {
|
||||
"items": [
|
||||
self.response_schema.model_validate(item) for item in items
|
||||
],
|
||||
"total": total,
|
||||
"page": page,
|
||||
"page_size": page_size,
|
||||
}
|
||||
except Exception as e:
|
||||
logger.error(f"Error validating {self.resource_name} response schema: {str(e)}", exc_info=True)
|
||||
raise HTTPException(
|
||||
status_code=500,
|
||||
detail=f"Data validation error in {self.resource_name}"
|
||||
)
|
||||
|
||||
# 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
|
||||
)
|
||||
|
||||
try:
|
||||
resource = self.service.get_by_id(
|
||||
db, resource_id, tenant_id, company_id
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error(f"Error in {self.resource_name} get service: {str(e)}", exc_info=True)
|
||||
raise HTTPException(
|
||||
status_code=500,
|
||||
detail=f"Error retrieving {self.resource_name}: {str(e)}"
|
||||
)
|
||||
|
||||
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(
|
||||
request: Request,
|
||||
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,
|
||||
)
|
||||
|
||||
# Inyectar sistema activo en el campo system si el recurso lo soporta
|
||||
if self.enable_filters:
|
||||
active_system = get_active_system(request)
|
||||
if active_system and hasattr(data, "system"):
|
||||
data = data.model_copy(update={"system": active_system})
|
||||
|
||||
# For child resources, parent_id validation would go here
|
||||
try:
|
||||
resource = self.service.create(db, data, tenant_id, company_id)
|
||||
return resource
|
||||
except CatalogValidationError as e:
|
||||
raise HTTPException(
|
||||
status_code=422,
|
||||
detail={
|
||||
"message": str(e),
|
||||
"errors": e.errors,
|
||||
},
|
||||
)
|
||||
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(
|
||||
request: Request,
|
||||
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,
|
||||
)
|
||||
|
||||
# Inyectar sistema activo en el campo system si el recurso lo soporta
|
||||
if self.enable_filters:
|
||||
active_system = get_active_system(request)
|
||||
if active_system and hasattr(data, "system"):
|
||||
data = data.model_copy(update={"system": active_system})
|
||||
|
||||
try:
|
||||
resource = self.service.create(db, data, tenant_id, company_id)
|
||||
return resource
|
||||
except CatalogValidationError as e:
|
||||
raise HTTPException(
|
||||
status_code=422,
|
||||
detail={
|
||||
"message": str(e),
|
||||
"errors": e.errors,
|
||||
},
|
||||
)
|
||||
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 CatalogValidationError as e:
|
||||
raise HTTPException(
|
||||
status_code=422,
|
||||
detail={
|
||||
"message": str(e),
|
||||
"errors": e.errors,
|
||||
},
|
||||
)
|
||||
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 CatalogValidationError as e:
|
||||
raise HTTPException(
|
||||
status_code=422,
|
||||
detail={
|
||||
"message": str(e),
|
||||
"errors": e.errors,
|
||||
},
|
||||
)
|
||||
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
|
||||
Reference in New Issue
Block a user