60 lines
1.5 KiB
Python
60 lines
1.5 KiB
Python
"""Base executor class for all task types."""
|
|
|
|
from abc import ABC, abstractmethod
|
|
from dataclasses import dataclass
|
|
from typing import Any, Optional
|
|
|
|
from app.utils.logger import get_logger
|
|
|
|
|
|
@dataclass
|
|
class ExecutionResult:
|
|
"""Result of task execution."""
|
|
|
|
success: bool
|
|
data: dict[str, Any]
|
|
error: Optional[str] = None
|
|
duration_ms: Optional[float] = None
|
|
|
|
|
|
class BaseExecutor(ABC):
|
|
"""Abstract base class for task executors.
|
|
|
|
All executors must implement the execute() method.
|
|
"""
|
|
|
|
def __init__(self):
|
|
self.logger = get_logger(self.__class__.__name__)
|
|
|
|
@property
|
|
@abstractmethod
|
|
def task_type(self) -> str:
|
|
"""Return the task type this executor handles."""
|
|
pass
|
|
|
|
@abstractmethod
|
|
async def execute(self, payload: dict[str, Any]) -> ExecutionResult:
|
|
"""Execute the task with the given payload.
|
|
|
|
Args:
|
|
payload: Task-specific payload data
|
|
|
|
Returns:
|
|
ExecutionResult with success status and result data
|
|
"""
|
|
pass
|
|
|
|
def validate_payload(self, payload: dict[str, Any], required_fields: list[str]) -> None:
|
|
"""Validate that required fields are present in payload.
|
|
|
|
Args:
|
|
payload: Task payload
|
|
required_fields: List of required field names
|
|
|
|
Raises:
|
|
ValueError: If a required field is missing
|
|
"""
|
|
missing = [f for f in required_fields if f not in payload]
|
|
if missing:
|
|
raise ValueError(f"Missing required fields: {', '.join(missing)}")
|