mirror of
https://github.com/Significant-Gravitas/AutoModerate.git
synced 2026-11-10 04:42:00 -10:00
1309 lines
53 KiB
Python
1309 lines
53 KiB
Python
"""
|
|
Async Centralized Database Service Layer for AutoModerate
|
|
High-performance async database operations with consistent error handling
|
|
"""
|
|
|
|
import asyncio
|
|
import logging
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from datetime import datetime
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
from sqlalchemy import func
|
|
from sqlalchemy.exc import IntegrityError, SQLAlchemyError
|
|
from sqlalchemy.orm import joinedload
|
|
|
|
from app import db
|
|
from app.models.api_key import APIKey
|
|
from app.models.api_user import APIUser
|
|
from app.models.content import Content
|
|
from app.models.moderation_result import ModerationResult
|
|
from app.models.moderation_rule import ModerationRule
|
|
from app.models.project import Project, ProjectMember
|
|
from app.models.user import User
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class DatabaseService:
|
|
"""Async centralized database operations with consistent error handling.
|
|
|
|
Every call runs inside a thread-pool worker so it doesn't block the event
|
|
loop. There is intentionally NO global lock around these operations:
|
|
Flask-SQLAlchemy's ``scoped_session`` hands each thread its own session,
|
|
and the underlying connection pool is already thread-safe. A previous
|
|
implementation held a single ``threading.RLock`` across every call, which
|
|
silently serialised every DB op in the entire process (turning the 8-wide
|
|
thread pool into a 1-wide one) and defeated the point of the async layer.
|
|
"""
|
|
|
|
def __init__(self):
|
|
from flask import current_app
|
|
try:
|
|
max_workers = current_app.config.get('DB_THREAD_POOL_WORKERS', 8)
|
|
except RuntimeError:
|
|
# No app context available during initialization
|
|
max_workers = 8
|
|
self._executor = ThreadPoolExecutor(
|
|
max_workers=max_workers, thread_name_prefix='db-service')
|
|
|
|
async def _safe_execute(self, operation_func, *args, **kwargs):
|
|
"""Execute a blocking DB operation in the thread pool, preserving
|
|
Flask's app context and normalising error handling.
|
|
"""
|
|
from flask import current_app, has_app_context
|
|
|
|
loop = asyncio.get_event_loop()
|
|
|
|
def _run_with_rollback(op):
|
|
try:
|
|
return op()
|
|
except SQLAlchemyError as e:
|
|
logger.error(f"Database error in thread: {str(e)}")
|
|
try:
|
|
db.session.rollback()
|
|
except SQLAlchemyError as rollback_error:
|
|
logger.error(f"Rollback error: {str(rollback_error)}")
|
|
raise
|
|
except (ValueError, TypeError, AttributeError) as e:
|
|
logger.error(f"Data processing error in thread: {str(e)}")
|
|
try:
|
|
db.session.rollback()
|
|
except SQLAlchemyError as rollback_error:
|
|
logger.error(f"Rollback error: {str(rollback_error)}")
|
|
raise
|
|
|
|
if has_app_context():
|
|
app = current_app._get_current_object()
|
|
|
|
def context_operation():
|
|
with app.app_context():
|
|
return _run_with_rollback(lambda: operation_func(*args, **kwargs))
|
|
target = context_operation
|
|
else:
|
|
# No app context, run directly (shouldn't happen in normal operation)
|
|
def bare_operation():
|
|
return _run_with_rollback(lambda: operation_func(*args, **kwargs))
|
|
target = bare_operation
|
|
|
|
try:
|
|
return await loop.run_in_executor(self._executor, target)
|
|
except SQLAlchemyError as e:
|
|
logger.error(f"Database error: {str(e)}")
|
|
return None
|
|
except (ValueError, TypeError, RuntimeError) as e:
|
|
logger.error(f"Runtime error: {str(e)}")
|
|
return None
|
|
|
|
# User Operations
|
|
async def create_user(self, username: str, email: str, password: str,
|
|
is_admin: bool = False, is_active: bool = True) -> Optional[User]:
|
|
"""Create a new user"""
|
|
def _create_user():
|
|
try:
|
|
user = User(username=username, email=email, is_admin=is_admin, is_active=is_active)
|
|
user.set_password(password)
|
|
db.session.add(user)
|
|
db.session.commit()
|
|
# Refresh the user to ensure it's attached to the session
|
|
db.session.refresh(user)
|
|
return user
|
|
except SQLAlchemyError:
|
|
db.session.rollback()
|
|
raise
|
|
|
|
return await self._safe_execute(_create_user)
|
|
|
|
async def get_user_by_email(self, email: str) -> Optional[User]:
|
|
"""Get user by email address"""
|
|
def _get_user():
|
|
return User.query.filter_by(email=email).first()
|
|
|
|
return await self._safe_execute(_get_user)
|
|
|
|
async def get_user_by_username(self, username: str) -> Optional[User]:
|
|
"""Get user by username"""
|
|
def _get_user():
|
|
return User.query.filter_by(username=username).first()
|
|
|
|
return await self._safe_execute(_get_user)
|
|
|
|
async def get_user_by_google_id(self, google_id: str) -> Optional[User]:
|
|
"""Get user by Google ID"""
|
|
def _get_user():
|
|
return User.query.filter_by(google_id=google_id).first()
|
|
|
|
return await self._safe_execute(_get_user)
|
|
|
|
async def create_google_user(self, username: str, email: str, google_id: str,
|
|
is_admin: bool = False, is_active: bool = True) -> Optional[User]:
|
|
"""Create a new user with Google SSO (no password)"""
|
|
def _create_user():
|
|
try:
|
|
user = User(username=username, email=email, google_id=google_id,
|
|
is_admin=is_admin, is_active=is_active)
|
|
db.session.add(user)
|
|
db.session.commit()
|
|
db.session.refresh(user)
|
|
return user
|
|
except SQLAlchemyError:
|
|
db.session.rollback()
|
|
raise
|
|
|
|
return await self._safe_execute(_create_user)
|
|
|
|
async def link_google_account(self, user_id: str, google_id: str) -> bool:
|
|
"""Link Google account to existing user"""
|
|
def _link_account():
|
|
try:
|
|
user = User.query.get(user_id)
|
|
if not user:
|
|
return False
|
|
user.google_id = google_id
|
|
db.session.commit()
|
|
return True
|
|
except SQLAlchemyError:
|
|
db.session.rollback()
|
|
raise
|
|
|
|
return await self._safe_execute(_link_account)
|
|
|
|
async def get_user_by_id(self, user_id: str) -> Optional[User]:
|
|
"""Get user by ID"""
|
|
def _get_user():
|
|
return User.query.get(user_id)
|
|
|
|
return await self._safe_execute(_get_user)
|
|
|
|
async def get_user_with_projects(self, user_id: str) -> Optional[User]:
|
|
"""Get user by ID with projects and related data loaded (DEPRECATED - causes OOM with large datasets)"""
|
|
def _get_user():
|
|
return User.query.options(
|
|
joinedload(User.projects)
|
|
.joinedload(Project.api_keys),
|
|
joinedload(User.projects)
|
|
.joinedload(Project.content)
|
|
).filter_by(id=user_id).first()
|
|
|
|
return await self._safe_execute(_get_user)
|
|
|
|
async def get_user_with_project_stats(self, user_id: str) -> Optional[dict]:
|
|
"""Get user with lightweight project statistics for profile page"""
|
|
def _get_user_stats():
|
|
from sqlalchemy import func
|
|
|
|
# Get basic user info
|
|
user = User.query.get(user_id)
|
|
if not user:
|
|
return None
|
|
|
|
# Get project stats efficiently with aggregated counts
|
|
project_stats = db.session.query(
|
|
Project.id,
|
|
Project.name,
|
|
Project.description,
|
|
Project.created_at,
|
|
func.count(APIKey.id.distinct()).label('api_key_count'),
|
|
func.count(Content.id.distinct()).label('content_count')
|
|
).outerjoin(APIKey, Project.id == APIKey.project_id)\
|
|
.outerjoin(Content, Project.id == Content.project_id)\
|
|
.filter(Project.user_id == user_id)\
|
|
.group_by(Project.id, Project.name, Project.description, Project.created_at)\
|
|
.all()
|
|
|
|
# Build user dict with stats (keep datetime objects for template)
|
|
user_data = {
|
|
'id': user.id,
|
|
'username': user.username,
|
|
'email': user.email,
|
|
'is_admin': user.is_admin,
|
|
'is_active': user.is_active,
|
|
'created_at': user.created_at, # Keep as datetime
|
|
'updated_at': user.updated_at, # Keep as datetime
|
|
'project_count': len(project_stats),
|
|
'total_api_keys': sum(stat.api_key_count for stat in project_stats),
|
|
'total_content': sum(stat.content_count for stat in project_stats),
|
|
'projects': [
|
|
{
|
|
'id': stat.id,
|
|
'name': stat.name,
|
|
'description': stat.description,
|
|
'created_at': stat.created_at,
|
|
'api_key_count': stat.api_key_count,
|
|
'content_count': stat.content_count
|
|
}
|
|
for stat in project_stats
|
|
]
|
|
}
|
|
|
|
return user_data
|
|
|
|
return await self._safe_execute(_get_user_stats)
|
|
|
|
async def update_user_profile(self, user: User, **kwargs) -> bool:
|
|
"""Update user profile information"""
|
|
def _update_user():
|
|
for key, value in kwargs.items():
|
|
if hasattr(user, key):
|
|
setattr(user, key, value)
|
|
db.session.commit()
|
|
return True
|
|
|
|
result = await self._safe_execute(_update_user)
|
|
return result is not None
|
|
|
|
async def toggle_user_admin_status(self, user_id: str) -> Optional[Dict[str, Any]]:
|
|
"""Toggle admin status for a user"""
|
|
def _toggle_admin():
|
|
user = User.query.get(user_id)
|
|
if not user:
|
|
return None
|
|
user.is_admin = not user.is_admin
|
|
db.session.commit()
|
|
# Return the needed data as a dictionary to avoid detached instance issues
|
|
return {
|
|
'id': user.id,
|
|
'username': user.username,
|
|
'email': user.email,
|
|
'is_admin': user.is_admin,
|
|
'is_active': user.is_active
|
|
}
|
|
|
|
return await self._safe_execute(_toggle_admin)
|
|
|
|
async def toggle_user_active_status(self, user_id: str) -> Optional[Dict[str, Any]]:
|
|
"""Toggle active status for a user"""
|
|
def _toggle_active():
|
|
user = User.query.get(user_id)
|
|
if not user:
|
|
return None
|
|
user.is_active = not user.is_active
|
|
db.session.commit()
|
|
# Return the needed data as a dictionary to avoid detached instance issues
|
|
return {
|
|
'id': user.id,
|
|
'username': user.username,
|
|
'email': user.email,
|
|
'is_admin': user.is_admin,
|
|
'is_active': user.is_active
|
|
}
|
|
|
|
return await self._safe_execute(_toggle_active)
|
|
|
|
async def update_user_password(self, user_id: str, new_password: str) -> bool:
|
|
"""Update user password"""
|
|
def _update_password():
|
|
user = User.query.get(user_id)
|
|
if not user:
|
|
return False
|
|
user.set_password(new_password)
|
|
db.session.commit()
|
|
return True
|
|
|
|
result = await self._safe_execute(_update_password)
|
|
return result is not None and result is not False
|
|
|
|
# Project Operations
|
|
async def create_project(self, name: str, description: str, user_id: str) -> Optional[Dict[str, Any]]:
|
|
"""Create a new project"""
|
|
def _create_project():
|
|
project = Project(
|
|
name=name, description=description, user_id=user_id)
|
|
db.session.add(project)
|
|
db.session.commit()
|
|
# Return the needed data as a dictionary to avoid detached instance issues
|
|
return {
|
|
'id': project.id,
|
|
'name': project.name,
|
|
'description': project.description,
|
|
'user_id': project.user_id,
|
|
'is_active': project.is_active,
|
|
'created_at': project.created_at,
|
|
'updated_at': project.updated_at
|
|
}
|
|
|
|
return await self._safe_execute(_create_project)
|
|
|
|
async def get_user_projects(self, user_id: str) -> List[Project]:
|
|
"""Get all projects accessible to a user (owned + member)"""
|
|
def _get_projects():
|
|
owned_project_ids = db.session.query(
|
|
Project.id).filter_by(user_id=user_id)
|
|
member_project_ids = db.session.query(Project.id).join(ProjectMember).filter(
|
|
ProjectMember.user_id == user_id
|
|
)
|
|
all_project_ids = owned_project_ids.union(member_project_ids)
|
|
return Project.query.filter(Project.id.in_(all_project_ids)).all()
|
|
|
|
return await self._safe_execute(_get_projects) or []
|
|
|
|
async def get_project_by_id(self, project_id: str) -> Optional[Project]:
|
|
"""Get project by ID with owner relationship loaded"""
|
|
def _get_project():
|
|
return Project.query.options(
|
|
joinedload(Project.owner)
|
|
).filter_by(id=project_id).first()
|
|
|
|
return await self._safe_execute(_get_project)
|
|
|
|
async def get_project_by_id_secure(self, project_id: str) -> Optional[Project]:
|
|
"""Get project by ID with 404 handling"""
|
|
def _get_project():
|
|
return Project.query.filter_by(id=project_id).first_or_404()
|
|
|
|
return await self._safe_execute(_get_project)
|
|
|
|
async def update_project(self, project: Project, **kwargs) -> bool:
|
|
"""Update project information"""
|
|
def _update_project():
|
|
for key, value in kwargs.items():
|
|
if hasattr(project, key):
|
|
setattr(project, key, value)
|
|
db.session.commit()
|
|
return True
|
|
|
|
result = await self._safe_execute(_update_project)
|
|
return result is not None
|
|
|
|
async def is_project_member(self, project_id: str, user_id: str) -> bool:
|
|
"""Check if user is a member or owner of a project"""
|
|
def _check_membership():
|
|
from app.models.project import ProjectMember
|
|
|
|
# Check if user is the owner
|
|
project = Project.query.get(project_id)
|
|
if not project:
|
|
return False
|
|
if project.user_id == user_id:
|
|
return True
|
|
|
|
# Check if user is a member
|
|
membership = ProjectMember.query.filter_by(
|
|
project_id=project_id,
|
|
user_id=user_id
|
|
).first()
|
|
return membership is not None
|
|
|
|
result = await self._safe_execute(_check_membership)
|
|
return result is not None and result
|
|
|
|
# Content Operations
|
|
async def create_content(self, project_id: str, content_text: str, content_type: str = 'text',
|
|
api_user_id: Optional[str] = None, meta_data: Optional[dict] = None) -> Optional[str]:
|
|
"""Create new content for moderation and return content ID"""
|
|
def _create_content():
|
|
content = Content(
|
|
project_id=project_id,
|
|
content_data=content_text,
|
|
content_type=content_type,
|
|
api_user_id=api_user_id,
|
|
meta_data=meta_data
|
|
)
|
|
db.session.add(content)
|
|
db.session.commit()
|
|
# Return just the ID to avoid detached instance issues
|
|
return content.id
|
|
|
|
return await self._safe_execute(_create_content)
|
|
|
|
async def get_project_content(self, project_id: str, limit: int = 100, offset: int = 0) -> List[Content]:
|
|
"""Get content for a project with pagination and moderation results"""
|
|
def _get_content():
|
|
return Content.query.filter_by(project_id=project_id)\
|
|
.options(joinedload(Content.moderation_results))\
|
|
.order_by(Content.created_at.desc())\
|
|
.limit(limit).offset(offset).all()
|
|
|
|
return await self._safe_execute(_get_content) or []
|
|
|
|
async def get_project_content_with_filters(self, project_id: str, status: str = None, limit: int = 50,
|
|
offset: int = 0) -> List[Content]:
|
|
"""Get content for a project with optional status filter and moderation results"""
|
|
def _get_content():
|
|
# Validate inputs to prevent any potential issues
|
|
validated_limit = limit
|
|
validated_offset = offset
|
|
if not isinstance(limit, int) or limit < 1 or limit > 1000:
|
|
validated_limit = 50
|
|
if not isinstance(offset, int) or offset < 0:
|
|
validated_offset = 0
|
|
|
|
query = Content.query.filter_by(project_id=project_id)\
|
|
.options(joinedload(Content.moderation_results))
|
|
if status and isinstance(status, str):
|
|
# Validate status against known values
|
|
valid_statuses = {'approved', 'rejected', 'flagged', 'pending'}
|
|
if status in valid_statuses:
|
|
query = query.filter_by(status=status)
|
|
return query.order_by(Content.created_at.desc()).limit(validated_limit).offset(validated_offset).all()
|
|
|
|
return await self._safe_execute(_get_content) or []
|
|
|
|
async def get_content_by_id(self, content_id: str) -> Optional[Content]:
|
|
"""Get content by ID with moderation results"""
|
|
def _get_content():
|
|
return Content.query.options(
|
|
joinedload(Content.moderation_results)
|
|
).filter_by(id=content_id).first()
|
|
|
|
return await self._safe_execute(_get_content)
|
|
|
|
async def get_content_by_id_and_project(self, content_id: str, project_id: str) -> Optional[Content]:
|
|
"""Get content by ID and project ID with moderation results"""
|
|
def _get_content():
|
|
return Content.query.options(
|
|
joinedload(Content.moderation_results)
|
|
).filter_by(id=content_id, project_id=project_id).first()
|
|
|
|
return await self._safe_execute(_get_content)
|
|
|
|
# API Key Operations
|
|
async def create_api_key(self, project_id: str, name: str, key_value: str) -> Optional[APIKey]:
|
|
"""Create new API key"""
|
|
def _create_key():
|
|
api_key = APIKey(project_id=project_id, name=name,
|
|
key=APIKey.hash_key(key_value))
|
|
db.session.add(api_key)
|
|
db.session.commit()
|
|
api_key.plaintext_key = key_value
|
|
return api_key
|
|
|
|
return await self._safe_execute(_create_key)
|
|
|
|
async def get_api_key_by_value(self, key_value: str) -> Optional[APIKey]:
|
|
"""Resolve a presented plaintext key to its record via its hash.
|
|
|
|
Legacy rows still holding plaintext are matched directly and upgraded
|
|
to their hash on first use.
|
|
"""
|
|
def _get_key():
|
|
hashed = APIKey.hash_key(key_value)
|
|
api_key = APIKey.query.options(joinedload(APIKey.project)).filter_by(
|
|
key=hashed, is_active=True).first()
|
|
if api_key is not None:
|
|
return api_key
|
|
|
|
legacy = APIKey.query.filter_by(key=key_value, is_active=True).first()
|
|
if legacy is None:
|
|
return None
|
|
legacy.key = hashed
|
|
try:
|
|
db.session.commit()
|
|
except SQLAlchemyError:
|
|
db.session.rollback()
|
|
return None
|
|
return APIKey.query.options(joinedload(APIKey.project)).filter_by(
|
|
key=hashed, is_active=True).first()
|
|
|
|
return await self._safe_execute(_get_key)
|
|
|
|
async def get_project_api_keys(self, project_id: str) -> List[APIKey]:
|
|
"""Get all API keys for a project"""
|
|
def _get_keys():
|
|
return APIKey.query.filter_by(project_id=project_id).all()
|
|
|
|
return await self._safe_execute(_get_keys) or []
|
|
|
|
# Moderation Rule Operations
|
|
async def create_moderation_rule(self, project_id: str, name: str, rule_type: str,
|
|
rule_data: dict, action: str, priority: int = 0,
|
|
description: str = '') -> Optional[Dict[str, Any]]:
|
|
"""Create new moderation rule"""
|
|
def _create_rule():
|
|
rule = ModerationRule(
|
|
project_id=project_id,
|
|
name=name,
|
|
description=description,
|
|
rule_type=rule_type,
|
|
rule_data=rule_data,
|
|
action=action,
|
|
priority=priority
|
|
)
|
|
db.session.add(rule)
|
|
db.session.commit()
|
|
# Return the needed data as a dictionary to avoid detached instance issues
|
|
return {
|
|
'id': rule.id,
|
|
'project_id': rule.project_id,
|
|
'name': rule.name,
|
|
'description': rule.description,
|
|
'rule_type': rule.rule_type,
|
|
'rule_data': rule.rule_data,
|
|
'action': rule.action,
|
|
'priority': rule.priority,
|
|
'is_active': rule.is_active,
|
|
'created_at': rule.created_at,
|
|
'updated_at': rule.updated_at
|
|
}
|
|
|
|
return await self._safe_execute(_create_rule)
|
|
|
|
async def get_project_rules(self, project_id: str) -> List[ModerationRule]:
|
|
"""Get all moderation rules for a project (active and inactive)"""
|
|
def _get_rules():
|
|
return ModerationRule.query.filter_by(project_id=project_id)\
|
|
.order_by(ModerationRule.priority.desc()).all()
|
|
|
|
return await self._safe_execute(_get_rules) or []
|
|
|
|
async def get_rule_by_id(self, rule_id: str) -> Optional[ModerationRule]:
|
|
"""Get moderation rule by ID"""
|
|
def _get_rule():
|
|
return ModerationRule.query.get(rule_id)
|
|
|
|
return await self._safe_execute(_get_rule)
|
|
|
|
async def get_all_rules_for_project(self, project_id: str, include_inactive: bool = True) -> List[ModerationRule]:
|
|
"""Get all rules for a project"""
|
|
def _get_rules():
|
|
query = ModerationRule.query.filter_by(project_id=project_id)
|
|
if not include_inactive:
|
|
query = query.filter_by(is_active=True)
|
|
return query.order_by(ModerationRule.priority.desc()).all()
|
|
|
|
return await self._safe_execute(_get_rules) or []
|
|
|
|
# Statistics and Analytics
|
|
async def get_project_stats(self, project_id: str) -> Dict[str, Any]:
|
|
"""Get comprehensive project statistics"""
|
|
def _get_stats():
|
|
from datetime import datetime, timedelta
|
|
|
|
total_content = Content.query.filter_by(
|
|
project_id=project_id).count()
|
|
total_rules = ModerationRule.query.filter_by(
|
|
project_id=project_id, is_active=True).count()
|
|
total_api_keys = APIKey.query.filter_by(
|
|
project_id=project_id, is_active=True).count()
|
|
flagged_content = Content.query.filter_by(
|
|
project_id=project_id, status='flagged').count()
|
|
approved_content = Content.query.filter_by(
|
|
project_id=project_id, status='approved').count()
|
|
|
|
# Recent activity (last 7 days)
|
|
week_ago = datetime.utcnow() - timedelta(days=7)
|
|
recent_content = Content.query.filter(
|
|
Content.project_id == project_id,
|
|
Content.created_at >= week_ago
|
|
).count()
|
|
|
|
return {
|
|
'total_content': total_content,
|
|
'total_rules': total_rules,
|
|
'total_api_keys': total_api_keys,
|
|
'flagged_content': flagged_content,
|
|
'approved_content': approved_content,
|
|
'recent_content': recent_content
|
|
}
|
|
|
|
return await self._safe_execute(_get_stats) or {}
|
|
|
|
async def get_admin_stats(self) -> Dict[str, Any]:
|
|
"""Get system-wide statistics for admin dashboard"""
|
|
def _get_stats():
|
|
from datetime import datetime, timedelta
|
|
|
|
week_ago = datetime.utcnow() - timedelta(days=7)
|
|
|
|
return {
|
|
'total_users': User.query.count(),
|
|
'active_users': User.query.filter_by(is_active=True).count(),
|
|
'total_projects': Project.query.count(),
|
|
'total_content': Content.query.count(),
|
|
'total_rules': ModerationRule.query.filter_by(is_active=True).count(),
|
|
'flagged_content': Content.query.filter_by(status='flagged').count(),
|
|
'new_users_week': User.query.filter(User.created_at >= week_ago).count(),
|
|
'new_content_week': Content.query.filter(Content.created_at >= week_ago).count(),
|
|
}
|
|
|
|
return await self._safe_execute(_get_stats) or {}
|
|
|
|
async def get_analytics_stats(self) -> Dict[str, Any]:
|
|
"""Get comprehensive analytics statistics"""
|
|
def _get_analytics_stats():
|
|
return {
|
|
'total_users': User.query.count(),
|
|
'active_users': User.query.filter_by(is_active=True).count(),
|
|
'total_projects': Project.query.count(),
|
|
'total_content': Content.query.count(),
|
|
'total_moderations': ModerationResult.query.count(),
|
|
'total_api_keys': APIKey.query.filter_by(is_active=True).count(),
|
|
}
|
|
|
|
return await self._safe_execute(_get_analytics_stats) or {}
|
|
|
|
async def get_api_stats(self) -> Dict[str, Any]:
|
|
"""Get API statistics with user, project, content, and moderation breakdowns"""
|
|
def _get_api_stats():
|
|
from datetime import datetime, timedelta
|
|
|
|
week_ago = datetime.utcnow() - timedelta(days=7)
|
|
|
|
return {
|
|
'users': {
|
|
'total': User.query.count(),
|
|
'active': User.query.filter_by(is_active=True).count(),
|
|
'admin': User.query.filter_by(is_admin=True).count(),
|
|
'recent': User.query.filter(User.created_at >= week_ago).count()
|
|
},
|
|
'projects': {
|
|
'total': Project.query.count(),
|
|
'recent': Project.query.filter(Project.created_at >= week_ago).count()
|
|
},
|
|
'content': {
|
|
'total': Content.query.count(),
|
|
'recent': Content.query.filter(Content.created_at >= week_ago).count()
|
|
},
|
|
'moderations': {
|
|
'total': ModerationResult.query.count(),
|
|
'approved': ModerationResult.query.filter_by(decision='approved').count(),
|
|
'rejected': ModerationResult.query.filter_by(decision='rejected').count(),
|
|
'flagged': ModerationResult.query.filter_by(decision='flagged').count(),
|
|
'recent': ModerationResult.query.filter(
|
|
ModerationResult.created_at >= week_ago
|
|
).count()
|
|
}
|
|
}
|
|
|
|
return await self._safe_execute(_get_api_stats) or {}
|
|
|
|
async def get_project_analytics_stats(self, project_id: str, start_date, end_date) -> Dict[str, Any]:
|
|
"""Get comprehensive analytics statistics for a specific project"""
|
|
def _get_project_analytics():
|
|
# Validate date parameters to prevent issues
|
|
validated_start_date = start_date
|
|
validated_end_date = end_date
|
|
if not start_date or not end_date:
|
|
from datetime import datetime, timedelta
|
|
validated_end_date = datetime.utcnow()
|
|
validated_start_date = validated_end_date - timedelta(days=30)
|
|
# Project overview stats
|
|
project_stats = {
|
|
'total_content': Content.query.filter_by(project_id=project_id).count(),
|
|
'total_moderations': ModerationResult.query.select_from(ModerationResult).join(
|
|
Content, ModerationResult.content_id == Content.id
|
|
).filter(Content.project_id == project_id).count(),
|
|
'total_rules': ModerationRule.query.filter_by(project_id=project_id).count(),
|
|
'total_api_keys': APIKey.query.filter_by(project_id=project_id).count(),
|
|
'active_api_keys': APIKey.query.filter_by(
|
|
project_id=project_id, is_active=True
|
|
).count(),
|
|
}
|
|
|
|
# Content submissions over time - using parameterized queries
|
|
content_submissions = db.session.query(
|
|
func.date(Content.created_at).label('date'),
|
|
func.count(Content.id).label('count')
|
|
).filter(
|
|
Content.project_id == project_id,
|
|
Content.created_at >= validated_start_date,
|
|
Content.created_at <= validated_end_date
|
|
).group_by(
|
|
func.date(Content.created_at)
|
|
).order_by('date').all()
|
|
|
|
# Moderation decisions breakdown - using parameterized queries
|
|
moderation_decisions = db.session.query(
|
|
ModerationResult.decision,
|
|
func.count(ModerationResult.id).label('count')
|
|
).select_from(ModerationResult).join(
|
|
Content, ModerationResult.content_id == Content.id
|
|
).filter(
|
|
Content.project_id == project_id,
|
|
ModerationResult.created_at >= validated_start_date,
|
|
ModerationResult.created_at <= validated_end_date
|
|
).group_by(ModerationResult.decision).all()
|
|
|
|
# Content type distribution
|
|
content_types = db.session.query(
|
|
Content.content_type,
|
|
func.count(Content.id).label('count')
|
|
).filter(
|
|
Content.project_id == project_id,
|
|
Content.created_at >= validated_start_date,
|
|
Content.created_at <= validated_end_date
|
|
).group_by(Content.content_type).all()
|
|
|
|
# Processing time stats for this project
|
|
processing_stats = db.session.query(
|
|
func.avg(ModerationResult.processing_time).label('avg_time'),
|
|
func.min(ModerationResult.processing_time).label('min_time'),
|
|
func.max(ModerationResult.processing_time).label('max_time')
|
|
).select_from(ModerationResult).join(
|
|
Content, ModerationResult.content_id == Content.id
|
|
).filter(
|
|
Content.project_id == project_id,
|
|
ModerationResult.created_at >= validated_start_date,
|
|
ModerationResult.created_at <= validated_end_date,
|
|
ModerationResult.processing_time.isnot(None)
|
|
).first()
|
|
|
|
# API keys and their details for this project
|
|
api_usage = db.session.query(
|
|
APIKey.name,
|
|
APIKey.usage_count,
|
|
APIKey.last_used,
|
|
APIKey.is_active
|
|
).filter(
|
|
APIKey.project_id == project_id
|
|
).order_by(APIKey.usage_count.desc()).limit(10).all()
|
|
|
|
# Top users for this project
|
|
from sqlalchemy import case
|
|
|
|
from app.models.api_user import APIUser
|
|
top_users = db.session.query(
|
|
APIUser.external_user_id,
|
|
func.count(Content.id).label('submissions'),
|
|
func.sum(case(
|
|
(ModerationResult.decision == 'approved', 1),
|
|
else_=0
|
|
)).label('approved'),
|
|
func.sum(case(
|
|
(ModerationResult.decision == 'rejected', 1),
|
|
else_=0
|
|
)).label('rejected')
|
|
).select_from(APIUser).join(
|
|
Content, APIUser.id == Content.api_user_id
|
|
).join(
|
|
ModerationResult, Content.id == ModerationResult.content_id
|
|
).filter(
|
|
Content.project_id == project_id,
|
|
Content.created_at >= validated_start_date,
|
|
Content.created_at <= validated_end_date
|
|
).group_by(APIUser.external_user_id).order_by(
|
|
func.count(Content.id).desc()
|
|
).limit(10).all()
|
|
|
|
# Rule effectiveness for this project
|
|
rule_stats = db.session.query(
|
|
ModerationRule.name,
|
|
func.count(ModerationResult.id).label('triggered'),
|
|
ModerationRule.rule_type
|
|
).select_from(ModerationRule).join(
|
|
ModerationResult, ModerationRule.id == ModerationResult.moderator_id
|
|
).join(Content, ModerationResult.content_id == Content.id).filter(
|
|
ModerationRule.project_id == project_id,
|
|
ModerationResult.created_at >= validated_start_date,
|
|
ModerationResult.created_at <= validated_end_date,
|
|
ModerationResult.moderator_type == 'rule',
|
|
ModerationResult.moderator_id.isnot(None)
|
|
).group_by(ModerationRule.id).order_by(
|
|
func.count(ModerationResult.id).desc()
|
|
).limit(10).all()
|
|
|
|
# Recent activity summary
|
|
from app.models.api_user import APIUser
|
|
recent_content = Content.query.options(
|
|
joinedload(Content.api_user)
|
|
).filter_by(
|
|
project_id=project_id
|
|
).order_by(Content.created_at.desc()).limit(5).all()
|
|
|
|
return {
|
|
'project_stats': project_stats,
|
|
'content_submissions': content_submissions,
|
|
'moderation_decisions': moderation_decisions,
|
|
'content_types': content_types,
|
|
'processing_stats': processing_stats,
|
|
'api_usage': api_usage,
|
|
'top_users': top_users,
|
|
'rule_stats': rule_stats,
|
|
'recent_content': recent_content
|
|
}
|
|
|
|
return await self._safe_execute(_get_project_analytics) or {}
|
|
|
|
# API User Operations
|
|
async def get_or_create_api_user(self, external_user_id: str, project_id: str) -> Optional[str]:
|
|
"""Get existing API user or create new one, returns the user ID.
|
|
|
|
Races safely against the (project_id, external_user_id) unique
|
|
constraint: if a concurrent request inserts the same pair first we
|
|
catch the IntegrityError, roll back, and re-read the row the winner
|
|
committed.
|
|
"""
|
|
def _get_or_create():
|
|
api_user = APIUser.query.filter_by(
|
|
external_user_id=external_user_id,
|
|
project_id=project_id
|
|
).first()
|
|
|
|
if api_user:
|
|
return api_user.id
|
|
|
|
api_user = APIUser(
|
|
external_user_id=external_user_id,
|
|
project_id=project_id,
|
|
)
|
|
db.session.add(api_user)
|
|
try:
|
|
db.session.commit()
|
|
return api_user.id
|
|
except IntegrityError:
|
|
db.session.rollback()
|
|
existing = APIUser.query.filter_by(
|
|
external_user_id=external_user_id,
|
|
project_id=project_id,
|
|
).first()
|
|
return existing.id if existing else None
|
|
|
|
return await self._safe_execute(_get_or_create)
|
|
|
|
async def get_api_user_by_id(self, api_user_id: str) -> Optional[APIUser]:
|
|
"""Get API user by ID"""
|
|
def _get_user():
|
|
return APIUser.query.get(api_user_id)
|
|
|
|
return await self._safe_execute(_get_user)
|
|
|
|
async def increment_api_user_stats(self, api_user_id: str, status: str) -> bool:
|
|
"""Atomically bump an APIUser's request counters for a moderation decision.
|
|
|
|
Uses a single ``UPDATE ... SET col = col + 1`` statement rather than a
|
|
read-modify-write on a fetched instance. Two concurrent moderations for
|
|
the same external user can otherwise both read the same counter value
|
|
and both write ``value + 1``, silently losing one increment. Pushing the
|
|
arithmetic into the database makes each increment atomic under the DB's
|
|
row lock.
|
|
|
|
Returns True on success, False if the user wasn't found or the update
|
|
failed.
|
|
"""
|
|
def _increment():
|
|
values = {
|
|
APIUser.total_requests: APIUser.total_requests + 1,
|
|
APIUser.last_seen: datetime.utcnow(),
|
|
}
|
|
if status == 'approved':
|
|
values[APIUser.approved_count] = APIUser.approved_count + 1
|
|
elif status == 'rejected':
|
|
values[APIUser.rejected_count] = APIUser.rejected_count + 1
|
|
elif status == 'flagged':
|
|
values[APIUser.flagged_count] = APIUser.flagged_count + 1
|
|
|
|
updated = db.session.query(APIUser).filter_by(
|
|
id=api_user_id).update(values, synchronize_session=False)
|
|
db.session.commit()
|
|
return updated > 0
|
|
|
|
result = await self._safe_execute(_increment)
|
|
return bool(result)
|
|
|
|
# Content Query Operations
|
|
async def get_content_counts_by_status(self, project_id: str) -> Dict[str, int]:
|
|
"""Get content counts by moderation status - optimized single query"""
|
|
def _get_counts():
|
|
# Use GROUP BY to get all counts in a single query
|
|
status_counts = db.session.query(
|
|
Content.status,
|
|
func.count(Content.id).label('count')
|
|
).filter(
|
|
Content.project_id == project_id
|
|
).group_by(Content.status).all()
|
|
|
|
# Convert to dictionary
|
|
counts = {status: count for status, count in status_counts}
|
|
|
|
# Calculate total from the status counts
|
|
total = sum(counts.values())
|
|
|
|
return {
|
|
'total': total,
|
|
'approved': counts.get('approved', 0),
|
|
'rejected': counts.get('rejected', 0),
|
|
'flagged': counts.get('flagged', 0),
|
|
'pending': counts.get('pending', 0),
|
|
'error': counts.get('error', 0)
|
|
}
|
|
|
|
return await self._safe_execute(_get_counts) or {
|
|
'total': 0, 'approved': 0, 'rejected': 0, 'flagged': 0, 'pending': 0, 'error': 0
|
|
}
|
|
|
|
async def update_content_status(self, content_id: str, **kwargs) -> bool:
|
|
"""Update content status and flags"""
|
|
def _update_content():
|
|
content = Content.query.get(content_id)
|
|
if not content:
|
|
return False
|
|
|
|
for key, value in kwargs.items():
|
|
if hasattr(content, key):
|
|
setattr(content, key, value)
|
|
|
|
db.session.commit()
|
|
return True
|
|
|
|
result = await self._safe_execute(_update_content)
|
|
return result is not None and result is not False
|
|
|
|
async def save_moderation_outcome(self, content_id: str, status: str,
|
|
api_user_id: Optional[str],
|
|
moderation_results: List) -> bool:
|
|
"""Persist a complete moderation outcome in a single transaction.
|
|
|
|
Sets the content status, increments the API-user counters (atomically),
|
|
and inserts every ModerationResult row under one commit. Previously these
|
|
were three independent commits followed by a no-op commit, so a failure
|
|
partway through could leave content marked approved/rejected with no
|
|
result rows saved. Doing it as one unit makes the outcome all-or-nothing.
|
|
"""
|
|
def _save():
|
|
content = Content.query.get(content_id)
|
|
if content is None:
|
|
return False
|
|
|
|
content.status = status
|
|
|
|
if api_user_id:
|
|
values = {
|
|
APIUser.total_requests: APIUser.total_requests + 1,
|
|
APIUser.last_seen: datetime.utcnow(),
|
|
}
|
|
if status == 'approved':
|
|
values[APIUser.approved_count] = APIUser.approved_count + 1
|
|
elif status == 'rejected':
|
|
values[APIUser.rejected_count] = APIUser.rejected_count + 1
|
|
elif status == 'flagged':
|
|
values[APIUser.flagged_count] = APIUser.flagged_count + 1
|
|
db.session.query(APIUser).filter_by(
|
|
id=api_user_id).update(values, synchronize_session=False)
|
|
|
|
for obj in moderation_results:
|
|
db.session.add(obj)
|
|
|
|
db.session.commit()
|
|
return True
|
|
|
|
result = await self._safe_execute(_save)
|
|
return bool(result)
|
|
|
|
# API Key Management
|
|
async def update_api_key_usage(self, api_key: APIKey) -> bool:
|
|
"""Update API key usage statistics.
|
|
|
|
Single atomic ``UPDATE`` on the counter so concurrent requests using the
|
|
same key don't lose increments to a read-modify-write race.
|
|
"""
|
|
def _update_usage():
|
|
updated = db.session.query(APIKey).filter_by(id=api_key.id).update(
|
|
{
|
|
APIKey.usage_count: APIKey.usage_count + 1,
|
|
APIKey.last_used: datetime.utcnow(),
|
|
},
|
|
synchronize_session=False,
|
|
)
|
|
db.session.commit()
|
|
return updated > 0
|
|
|
|
result = await self._safe_execute(_update_usage)
|
|
return bool(result)
|
|
|
|
# Transaction Management
|
|
async def commit_transaction(self) -> bool:
|
|
"""Commit current transaction"""
|
|
def _commit():
|
|
db.session.commit()
|
|
return True
|
|
|
|
result = await self._safe_execute(_commit)
|
|
return result is not None
|
|
|
|
async def rollback_transaction(self) -> None:
|
|
"""Rollback current transaction"""
|
|
def _rollback():
|
|
db.session.rollback()
|
|
|
|
await self._safe_execute(_rollback)
|
|
|
|
# Admin-specific methods
|
|
async def get_recent_users(self, limit: int = 5) -> List[User]:
|
|
"""Get recent users for admin dashboard"""
|
|
def _get_recent():
|
|
return User.query.order_by(User.created_at.desc()).limit(limit).all()
|
|
|
|
return await self._safe_execute(_get_recent) or []
|
|
|
|
async def get_recent_projects(self, limit: int = 5) -> List[Project]:
|
|
"""Get recent projects for admin dashboard"""
|
|
def _get_recent():
|
|
return Project.query.options(
|
|
joinedload(Project.owner)
|
|
).order_by(Project.created_at.desc()).limit(limit).all()
|
|
|
|
return await self._safe_execute(_get_recent) or []
|
|
|
|
async def get_recent_content_admin(self, limit: int = 10) -> List[Content]:
|
|
"""Get recent content for admin dashboard"""
|
|
def _get_recent():
|
|
return Content.query.options(
|
|
joinedload(Content.project).joinedload(Project.owner)
|
|
).order_by(Content.created_at.desc()).limit(limit).all()
|
|
|
|
return await self._safe_execute(_get_recent) or []
|
|
|
|
async def get_moderation_result_stats(self) -> Dict[str, Any]:
|
|
"""Get moderation result statistics"""
|
|
def _get_stats():
|
|
return {
|
|
'total_moderations': ModerationResult.query.count(),
|
|
'approved_decisions': ModerationResult.query.filter_by(decision='approved').count(),
|
|
'rejected_decisions': ModerationResult.query.filter_by(decision='rejected').count(),
|
|
'flagged_decisions': ModerationResult.query.filter_by(decision='flagged').count(),
|
|
}
|
|
|
|
return await self._safe_execute(_get_stats) or {}
|
|
|
|
async def get_user_projects_for_admin(self, user_id: str) -> List[Dict]:
|
|
"""Get user projects for admin view with counts instead of loading all related data"""
|
|
def _get_projects():
|
|
# Get projects with counts using subqueries to avoid N+1 problem
|
|
content_count = db.session.query(func.count(Content.id)).filter(
|
|
Content.project_id == Project.id
|
|
).scalar_subquery()
|
|
|
|
rules_count = db.session.query(func.count(ModerationRule.id)).filter(
|
|
ModerationRule.project_id == Project.id
|
|
).scalar_subquery()
|
|
|
|
api_keys_count = db.session.query(func.count(APIKey.id)).filter(
|
|
APIKey.project_id == Project.id
|
|
).scalar_subquery()
|
|
|
|
projects = db.session.query(
|
|
Project,
|
|
content_count.label('content_count'),
|
|
rules_count.label('rules_count'),
|
|
api_keys_count.label('api_keys_count')
|
|
).filter_by(user_id=user_id).all()
|
|
|
|
# Convert to dict format for template
|
|
result = []
|
|
for project, content_count, rules_count, api_keys_count in projects:
|
|
result.append({
|
|
'project': project,
|
|
'content_count': content_count or 0,
|
|
'rules_count': rules_count or 0,
|
|
'api_keys_count': api_keys_count or 0
|
|
})
|
|
|
|
return result
|
|
|
|
return await self._safe_execute(_get_projects) or []
|
|
|
|
async def get_all_projects_for_admin(self, page: int = 1, per_page: int = 20) -> List[Project]:
|
|
"""Get all projects for admin view with pagination (lightweight)"""
|
|
def _get_projects():
|
|
offset = (page - 1) * per_page
|
|
# Only load owner data, not all content/rules/keys which can be huge
|
|
return Project.query.options(
|
|
joinedload(Project.owner)
|
|
).offset(offset).limit(per_page).all()
|
|
|
|
return await self._safe_execute(_get_projects) or []
|
|
|
|
async def get_project_bulk_stats(self, project_ids: list) -> dict:
|
|
"""Get bulk statistics for multiple projects"""
|
|
def _get_stats():
|
|
# Get counts for all projects in bulk
|
|
content_counts = dict(db.session.query(
|
|
Content.project_id, func.count(Content.id)
|
|
).filter(Content.project_id.in_(project_ids)).group_by(Content.project_id).all())
|
|
|
|
rules_counts = dict(db.session.query(
|
|
ModerationRule.project_id, func.count(ModerationRule.id)
|
|
).filter(ModerationRule.project_id.in_(project_ids)).group_by(ModerationRule.project_id).all())
|
|
|
|
keys_counts = dict(db.session.query(
|
|
APIKey.project_id, func.count(APIKey.id)
|
|
).filter(APIKey.project_id.in_(project_ids)).group_by(APIKey.project_id).all())
|
|
|
|
return {
|
|
'content_counts': content_counts,
|
|
'rules_counts': rules_counts,
|
|
'keys_counts': keys_counts
|
|
}
|
|
|
|
return await self._safe_execute(_get_stats)
|
|
|
|
async def get_flagged_content_for_projects(self, project_ids: list, page: int = 1, per_page: int = 50) -> tuple:
|
|
"""Get flagged content for multiple projects with pagination"""
|
|
def _get_flagged_content():
|
|
if not project_ids:
|
|
return [], 0
|
|
|
|
query = Content.query.filter(
|
|
Content.project_id.in_(project_ids),
|
|
Content.status == 'flagged'
|
|
).options(
|
|
joinedload(Content.moderation_results),
|
|
joinedload(Content.project)
|
|
).order_by(Content.created_at.desc())
|
|
|
|
# Get total count
|
|
total = query.count()
|
|
|
|
# Apply pagination
|
|
offset = (page - 1) * per_page
|
|
items = query.offset(offset).limit(per_page).all()
|
|
|
|
return items, total
|
|
|
|
return await self._safe_execute(_get_flagged_content)
|
|
|
|
async def bulk_save_objects(self, objects: List) -> bool:
|
|
"""Bulk save objects to database"""
|
|
def _bulk_save():
|
|
for obj in objects:
|
|
db.session.add(obj)
|
|
db.session.commit()
|
|
return True
|
|
|
|
result = await self._safe_execute(_bulk_save)
|
|
return result is not None
|
|
|
|
async def delete_user_data_by_external_id(self, project_id: str, external_user_id: str) -> Dict[str, Any]:
|
|
"""
|
|
Delete all data for a user identified by external_user_id
|
|
|
|
Args:
|
|
project_id: The project ID to scope the deletion to
|
|
external_user_id: The external user ID to delete data for
|
|
|
|
Returns:
|
|
Dictionary containing deletion statistics and success status
|
|
"""
|
|
def _delete_user_data():
|
|
try:
|
|
# Find the APIUser record
|
|
api_user = APIUser.query.filter_by(
|
|
project_id=project_id,
|
|
external_user_id=external_user_id
|
|
).first()
|
|
|
|
if not api_user:
|
|
return {
|
|
'success': False,
|
|
'error': 'User not found',
|
|
'external_user_id': external_user_id,
|
|
'project_id': project_id
|
|
}
|
|
|
|
# Get counts before deletion for reporting
|
|
content_count = Content.query.filter_by(api_user_id=api_user.id).count()
|
|
|
|
# Get moderation results count
|
|
moderation_results_count = db.session.query(func.count(ModerationResult.id)).join(
|
|
Content, ModerationResult.content_id == Content.id
|
|
).filter(Content.api_user_id == api_user.id).scalar()
|
|
|
|
# Get all content IDs for this user
|
|
content_ids = [c.id for c in Content.query.filter_by(api_user_id=api_user.id).with_entities(Content.id).all()]
|
|
|
|
# Delete in correct order to avoid foreign key violations:
|
|
# 1. Delete moderation results first
|
|
if content_ids:
|
|
ModerationResult.query.filter(
|
|
ModerationResult.content_id.in_(content_ids)).delete(
|
|
synchronize_session=False)
|
|
|
|
# 2. Delete content
|
|
Content.query.filter_by(api_user_id=api_user.id).delete(synchronize_session=False)
|
|
|
|
# 3. Delete the APIUser record
|
|
db.session.delete(api_user)
|
|
|
|
# Commit the transaction
|
|
db.session.commit()
|
|
|
|
logger.info(
|
|
f"Deleted user data for external_user_id={external_user_id} "
|
|
f"in project_id={project_id}: "
|
|
f"{content_count} content items, {moderation_results_count} moderation results"
|
|
)
|
|
|
|
return {
|
|
'success': True,
|
|
'external_user_id': external_user_id,
|
|
'project_id': project_id,
|
|
'deleted_counts': {
|
|
'api_user': 1,
|
|
'content': content_count,
|
|
'moderation_results': moderation_results_count
|
|
}
|
|
}
|
|
|
|
except SQLAlchemyError as e:
|
|
db.session.rollback()
|
|
logger.error(f"Error deleting user data for {external_user_id}: {str(e)}")
|
|
return {
|
|
'success': False,
|
|
'error': str(e),
|
|
'external_user_id': external_user_id,
|
|
'project_id': project_id
|
|
}
|
|
|
|
return await self._safe_execute(_delete_user_data)
|
|
|
|
async def search_user_by_external_id(self, project_id: str, external_user_id: str) -> Optional[Dict[str, Any]]:
|
|
"""
|
|
Search for a user by external_user_id and return their data summary
|
|
|
|
Args:
|
|
project_id: The project ID to scope the search to
|
|
external_user_id: The external user ID to search for
|
|
|
|
Returns:
|
|
Dictionary containing user information and data counts, or None if not found
|
|
"""
|
|
def _search_user():
|
|
try:
|
|
# Find the APIUser record
|
|
api_user = APIUser.query.filter_by(
|
|
project_id=project_id,
|
|
external_user_id=external_user_id
|
|
).first()
|
|
|
|
if not api_user:
|
|
return None
|
|
|
|
# Get content count
|
|
content_count = Content.query.filter_by(api_user_id=api_user.id).count()
|
|
|
|
# Get moderation results count
|
|
moderation_results_count = db.session.query(func.count(ModerationResult.id)).join(
|
|
Content, ModerationResult.content_id == Content.id
|
|
).filter(Content.api_user_id == api_user.id).scalar()
|
|
|
|
# Get status breakdown
|
|
status_breakdown = db.session.query(
|
|
Content.status,
|
|
func.count(Content.id).label('count')
|
|
).filter(
|
|
Content.api_user_id == api_user.id
|
|
).group_by(Content.status).all()
|
|
|
|
return {
|
|
'api_user_id': api_user.id,
|
|
'external_user_id': api_user.external_user_id,
|
|
'project_id': api_user.project_id,
|
|
'first_seen': api_user.first_seen.isoformat() if api_user.first_seen else None,
|
|
'last_seen': api_user.last_seen.isoformat() if api_user.last_seen else None,
|
|
'data_counts': {
|
|
'content': content_count,
|
|
'moderation_results': moderation_results_count,
|
|
'status_breakdown': {status: count for status, count in status_breakdown}
|
|
}
|
|
}
|
|
|
|
except SQLAlchemyError as e:
|
|
logger.error(f"Error searching user by external_id {external_user_id}: {str(e)}")
|
|
return None
|
|
|
|
return await self._safe_execute(_search_user)
|
|
|
|
|
|
# Global async database service instance
|
|
db_service = DatabaseService()
|