starmen1265
3ad2576aed
- Test BaseProcessor cannot be instantiated directly - Test subclass must implement process() method - Test state_manager attribute is provided - Test STEP_NAME class attribute requirement Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
56 lines
1.6 KiB
Python
56 lines
1.6 KiB
Python
"""BaseProcessor - Abstract base class for all pipeline processors
|
|
|
|
Provides a unified interface that all processors (parse, download, asr, analyze)
|
|
must follow, enabling consistent state management and error handling.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from abc import ABC, abstractmethod
|
|
from typing import Optional, TYPE_CHECKING
|
|
|
|
from kb_pipeline.config import get_settings
|
|
from kb_pipeline.schemas import StepInput, StepOutput
|
|
|
|
if TYPE_CHECKING:
|
|
from kb_pipeline.state import WorkflowStateManager
|
|
|
|
|
|
class BaseProcessor(ABC):
|
|
"""Abstract base class for all pipeline processors.
|
|
|
|
All processors (ParseProcessor, DownloadProcessor, ASRProcessor, AnalyzeProcessor)
|
|
must inherit from this class and implement the process() method.
|
|
|
|
Attributes:
|
|
STEP_NAME: Class attribute identifying the processor step name.
|
|
settings: KBSettings instance from get_settings().
|
|
state_manager: Optional WorkflowStateManager for state recording.
|
|
"""
|
|
|
|
STEP_NAME: str
|
|
|
|
def __init__(self, state_manager: Optional[WorkflowStateManager] = None):
|
|
"""Initialize the processor.
|
|
|
|
Args:
|
|
state_manager: Optional WorkflowStateManager for recording step state.
|
|
"""
|
|
self.settings = get_settings()
|
|
self.state_manager = state_manager
|
|
|
|
@abstractmethod
|
|
async def process(self, input_data: StepInput) -> StepOutput:
|
|
"""Process the input data and return the result.
|
|
|
|
Args:
|
|
input_data: Input data for the processor step.
|
|
|
|
Returns:
|
|
StepOutput with processing results.
|
|
|
|
Raises:
|
|
Exception: If processing fails.
|
|
"""
|
|
pass
|