Spaces:
Paused
Paused
| from typing import Any, Optional | |
| from app.modules.users.service import UserService | |
| from core.plugin_system.facades import BaseFacade | |
| from core.plugin_system.permissions import PluginPermission | |
| class UserPluginFacade(BaseFacade): | |
| """ | |
| Facade for User operations exposed to plugins. | |
| Wrapped around UserService to ensure secure access. | |
| """ | |
| def __init__( | |
| self, service: UserService, plugin_id: str, permissions: list[str] = None | |
| ): | |
| super().__init__(service, plugin_id, permissions) | |
| def get_user(self, user_id: str) -> Optional[dict[str, Any]]: | |
| """ | |
| Get safe user details. | |
| Requires: READ_USER | |
| """ | |
| if not self._check_permission(PluginPermission.READ_USER): | |
| return None | |
| try: | |
| user = self._service.get_user(user_id) | |
| if not user: | |
| return None | |
| # Return safe dictionary, preventing modification of ORM object | |
| return { | |
| "id": user.id, | |
| "username": user.username, | |
| # Email might be PII, restrict based on stricter permission if needed | |
| # For now assuming READ_USER allows seeing email | |
| "email": user.email, | |
| "role": user.role, | |
| "is_active": user.is_active, | |
| "department": getattr(user, "department", None), | |
| } | |
| except Exception: | |
| # Service raises HTTPException if not found, we catch and return None | |
| return None | |
| def get_active_users_count(self) -> int: | |
| """ | |
| Get count of active users. | |
| Requires: READ_DATA | |
| """ | |
| if not self._check_permission(PluginPermission.READ_DATA): | |
| return 0 | |
| try: | |
| # Ideally service has count method. | |
| # For now return fake 0 or implement service logic | |
| return 0 | |
| except Exception: | |
| return 0 | |