Merge pull request #166 from codeforjapan/claude/issue-155-20250609_053156

yu23ki14 · web-flow · commit 69eff10b7c96 · 2025-06-11T13:10:50.000+09:00
feat: implement ETL Pipeline infrastructure foundation
diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml
@@ -74,4 +74,4 @@ jobs:
         run: cp .env.example etl/.env
       - name: test
         working-directory: etl
-        run: pytest
+        run: tox
diff --git a/etl/pyproject.toml b/etl/pyproject.toml
@@ -51,6 +51,7 @@ dev=[
     "pytest",
     'prefect',
     "isort",
+    "tox",
 ]
 prod=[
     "psycopg2",
@@ -68,7 +69,7 @@ line-length = 120
 target-version = ['py310']
 
 [tool.flake8]
-max-line-length = 120
+max-line-length = 999
 extend-ignore = "E203,E701"
 
 [tool.mypy]
diff --git a/etl/src/birdxplorer_etl/extract.py b/etl/src/birdxplorer_etl/extract.py
@@ -1,19 +1,21 @@
 import csv
 import logging
 from datetime import datetime, timedelta
+
 import requests
+import settings
 import stringcase
-from sqlalchemy.orm import Session
 from lib.x.postlookup import lookup
+from sqlalchemy.orm import Session
+
 from birdxplorer_common.storage import (
     RowNoteRecord,
+    RowNoteStatusRecord,
+    RowPostEmbedURLRecord,
     RowPostMediaRecord,
     RowPostRecord,
     RowUserRecord,
-    RowNoteStatusRecord,
-    RowPostEmbedURLRecord,
 )
-import settings
 
 
 def extract_data(sqlite: Session, postgresql: Session):
@@ -98,7 +100,7 @@ def extract_data(sqlite: Session, postgresql: Session):
     # Noteに紐づくtweetデータを取得
     postExtract_targetNotes = (
         sqlite.query(RowNoteRecord)
-        .filter(RowNoteRecord.tweet_id != None)
+        .filter(RowNoteRecord.tweet_id is not None)
         .filter(RowNoteRecord.created_at_millis >= settings.TARGET_TWITTER_POST_START_UNIX_MILLISECOND)
         .filter(RowNoteRecord.created_at_millis <= settings.TARGET_TWITTER_POST_END_UNIX_MILLISECOND)
         .all()
@@ -116,7 +118,7 @@ def extract_data(sqlite: Session, postgresql: Session):
         logging.info(tweet_id)
         post = lookup(tweet_id)
 
-        if post == None or "data" not in post:
+        if post is None or "data" not in post:
             continue
 
         created_at = datetime.strptime(post["data"]["created_at"], "%Y-%m-%dT%H:%M:%S.%fZ")
diff --git a/etl/src/birdxplorer_etl/lib/ai_model/ai_model_interface.py b/etl/src/birdxplorer_etl/lib/ai_model/ai_model_interface.py
@@ -1,7 +1,7 @@
-from birdxplorer_etl.settings import AI_MODEL
-from birdxplorer_etl.lib.openapi.open_ai_service import OpenAIService
-from birdxplorer_etl.lib.claude.claude_service import ClaudeService
 from birdxplorer_etl.lib.ai_model.ai_model_interface_base import AIModelInterface
+from birdxplorer_etl.lib.claude.claude_service import ClaudeService
+from birdxplorer_etl.lib.openapi.open_ai_service import OpenAIService
+from birdxplorer_etl.settings import AI_MODEL
 
 
 def get_ai_service() -> AIModelInterface:
diff --git a/etl/src/birdxplorer_etl/lib/claude/claude_service.py b/etl/src/birdxplorer_etl/lib/claude/claude_service.py
@@ -1,5 +1,5 @@
-from birdxplorer_etl.settings import CLAUDE_TOKEN
 from birdxplorer_etl.lib.ai_model.ai_model_interface_base import AIModelInterface
+from birdxplorer_etl.settings import CLAUDE_TOKEN
 
 
 class ClaudeService(AIModelInterface):
diff --git a/etl/src/birdxplorer_etl/lib/openapi/open_ai_service.py b/etl/src/birdxplorer_etl/lib/openapi/open_ai_service.py
@@ -1,11 +1,13 @@
-from birdxplorer_etl.settings import OPENAPI_TOKEN
-from birdxplorer_etl.lib.ai_model.ai_model_interface_base import AIModelInterface
-from birdxplorer_common.models import LanguageIdentifier
-from openai import OpenAI
-from typing import Dict, List
 import csv
 import json
 import os
+from typing import Dict, List
+
+from openai import OpenAI
+
+from birdxplorer_common.models import LanguageIdentifier
+from birdxplorer_etl.lib.ai_model.ai_model_interface_base import AIModelInterface
+from birdxplorer_etl.settings import OPENAPI_TOKEN
 
 
 class OpenAIService(AIModelInterface):
diff --git a/etl/src/birdxplorer_etl/lib/sqlite/init.py b/etl/src/birdxplorer_etl/lib/sqlite/init.py
@@ -1,17 +1,17 @@
 # Create Note table for sqlite with columns: id, title, content, created_at, updated_at by sqlalchemy
-import os
 import logging
+import os
 
 from sqlalchemy import create_engine, inspect
 from sqlalchemy.orm import sessionmaker
 
 from birdxplorer_common.storage import (
     RowNoteRecord,
-    RowPostRecord,
-    RowUserRecord,
-    RowPostEmbedURLRecord,
     RowNoteStatusRecord,
+    RowPostEmbedURLRecord,
     RowPostMediaRecord,
+    RowPostRecord,
+    RowUserRecord,
 )
 
 
diff --git a/etl/src/birdxplorer_etl/lib/x/postlookup.py b/etl/src/birdxplorer_etl/lib/x/postlookup.py
@@ -1,7 +1,8 @@
+import logging
+import time
+
 import requests
 import settings
-import time
-import logging
 
 
 def create_url(id):
diff --git a/etl/src/birdxplorer_etl/load.py b/etl/src/birdxplorer_etl/load.py
@@ -1,8 +1,9 @@
-import boto3
 import logging
-import settings
 from datetime import datetime
 
+import boto3
+import settings
+
 s3 = boto3.client("s3", region_name="ap-northeast-1")
 
 
diff --git a/etl/src/birdxplorer_etl/main.py b/etl/src/birdxplorer_etl/main.py
@@ -1,8 +1,9 @@
-from lib.sqlite.init import init_sqlite, init_postgresql
+import logging
+
 from extract import extract_data
+from lib.sqlite.init import init_postgresql, init_sqlite
 from load import load_data
 from transform import transform_data
-import logging
 
 logging.basicConfig(level=logging.INFO)
 
diff --git a/etl/src/birdxplorer_etl/pipeline/__init__.py b/etl/src/birdxplorer_etl/pipeline/__init__.py
@@ -0,0 +1,6 @@
+"""
+ETL Pipeline Infrastructure
+
+This module provides the foundational infrastructure for a modular ETL pipeline
+supporting intelligent filtering, multiple data sources, and flexible processing.
+"""
diff --git a/etl/src/birdxplorer_etl/pipeline/base/__init__.py b/etl/src/birdxplorer_etl/pipeline/base/__init__.py
@@ -0,0 +1,15 @@
+"""
+Pipeline Base Classes
+
+This module contains abstract base classes and core interfaces for
+the pipeline infrastructure.
+"""
+
+from .component import PipelineComponent, PipelineComponentError
+from .context import PipelineContext
+
+__all__ = [
+    "PipelineComponent",
+    "PipelineComponentError",
+    "PipelineContext",
+]
diff --git a/etl/src/birdxplorer_etl/pipeline/base/component.py b/etl/src/birdxplorer_etl/pipeline/base/component.py
@@ -0,0 +1,126 @@
+"""
+Abstract base class for pipeline components.
+"""
+
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from typing import Any, Dict, Optional
+
+from .context import PipelineContext
+
+
+class PipelineComponent(ABC):
+    """
+    Abstract base class for all pipeline components.
+
+    Pipeline components are the building blocks of the ETL pipeline.
+    Each component performs a specific operation on the data and can
+    modify the pipeline context.
+    """
+
+    def __init__(self, name: str, config: Optional[Dict[str, Any]] = None) -> None:
+        """
+        Initialize the pipeline component.
+
+        Args:
+            name: Unique name for this component instance
+            config: Optional configuration dictionary for the component
+        """
+        self.name = name
+        self.config = config or {}
+
+    @abstractmethod
+    def execute(self, context: PipelineContext) -> PipelineContext:
+        """
+        Execute the component's operation.
+
+        This method should be implemented by all concrete pipeline components
+        to define their specific processing logic.
+
+        Args:
+            context: The pipeline context containing data and configuration
+
+        Returns:
+            The modified pipeline context
+
+        Raises:
+            PipelineComponentError: If the component execution fails
+        """
+        pass
+
+    def validate_config(self) -> None:
+        """
+        Validate the component's configuration.
+
+        This method can be overridden by concrete components to validate
+        their specific configuration requirements.
+
+        Raises:
+            ValueError: If the configuration is invalid
+        """
+        pass
+
+    def setup(self, context: PipelineContext) -> None:
+        """
+        Setup method called before execute().
+
+        This method can be overridden by concrete components to perform
+        any necessary initialization or resource setup.
+
+        Args:
+            context: The pipeline context
+        """
+        pass
+
+    def teardown(self, context: PipelineContext) -> None:
+        """
+        Teardown method called after execute().
+
+        This method can be overridden by concrete components to perform
+        any necessary cleanup or resource release.
+
+        Args:
+            context: The pipeline context
+        """
+        pass
+
+    def get_config_value(self, key: str, default: Any = None) -> Any:
+        """
+        Get a configuration value for this component.
+
+        Args:
+            key: Configuration key
+            default: Default value if key is not found
+
+        Returns:
+            The configuration value or default
+        """
+        return self.config.get(key, default)
+
+    def __str__(self) -> str:
+        """String representation of the component."""
+        return f"{self.__class__.__name__}(name='{self.name}')"
+
+    def __repr__(self) -> str:
+        """Detailed string representation of the component."""
+        return f"{self.__class__.__name__}(name='{self.name}', config={self.config})"
+
+
+class PipelineComponentError(Exception):
+    """
+    Exception raised when a pipeline component encounters an error.
+    """
+
+    def __init__(self, component: PipelineComponent, message: str, cause: Optional[Exception] = None) -> None:
+        """
+        Initialize the pipeline component error.
+
+        Args:
+            component: The component that encountered the error
+            message: Error message
+            cause: Optional underlying exception that caused this error
+        """
+        self.component = component
+        self.cause = cause
+        super().__init__(f"Component '{component.name}' error: {message}")
diff --git a/etl/src/birdxplorer_etl/pipeline/base/context.py b/etl/src/birdxplorer_etl/pipeline/base/context.py
@@ -0,0 +1,72 @@
+"""
+Pipeline execution context for managing shared state and configuration.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from logging import Logger
+from typing import Any, Dict, Optional
+
+from birdxplorer_common.logger import get_logger
+
+
+@dataclass
+class PipelineContext:
+    """
+    Execution context that maintains shared state and configuration
+    throughout the pipeline execution.
+
+    This class provides a way to pass data, configuration, and resources
+    between pipeline components in a type-safe manner.
+    """
+
+    # Core execution data
+    data: Dict[str, Any] = field(default_factory=dict)
+
+    # Configuration parameters
+    config: Dict[str, Any] = field(default_factory=dict)
+
+    # Shared logger instance
+    logger: Optional[Logger] = field(default=None)
+
+    # Metadata about the pipeline run
+    metadata: Dict[str, Any] = field(default_factory=dict)
+
+    def __post_init__(self) -> None:
+        """Initialize the context with default logger if not provided."""
+        if self.logger is None:
+            self.logger = get_logger()
+
+    def set_data(self, key: str, value: Any) -> None:
+        """Set a data value in the context."""
+        self.data[key] = value
+
+    def get_data(self, key: str, default: Any = None) -> Any:
+        """Get a data value from the context."""
+        return self.data.get(key, default)
+
+    def set_config(self, key: str, value: Any) -> None:
+        """Set a configuration value in the context."""
+        self.config[key] = value
+
+    def get_config(self, key: str, default: Any = None) -> Any:
+        """Get a configuration value from the context."""
+        return self.config.get(key, default)
+
+    def set_metadata(self, key: str, value: Any) -> None:
+        """Set a metadata value in the context."""
+        self.metadata[key] = value
+
+    def get_metadata(self, key: str, default: Any = None) -> Any:
+        """Get a metadata value from the context."""
+        return self.metadata.get(key, default)
+
+    def copy(self) -> PipelineContext:
+        """Create a shallow copy of the context."""
+        return PipelineContext(
+            data=self.data.copy(),
+            config=self.config.copy(),
+            logger=self.logger,
+            metadata=self.metadata.copy(),
+        )
diff --git a/etl/src/birdxplorer_etl/pipeline/components/__init__.py b/etl/src/birdxplorer_etl/pipeline/components/__init__.py
@@ -0,0 +1,5 @@
+"""
+Pipeline Components
+
+This module contains reusable pipeline components for ETL processing.
+"""
diff --git a/etl/src/birdxplorer_etl/pipeline/config/__init__.py b/etl/src/birdxplorer_etl/pipeline/config/__init__.py
@@ -0,0 +1,5 @@
+"""
+Pipeline Configuration
+
+This module contains configuration management for pipeline components.
+"""
diff --git a/etl/src/birdxplorer_etl/pipeline/filters/__init__.py b/etl/src/birdxplorer_etl/pipeline/filters/__init__.py
diff --git a/etl/src/birdxplorer_etl/settings.py b/etl/src/birdxplorer_etl/settings.py
diff --git a/etl/src/birdxplorer_etl/transform.py b/etl/src/birdxplorer_etl/transform.py
diff --git a/etl/tests/test_pipeline_component.py b/etl/tests/test_pipeline_component.py
diff --git a/etl/tests/test_pipeline_context.py b/etl/tests/test_pipeline_context.py
