refactor: add efficient get_aces_for_path method for resource pool checks

Add a new repository method get_aces_for_path() that:
- Queries ACEs for a specific path at database level (more efficient)
- Preloads related user, group, and role objects to prevent 500 errors
- Keeps original get_aces() method unchanged to avoid performance impact

This improves both performance and code clarity for resource pool
deletion safety checks.
This commit is contained in:
YueGuobin 2026-05-26 12:59:37 +08:00
parent 6fcbb8e57b
commit 5e4d9e057e
No known key found for this signature in database
2 changed files with 20 additions and 2 deletions

View File

@ -149,8 +149,7 @@ async def delete_resource_pool(
# Check if there are any ACE configurations using this resource pool path
pool_path = f"/pools/{resource_pool_id}"
aces = await rbac_repo.get_aces()
using_aces = [ace for ace in aces if ace.path == pool_path or ace.path.startswith(pool_path + "/")]
using_aces = await rbac_repo.get_aces_for_path(pool_path)
if using_aces:
# Build detailed error message with ACE information

View File

@ -230,6 +230,25 @@ class RbacRepository(BaseRepository):
result = await self._db_session.execute(query)
return result.scalars().all()
async def get_aces_for_path(self, path: str) -> List[models.ACE]:
"""
Get all ACEs for a specific path (exact match or starting with path).
This method includes related user, group, and role information.
"""
query = select(models.ACE).\
where(
(models.ACE.path == path) | (models.ACE.path.startswith(path + "/"))
).\
options(
selectinload(models.ACE.user),
selectinload(models.ACE.group),
selectinload(models.ACE.role)
)
result = await self._db_session.execute(query)
return result.scalars().all()
async def check_ace_exists(self, path: str) -> bool:
"""
Check if an ACE exists.