169 lines
4.2 KiB
Python
169 lines
4.2 KiB
Python
from typing import Optional, Dict, List, Union, Any
|
|
from datetime import datetime
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy import or_
|
|
from pydantic import BaseModel, validator
|
|
|
|
class GenericValidator:
|
|
"""Validator class for Generic entity"""
|
|
|
|
@staticmethod
|
|
def validate_name(name: str) -> bool:
|
|
"""
|
|
Validate generic name format and length
|
|
|
|
Args:
|
|
name: Name to validate
|
|
|
|
Returns:
|
|
bool: True if valid, False otherwise
|
|
"""
|
|
if not name or len(name) < 2 or len(name) > 100:
|
|
return False
|
|
return bool(name.strip())
|
|
|
|
@staticmethod
|
|
def validate_status(status: str) -> bool:
|
|
"""
|
|
Validate generic status
|
|
|
|
Args:
|
|
status: Status to validate
|
|
|
|
Returns:
|
|
bool: True if valid, False otherwise
|
|
"""
|
|
valid_statuses = ['active', 'inactive', 'pending', 'archived']
|
|
return status.lower() in valid_statuses
|
|
|
|
|
|
def format_generic_response(generic: Any) -> Dict:
|
|
"""
|
|
Format generic entity for API response
|
|
|
|
Args:
|
|
generic: Generic entity to format
|
|
|
|
Returns:
|
|
Dict containing formatted generic data
|
|
"""
|
|
return {
|
|
'id': generic.id,
|
|
'name': generic.name,
|
|
'description': generic.description,
|
|
'status': generic.status,
|
|
'created_at': generic.created_at.isoformat() if generic.created_at else None,
|
|
'updated_at': generic.updated_at.isoformat() if generic.updated_at else None
|
|
}
|
|
|
|
|
|
def search_generics(
|
|
db: Session,
|
|
search_term: str,
|
|
status: Optional[str] = None,
|
|
limit: int = 10,
|
|
offset: int = 0
|
|
) -> List[Any]:
|
|
"""
|
|
Search generic entities by name or description
|
|
|
|
Args:
|
|
db: Database session
|
|
search_term: Term to search for
|
|
status: Optional status filter
|
|
limit: Max number of results
|
|
offset: Number of results to skip
|
|
|
|
Returns:
|
|
List of matching generic entities
|
|
"""
|
|
query = db.query(Generic)
|
|
|
|
if search_term:
|
|
query = query.filter(
|
|
or_(
|
|
Generic.name.ilike(f"%{search_term}%"),
|
|
Generic.description.ilike(f"%{search_term}%")
|
|
)
|
|
)
|
|
|
|
if status:
|
|
query = query.filter(Generic.status == status)
|
|
|
|
return query.offset(offset).limit(limit).all()
|
|
|
|
|
|
def bulk_update_status(
|
|
db: Session,
|
|
generic_ids: List[int],
|
|
new_status: str
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Bulk update status for multiple generic entities
|
|
|
|
Args:
|
|
db: Database session
|
|
generic_ids: List of generic IDs to update
|
|
new_status: New status to set
|
|
|
|
Returns:
|
|
Dict with update results
|
|
"""
|
|
if not GenericValidator.validate_status(new_status):
|
|
return {'error': 'Invalid status value'}
|
|
|
|
try:
|
|
updated = db.query(Generic).filter(
|
|
Generic.id.in_(generic_ids)
|
|
).update(
|
|
{Generic.status: new_status},
|
|
synchronize_session=False
|
|
)
|
|
db.commit()
|
|
return {
|
|
'success': True,
|
|
'updated_count': updated
|
|
}
|
|
except Exception as e:
|
|
db.rollback()
|
|
return {
|
|
'success': False,
|
|
'error': str(e)
|
|
}
|
|
|
|
|
|
def archive_old_generics(
|
|
db: Session,
|
|
days_threshold: int = 90
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Archive generic entities older than threshold
|
|
|
|
Args:
|
|
db: Database session
|
|
days_threshold: Days after which to archive
|
|
|
|
Returns:
|
|
Dict with archive results
|
|
"""
|
|
cutoff_date = datetime.utcnow() - timedelta(days=days_threshold)
|
|
|
|
try:
|
|
archived = db.query(Generic).filter(
|
|
Generic.created_at < cutoff_date,
|
|
Generic.status != 'archived'
|
|
).update(
|
|
{Generic.status: 'archived'},
|
|
synchronize_session=False
|
|
)
|
|
db.commit()
|
|
return {
|
|
'success': True,
|
|
'archived_count': archived
|
|
}
|
|
except Exception as e:
|
|
db.rollback()
|
|
return {
|
|
'success': False,
|
|
'error': str(e)
|
|
} |