Spaces:
Sleeping
Sleeping
File size: 2,471 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 |
"""
QueryService - Main entry point for database operations.
Factory class that provides access to SelectQuery, UpdateQuery, and DeleteQuery.
All query types inherit from BaseQuery for shared filtering logic.
"""
import os
from sqlalchemy.ext.asyncio import AsyncSession
from core.models import User
from services.db_service.select_query import SelectQuery
from services.db_service.update_query import UpdateQuery
from services.db_service.delete_query import DeleteQuery
class QueryService:
"""
Main query service factory.
Provides access to:
- SelectQuery: Read operations (SELECT)
- UpdateQuery: Update operations (UPDATE)
- DeleteQuery: Soft delete operations (soft DELETE)
Usage:
qs = QueryService(user, db)
# SELECT
jobs = await qs.select().execute(select(GeminiJob))
# UPDATE
await qs.update().update(Job, {...}, id=123)
# DELETE
await qs.delete().soft_delete(Job, id=123)
# Check admin status
if qs.is_admin():
# Admin-only logic
"""
def __init__(self, user: User, db: AsyncSession, is_system: bool = False):
"""
Initialize QueryService.
Args:
user: Current authenticated user (or system user for background ops)
db: Database session
is_system: True if this is a system/background operation (OAuth, webhooks, workers)
"""
self.user = user
self.db = db
self.is_system = is_system
def select(self) -> SelectQuery:
"""Get SelectQuery instance for read operations."""
return SelectQuery(self.user, self.db, self.is_system)
def update(self) -> UpdateQuery:
"""Get UpdateQuery instance for update operations."""
return UpdateQuery(self.user, self.db, self.is_system)
def delete(self) -> DeleteQuery:
"""Get DeleteQuery instance for soft delete operations."""
return DeleteQuery(self.user, self.db, self.is_system)
def get_query_service(user: User, db: AsyncSession) -> QueryService:
"""
Factory function to create QueryService.
Can be used as a FastAPI dependency:
from services.db_service import get_query_service
qs: QueryService = Depends(lambda user=Depends(get_current_user), db=Depends(get_db): get_query_service(user, db))
"""
return QueryService(user, db)
|