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

feature/correcion-al-borrar-datos-borrar-tambien-del-bucket

Browse Source
This commit is contained in:
hreyes
2026-04-06 08:13:01 -06:00
parent 1355bf34c4
commit 0c25572d70
3 changed files with 144 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

79
backend/api/v1/modules/a76/audit_log/router.py
View File

@@ -89,9 +89,22 @@ def _companies_map(db: Session, tenant_id: int) -> Dict[str, str]:
return out
def _display_segment(part: str, prev_part: Optional[str], company_names: Dict[str, str]) -> str:
def _display_segment(
part: str,
prev_part: Optional[str],
company_names: Dict[str, str],
current_user_id: Optional[str] = None,
current_user_label: Optional[str] = None,
) -> str:
if prev_part == "companies":
return company_names.get(part, "Compania")
if prev_part == "users":
# Para carpetas de usuarios, mostrar un nombre amigable:
# - Si es el propio usuario actual, usar preferred_username/email/nombre.
# - Para otros IDs (UUIDs) mostrar un label genérico.
if current_user_id and part == str(current_user_id):
return (current_user_label or "").strip() or "Usuario"
return "Usuario"
if part in _SEGMENT_LABELS:
return _SEGMENT_LABELS[part]
# Evita exponer IDs puros en UI.
@@ -100,14 +113,27 @@ def _display_segment(part: str, prev_part: Optional[str], company_names: Dict[st
return part.replace("_", " ").strip().title() or "Elemento"
def _display_path(rel_path: str, company_names: Dict[str, str]) -> str:
def _display_path(
rel_path: str,
company_names: Dict[str, str],
current_user_id: Optional[str] = None,
current_user_label: Optional[str] = None,
) -> str:
if not rel_path:
return "Raiz de archivos"
parts = [p for p in rel_path.split("/") if p]
labels: List[str] = []
prev: Optional[str] = None
for part in parts:
labels.append(_display_segment(part, prev, company_names))
labels.append(
_display_segment(
part,
prev,
company_names,
current_user_id=current_user_id,
current_user_label=current_user_label,
)
)
prev = part
return " / ".join(labels)
@@ -122,7 +148,12 @@ def _display_file_name(filename: str) -> str:
return filename
def _build_breadcrumbs(rel_path: str, company_names: Dict[str, str]) -> List[AuditFileBreadcrumb]:
def _build_breadcrumbs(
rel_path: str,
company_names: Dict[str, str],
current_user_id: Optional[str] = None,
current_user_label: Optional[str] = None,
) -> List[AuditFileBreadcrumb]:
breadcrumbs: List[AuditFileBreadcrumb] = [
AuditFileBreadcrumb(path="", display_name="Raiz de archivos")
]
@@ -136,7 +167,13 @@ def _build_breadcrumbs(rel_path: str, company_names: Dict[str, str]) -> List[Aud
breadcrumbs.append(
AuditFileBreadcrumb(
path="/".join(acc),
display_name=_display_segment(part, prev, company_names),
display_name=_display_segment(
part,
prev,
company_names,
current_user_id=current_user_id,
current_user_label=current_user_label,
),
)
)
prev = part
@@ -240,6 +277,15 @@ async def list_tenant_files(
rel_path = _normalize_relative_path(path)
list_prefix = f"{tenant_prefix}{rel_path}/" if rel_path else tenant_prefix
# Datos del usuario actual para etiquetas amigables bajo /users/{id}/...
current_user_id = str(current_user.get("sub") or "")
current_user_label = (
(current_user.get("preferred_username") or "").strip()
or (current_user.get("name") or "").strip()
or (current_user.get("email") or "").strip()
or "Usuario"
)
data = list_objects_tree(
prefix=list_prefix,
delimiter="/",
@@ -254,7 +300,14 @@ async def list_tenant_files(
folders.append(
AuditFileFolderItem(
path=rel,
display_name=_display_path(rel, company_names).split(" / ")[-1],
# Para el gestor de archivos mostramos el nombre amigable del último segmento
# (empresa, usuario actual, etc.), no el ID bruto.
display_name=_display_path(
rel,
company_names,
current_user_id=current_user_id,
current_user_label=current_user_label,
).split(" / ")[-1],
)
)
@@ -276,8 +329,18 @@ async def list_tenant_files(
return AuditFileBrowserResponse(
current_path=rel_path,
display_path=_display_path(rel_path, company_names),
breadcrumbs=_build_breadcrumbs(rel_path, company_names),
display_path=_display_path(
rel_path,
company_names,
current_user_id=current_user_id,
current_user_label=current_user_label,
),
breadcrumbs=_build_breadcrumbs(
rel_path,
company_names,
current_user_id=current_user_id,
current_user_label=current_user_label,
),
folders=sorted(folders, key=lambda x: x.display_name.lower()),
files=sorted(files, key=lambda x: x.display_name.lower()),
next_token=data.get("next_continuation_token"),

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

@@ -10,6 +10,10 @@ from fastapi import HTTPException
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session
from core.config import settings
from core.s3_keys import tenant_company_prefix
from core.storage_s3 import delete_objects_with_prefix
from .dto import CompanyCreateDTO, CompanyResponseDTO, CompanyUpdateDTO
from .models import Company
from ...audit_log.services.service import AuditService
@@ -594,9 +598,24 @@ class CompanyService:
# ----------------------
try:
# 1) Borrado lógico en base de datos
company.deleted_at = datetime.utcnow()
db.flush()
# 2) Limpieza de objetos S3/MinIO asociados a la compañía
if settings.use_s3_object_storage:
try:
prefix = tenant_company_prefix(tenant_id, company_id)
delete_objects_with_prefix(prefix)
except Exception as e:
# No bloquear la eliminación lógica si falla la limpieza de objetos
logger.error(
"Error deleting S3 objects for company %s (tenant %s): %s",
company_id,
tenant_id,
e,
)
db.commit()
# --- Audit Log ---

53
backend/core/storage_s3.py
View File

@@ -94,6 +94,59 @@ def delete_object_if_exists(key: str) -> None:
logger.warning("delete_object %s: %s", key, e)
def delete_objects_with_prefix(prefix: str, batch_size: int = 1000) -> None:
"""
Elimina en cascada todos los objetos cuyo Key empieza con `prefix`.
Pensado para limpiar recursos ligados a una entidad (por ejemplo,
todos los objetos de una compañía bajo `tenants/{tid}/companies/{cid}/`).
"""
if not settings.use_s3_object_storage:
return
client = _client()
continuation_token: Optional[str] = None
while True:
params: Dict[str, Any] = {
"Bucket": settings.S3_BUCKET,
"Prefix": prefix,
"MaxKeys": max(1, min(int(batch_size), 1000)),
}
if continuation_token:
params["ContinuationToken"] = continuation_token
try:
resp = client.list_objects_v2(**params)
except ClientError as e:
logger.warning("list_objects_v2 for prefix %s failed: %s", prefix, e)
break
contents = resp.get("Contents") or []
if not contents:
break
to_delete = [{"Key": obj.get("Key")} for obj in contents if obj.get("Key")]
if to_delete:
try:
client.delete_objects(
Bucket=settings.S3_BUCKET,
Delete={"Objects": to_delete, "Quiet": True},
)
except ClientError as e:
logger.warning(
"delete_objects_with_prefix %s (batch_size=%s) failed: %s",
prefix,
len(to_delete),
e,
)
if not resp.get("IsTruncated"):
break
continuation_token = resp.get("NextContinuationToken")
def object_exists(key: str) -> bool:
try:
_client().head_object(Bucket=settings.S3_BUCKET, Key=key)