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

Merge pull request 'feature/hub-admin-global-access' (#410) from feature/hub-admin-global-access into development

Browse Source 
Reviewed-on: ADUANASOFT/anexo76#410
This commit is contained in:
acazares
2026-05-20 16:31:09 +00:00
parent 4c844bdb05 19bc7a2b57
commit 71c19d28b0
15 changed files with 251 additions and 174 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -12,6 +12,9 @@ from pydantic import BaseModel, ConfigDict, Field, field_validator
class CompanyCreateDTO(BaseModel):
"""DTO para crear una empresa"""
# Solo requerido cuando hub_admin crea una empresa (no tiene tenant_id en token)
tenant_id: Optional[int] = Field(None, description="Tenant ID (requerido para hub_admin)")
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(

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

@@ -10,7 +10,7 @@ from typing import List, Optional
from pathlib import Path
from datetime import datetime
from fastapi import APIRouter, Depends, HTTPException, status, File, UploadFile
from fastapi import APIRouter, Depends, HTTPException, Request, status, File, UploadFile
from fastapi.responses import FileResponse, Response
from sqlalchemy.orm import Session
@@ -23,7 +23,9 @@ from core.security import (
collect_user_role_names,
get_current_user,
get_tenant_from_token,
is_hub_admin,
resolve_effective_tenant_id_from_user,
resolve_tenant_id_required,
validate_access_to_resource,
)
from .....common.tenant_crud_routes import TenantCRUDRoutes
@@ -40,7 +42,7 @@ logger = logging.getLogger(__name__)
def _user_is_admin(current_user: dict) -> bool:
return "admin" in collect_user_role_names(current_user)
return "admin" in collect_user_role_names(current_user) or is_hub_admin(current_user)
def _assert_permission_any_company(
@@ -81,26 +83,13 @@ def _assert_permission_for_company(
return validate_access_to_resource(db, company_id, current_user, [permission_code])
def _resolve_tenant_id_int(current_user: dict) -> int:
"""Misma lógica que validate_access_to_resource: entero estable para BD y claves S3."""
tid = get_tenant_from_token(current_user)
if tid is not None:
return int(tid)
raw = current_user.get("tenant_id")
if isinstance(raw, list) and raw:
raw = raw[0]
if raw is None:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Tenant ID not found in user data",
)
try:
return int(raw)
except (TypeError, ValueError):
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Invalid tenant ID in token",
)
def _resolve_tenant_id_int(
current_user: dict,
db: Session = None,
company_id: int = None,
) -> Optional[int]:
"""Delega a resolve_tenant_id_required. Hub admin resuelve tenant desde la empresa."""
return resolve_tenant_id_required(current_user, db=db, company_id=company_id)
def _is_s3_object_key(ref: Optional[str]) -> bool:
@@ -127,18 +116,27 @@ router = APIRouter(prefix="/company")
summary="Create a new company",
)
async def create_company(
request: Request,
data: CompanyCreateDTO,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
_assert_permission_any_company(db, current_user, "cat_company.create")
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",
)
tenant_id = resolve_tenant_id_required(current_user, db=db)
if tenant_id is None:
# Prioridad: X-Tenant-Override del header (tenant seleccionado en el sidebar)
override = request.headers.get("X-Tenant-Override")
if override and str(override).strip().isdigit():
tenant_id = int(override)
# Fallback: campo tenant_id en el body (enviado por el frontend como respaldo)
elif data.tenant_id:
tenant_id = data.tenant_id
else:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Hub admin: selecciona un tenant en el sidebar antes de crear una empresa",
)
service = CompanyService(db)
new_company = service.create_company_manually(data, tenant_id=tenant_id)
@@ -151,6 +149,7 @@ async def create_company(
summary="Get companies with pagination",
)
async def list_companies(
request: Request,
page: int = 1,
page_size: int = 50,
name: Optional[str] = None,
@@ -161,12 +160,12 @@ async def list_companies(
"""Get paginated list of companies for current tenant with optional filters"""
_assert_permission_any_company(db, current_user, "cat_company.view")
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",
)
tenant_id = resolve_tenant_id_required(current_user, db=db)
# Hub admin: si hay X-Tenant-Override y aún no se resolvió tenant, filtra por ese tenant
if tenant_id is None and is_hub_admin(current_user):
override = request.headers.get("X-Tenant-Override")
if override and str(override).strip().isdigit():
tenant_id = int(override)
skip = (page - 1) * page_size
filters = {}
@@ -205,15 +204,32 @@ async def list_companies(
summary="Get all companies for current tenant",
)
async def get_my_companies(
request: Request,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
"""
Compañías accesibles: tenant del token/Hub (si existe) unión con membership en BD
(``user_company_roles``, ``user_company_permissions``, ``user_tenants``).
Un usuario solo con roles de app y sin ``tenant_id`` en /auth/me sigue pudiendo
listar sus compañías asignadas.
Hub admin: si hay X-Tenant-Override muestra solo ese tenant; si no, muestra todos.
"""
from core.security import collect_user_role_names
user_roles = collect_user_role_names(current_user)
# Hub admin: respeta X-Tenant-Override para filtrar por tenant seleccionado;
# sin override devuelve todo (visibilidad global).
if "hub_admin" in user_roles:
service = CompanyService(db)
override = request.headers.get("X-Tenant-Override")
query = db.query(Company).filter(Company.deleted_at.is_(None))
if override and str(override).strip().isdigit():
query = query.filter(Company.tenant_id == int(override))
all_companies = query.order_by(Company.name).all()
return [
CompanyResponseDTO.model_validate(service.flatten_company_dto(c))
for c in all_companies
]
_assert_permission_any_company(db, current_user, "cat_company.view")
user_id = current_user.get("sub") or current_user.get("id")
@@ -245,12 +261,7 @@ async def get_company(
"""Get a specific company by ID"""
_assert_permission_for_company(db, company_id, current_user, "cat_company.view")
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",
)
tenant_id = resolve_tenant_id_required(current_user, db=db, company_id=company_id)
service = CompanyService(db)
company = CompanyService.get_by_id(db, company_id, tenant_id, 0)
@@ -277,12 +288,7 @@ async def update_company(
"""Update a company"""
_assert_permission_for_company(db, company_id, current_user, "cat_company.edit")
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",
)
tenant_id = resolve_tenant_id_required(current_user, db=db, company_id=company_id)
service = CompanyService(db)
updated_company = service.update(db, company_id, tenant_id, 0, data)
@@ -354,12 +360,7 @@ async def delete_company(
"""Delete a company"""
_assert_permission_for_company(db, company_id, current_user, "cat_company.delete")
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",
)
tenant_id = resolve_tenant_id_required(current_user, db=db, company_id=company_id)
success = CompanyService.delete(db, company_id, tenant_id, 0)
if not success:
@@ -385,7 +386,7 @@ async def upload_company_logo(
"""Upload a logo for a company"""
_assert_permission_for_company(db, company_id, current_user, "cat_company.edit")
tenant_id = _resolve_tenant_id_int(current_user)
tenant_id = _resolve_tenant_id_int(current_user, db=db, company_id=company_id)
# Validar que la empresa existe
company = CompanyService.get_by_id(db, company_id, tenant_id, 0)
@@ -468,7 +469,7 @@ async def upload_company_certificate(
"""
_assert_permission_for_company(db, company_id, current_user, "cat_company.edit")
tenant_id = _resolve_tenant_id_int(current_user)
tenant_id = _resolve_tenant_id_int(current_user, db=db, company_id=company_id)
# Validar que la empresa existe
service = CompanyService(db)

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

@@ -44,7 +44,9 @@ class CompanyService:
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, Company.deleted_at.is_(None))
query = db.query(Company).filter(Company.deleted_at.is_(None))
if tenant_id is not None:
query = query.filter(Company.tenant_id == tenant_id)
# Apply filters if provided
if filters:
@@ -67,15 +69,10 @@ class CompanyService:
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,
Company.deleted_at.is_(None)
)
.first()
)
query = db.query(Company).filter(Company.id == company_id, Company.deleted_at.is_(None))
if tenant_id is not None:
query = query.filter(Company.tenant_id == tenant_id)
return query.first()
# ESTE ES EL MÉTODO VIEJO QUE CAUSABA PROBLEMAS (Lo dejamos por si acaso)
@staticmethod