9e9c7ac270
Code quality: - Remove empty except blocks with proper logging - Create shared pagination utility function - Remove duplicate UUID validation code - Fix dead code in translation.py Performance: - Fix N+1 query in followup engine (use join instead of loop) - Add eager loading for customer health scores - Create database indexes for common query patterns: - customers: (user_id, status), (user_id, last_contact_at) - payment_transactions: (user_id, created_at) - followup_logs: (user_id, customer_id) - notifications: (user_id, is_read) Configuration: - Centralize magic numbers in config.py: - Payment prices - File upload limits - Rate limiting settings - Pagination defaults - Update auth.py to use centralized rate limiting config - Update customer/product imports to use centralized upload limits - Update import_service.py to use centralized MAX_ROWS
60 lines
1.6 KiB
Python
60 lines
1.6 KiB
Python
"""Shared utility functions"""
|
|
import uuid
|
|
from typing import Any, Optional
|
|
from sqlalchemy import select, func
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
|
|
def validate_uuid(value: str) -> str:
|
|
"""Validate UUID format and return the value"""
|
|
try:
|
|
uuid.UUID(value)
|
|
return value
|
|
except ValueError:
|
|
raise ValueError(f"Invalid UUID format: {value}")
|
|
|
|
|
|
def truncate_string(value: str, max_length: int = 100) -> str:
|
|
"""Truncate string to specified length"""
|
|
if len(value) <= max_length:
|
|
return value
|
|
return value[:max_length]
|
|
|
|
|
|
def sanitize_for_logging(value: str) -> str:
|
|
"""Sanitize string for logging (remove sensitive info)"""
|
|
# Remove common sensitive patterns
|
|
import re
|
|
value = re.sub(r'[^a-zA-Z0-9\s\-_.,:;!?\'"]', '', value)
|
|
return value[:200] # Limit length for log safety
|
|
|
|
|
|
def paginate_query(query, page: int = 1, size: int = 20) -> dict:
|
|
"""
|
|
Paginate a SQLAlchemy query and return results with metadata.
|
|
|
|
Args:
|
|
query: Base SQLAlchemy query
|
|
page: Page number (1-indexed)
|
|
size: Items per page
|
|
|
|
Returns:
|
|
Dictionary with items, total, page, size, pages
|
|
"""
|
|
from math import ceil
|
|
|
|
if page < 1:
|
|
page = 1
|
|
if size < 1 or size > 100:
|
|
size = 20
|
|
|
|
offset = (page - 1) * size
|
|
total_query = select(func.count()).select_from(query.subquery())
|
|
|
|
return {
|
|
"items": query.offset(offset).limit(size).all(),
|
|
"total": total,
|
|
"page": page,
|
|
"size": size,
|
|
"pages": ceil(total / size) if total > 0 else 0,
|
|
} |