Pipeline Overview

@dataclass
class PipelineData:
    """In-memory state passed between pipeline steps. Not persisted to database."""
    
    # Input data (set by Celery task from API request)
    task_id: str
    user_id: str
    email_template: str
    recipient_name: str
    recipient_interest: str
    
    # Step 1 outputs (TemplateParser)
    search_terms: List[str] = field(default_factory=list)
    template_type: TemplateType | None = None
    
    # Step 2 outputs (WebScraper)
    scraped_content: str = ""
    scraped_urls: List[str] = field(default_factory=list)
    
    # Step 3 outputs (ArxivEnricher)
    arxiv_papers: List[Dict[str, Any]] = field(default_factory=list)
    
    # Step 4 outputs (EmailComposer)
    final_email: str = ""
    
    # Metadata for final DB write
    metadata: Dict[str, Any] = field(default_factory=dict)
    
    # Transient data (logged to Logfire, not persisted)
    step_timings: Dict[str, float] = field(default_factory=dict)
    errors: List[str] = field(default_factory=list)

async def execute(
    self,
    pipeline_data: PipelineData,
    progress_callback: Optional[Callable] = None
) -> StepResult:
    """Execute step with full observability."""
    start_time = time.perf_counter()
    
    # Create Logfire span for this step
    with logfire.span(
        f"pipeline.{self.step_name}",
        task_id=pipeline_data.task_id,
        step=self.step_name
    ):
        try:
            # Execute step-specific logic
            result = await self._execute_step(pipeline_data)
            
            # Record timing
            duration = time.perf_counter() - start_time
            pipeline_data.add_timing(self.step_name, duration)
            
            logfire.info(
                f"{self.step_name} completed",
                task_id=pipeline_data.task_id,
                duration=duration
            )
            return result
        except Exception as e:
            logfire.error(
                f"{self.step_name} failed",
                task_id=pipeline_data.task_id,
                error=str(e),
                exc_info=True
            )
            raise StepExecutionError(self.step_name, e)