Spaces:
Sleeping
Sleeping
File size: 6,123 Bytes
50c20bf |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 |
"""
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()
|