Spaces:
Sleeping
Sleeping
File size: 2,953 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 |
"""
SelectQuery - Read operations with access control.
Inherits from BaseQuery for shared filtering logic.
"""
import logging
from typing import TypeVar, Optional, List, Type
from fastapi import HTTPException, status as http_status
from sqlalchemy import Select, select, func
from services.db_service.base_query import BaseQuery
logger = logging.getLogger(__name__)
T = TypeVar('T')
class SelectQuery(BaseQuery):
"""
Handles SELECT operations with automatic filtering.
Inherits filtering logic from BaseQuery:
- User filtering (_apply_user_filter)
- Deleted record filtering (_filter_deleted)
- Admin checks (_check_admin, _is_admin)
"""
async def execute(self, query: Select) -> List[T]:
"""
Execute a query with automatic filtering.
Filtering is applied automatically based on:
- Model type (USER_SCOPED, ADMIN_ONLY)
- User's admin status
- Deleted records (always excluded)
Returns:
List of results
"""
query = self._apply_user_filter(query)
query = self._filter_deleted(query)
result = await self.db.execute(query)
return result.scalars().all()
async def execute_one(self, query: Select) -> Optional[T]:
"""
Execute a query expecting a single result.
Automatic filtering applied.
Returns:
Single result or None
"""
query = self._apply_user_filter(query)
query = self._filter_deleted(query)
result = await self.db.execute(query)
return result.scalar_one_or_none()
async def count(self, query: Select) -> int:
"""
Count query results with automatic filtering.
Returns:
Count of results
"""
query = self._apply_user_filter(query)
query = self._filter_deleted(query)
# Convert to count query
count_query = select(func.count()).select_from(query.alias())
result = await self.db.execute(count_query)
return result.scalar() or 0
async def count_deleted(self, model_class: Type[T]) -> int:
"""
Count soft-deleted records for a model.
Only admins can access this.
Returns:
Count of deleted records
Raises:
HTTPException: 403 if non-admin tries to access
"""
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(func.count()).select_from(model_class).where(
delete_col != None
)
result = await self.db.execute(query)
return result.scalar() or 0
|