Spaces:
Sleeping
Sleeping
| """ | |
| DeleteQuery - Soft delete operations with access control. | |
| Inherits from BaseQuery for shared filtering logic. | |
| """ | |
| import logging | |
| from typing import Type, TypeVar, List | |
| from fastapi import HTTPException, status as http_status | |
| from sqlalchemy import update, select | |
| from sqlalchemy.sql import func | |
| from services.db_service.base_query import BaseQuery | |
| logger = logging.getLogger(__name__) | |
| T = TypeVar('T') | |
| class DeleteQuery(BaseQuery): | |
| """ | |
| Handles soft DELETE operations. | |
| Inherits filtering logic from BaseQuery: | |
| - User ownership checks (_is_admin) | |
| - Records are marked as deleted (deleted_at = NOW()) instead of being physically removed | |
| """ | |
| async def soft_delete(self, model_class: Type[T], **filters) -> int: | |
| """ | |
| Soft delete records matching filters. | |
| Sets deleted_at = NOW() instead of physically deleting. | |
| """ | |
| # Verify user has permission to delete this model | |
| self._verify_operation_access(model_class, 'delete') | |
| # Build update statement to set deleted_at | |
| delete_col = getattr(model_class, self._config.soft_delete_column) | |
| stmt = update(model_class).where( | |
| delete_col == None # Only soft-delete non-deleted records | |
| ) | |
| # Apply ownership filter (shared method from BaseQuery) | |
| stmt = self._apply_ownership_filter(stmt, model_class, 'delet') | |
| # Apply user's filters | |
| for key, value in filters.items(): | |
| if not hasattr(model_class, key): | |
| raise ValueError(f"{model_class.__name__} has no attribute '{key}'") | |
| stmt = stmt.where(getattr(model_class, key) == value) | |
| # Set deleted_at timestamp | |
| stmt = stmt.values({self._config.soft_delete_column: func.now()}) | |
| result = await self.db.execute(stmt) | |
| await self.db.commit() | |
| count = result.rowcount | |
| logger.info(f"Soft-deleted {count} {model_class.__name__} record(s)") | |
| return count | |
| async def soft_delete_one(self, instance: T) -> bool: | |
| """Soft delete a single model instance.""" | |
| # Check if already deleted | |
| delete_column = self._config.soft_delete_column | |
| if hasattr(instance, delete_column) and getattr(instance, delete_column) is not None: | |
| logger.warning(f"Attempted to delete already-deleted {instance.__class__.__name__}") | |
| return False | |
| # Verify user has permission | |
| self._verify_operation_access(instance.__class__, 'delete') | |
| # Check ownership for non-admins | |
| filter_column = self._config.user_filter_column | |
| if not self.is_admin and hasattr(instance, filter_column): | |
| if getattr(instance, filter_column) != self.user.id: | |
| raise HTTPException( | |
| status_code=http_status.HTTP_403_FORBIDDEN, | |
| detail="You do not have permission to delete this record" | |
| ) | |
| # Set deleted_at | |
| setattr(instance, self._config.soft_delete_column, func.now()) | |
| await self.db.commit() | |
| await self.db.refresh(instance) | |
| logger.info(f"Soft-deleted {instance.__class__.__name__} instance") | |
| return True | |
| async def restore(self, model_class: Type[T], **filters) -> int: | |
| """ | |
| Restore soft-deleted records. | |
| Only admins can restore. | |
| """ | |
| if not self.is_admin: | |
| raise HTTPException( | |
| status_code=http_status.HTTP_403_FORBIDDEN, | |
| detail="Only administrators can restore deleted records" | |
| ) | |
| # Build update to clear deleted_at | |
| delete_col = getattr(model_class, self._config.soft_delete_column) | |
| stmt = update(model_class).where( | |
| delete_col != None # Only restore deleted records | |
| ) | |
| # Apply filters | |
| for key, value in filters.items(): | |
| if not hasattr(model_class, key): | |
| raise ValueError(f"{model_class.__name__} has no attribute '{key}'") | |
| stmt = stmt.where(getattr(model_class, key) == value) | |
| # Clear deleted_at | |
| stmt = stmt.values({self._config.soft_delete_column: None}) | |
| result = await self.db.execute(stmt) | |
| await self.db.commit() | |
| count = result.rowcount | |
| logger.info(f"Admin {self.user.email} restored {count} {model_class.__name__} record(s)") | |
| return count | |
| async def restore_one(self, instance: T) -> bool: | |
| """Restore a single soft-deleted model instance. Admin only.""" | |
| if not self.is_admin: | |
| raise HTTPException( | |
| status_code=http_status.HTTP_403_FORBIDDEN, | |
| detail="Only administrators can restore deleted records" | |
| ) | |
| # Check if actually deleted | |
| delete_column = self._config.soft_delete_column | |
| if not hasattr(instance, delete_column) or getattr(instance, delete_column) is None: | |
| logger.warning(f"Attempted to restore non-deleted {instance.__class__.__name__}") | |
| return False | |
| # Clear deleted_at | |
| setattr(instance, self._config.soft_delete_column, None) | |
| await self.db.commit() | |
| await self.db.refresh(instance) | |
| logger.info(f"Admin restored {instance.__class__.__name__} instance") | |
| return True | |
| async def list_deleted(self, model_class: Type[T], limit: int = 100) -> List[T]: | |
| """List soft-deleted records. Admin only.""" | |
| if not self.is_admin: | |
| raise HTTPException( | |
| status_code=http_status.HTTP_403_FORBIDDEN, | |
| detail="Only administrators can view deleted records" | |
| ) | |
| delete_col = getattr(model_class, self._config.soft_delete_column) | |
| query = select(model_class).where( | |
| delete_col != None | |
| ).order_by(delete_col.desc()).limit(limit) | |
| result = await self.db.execute(query) | |
| return result.scalars().all() | |