33 lines
917 B
Python
33 lines
917 B
Python
"""
|
|
Advertools Task Loader Utility
|
|
Utility functions for loading due Advertools tasks from the database.
|
|
"""
|
|
|
|
from typing import List, Optional
|
|
from datetime import datetime
|
|
from sqlalchemy.orm import Session
|
|
from models.advertools_monitoring_models import AdvertoolsTask
|
|
|
|
def load_due_advertools_tasks(db: Session, user_id: Optional[str] = None) -> List[AdvertoolsTask]:
|
|
"""
|
|
Load Advertools tasks that are due for execution.
|
|
|
|
Args:
|
|
db: Database session
|
|
user_id: Optional user ID to filter tasks (for multi-tenant support)
|
|
|
|
Returns:
|
|
List of due AdvertoolsTask objects
|
|
"""
|
|
now = datetime.utcnow()
|
|
|
|
query = db.query(AdvertoolsTask).filter(
|
|
AdvertoolsTask.status == 'active',
|
|
AdvertoolsTask.next_execution <= now
|
|
)
|
|
|
|
if user_id:
|
|
query = query.filter(AdvertoolsTask.user_id == user_id)
|
|
|
|
return query.all()
|