"""Base adapter class for external task management systems.
Defines the interface that all task adapters must implement.
"""
from __future__ import annotations
from abc import ABC, abstractmethod
from typing import TYPE_CHECKING, Any
if TYPE_CHECKING:
from mcp_task_aggregator.models import Todo, TodoStatus
class TaskAdapter(ABC):
"""Abstract base class for external task system adapters.
All adapters (Jira, GitHub, Linear) must implement this interface
to provide consistent task synchronization.
"""
@abstractmethod
def fetch_tasks(self, **kwargs: Any) -> list[dict[str, Any]]:
"""Fetch tasks from the external system.
Args:
**kwargs: Adapter-specific parameters (e.g., project, filters).
Returns:
List of raw task data from the external system.
"""
...
@abstractmethod
def normalize_task(self, raw_task: dict[str, Any]) -> Todo:
"""Normalize a raw external task to the internal Todo model.
Args:
raw_task: Raw task data from the external system.
Returns:
Normalized Todo instance.
"""
...
@abstractmethod
def map_status(self, external_status: str) -> TodoStatus:
"""Map external system status to internal TodoStatus.
Args:
external_status: Status string from the external system.
Returns:
Corresponding TodoStatus enum value.
"""
...
def push_update(self, task: Todo, fields: dict[str, Any]) -> bool: # noqa: ARG002
"""Push task updates to the external system.
Args:
task: Task to update.
fields: Fields to update.
Returns:
True if update succeeded, False otherwise.
Note:
This is optional for Phase 1 (read-only). Default implementation
returns False to indicate write-back is not supported.
"""
return False
def generate_sync_hash(self, raw_task: dict[str, Any]) -> str:
"""Generate a hash for change detection.
Args:
raw_task: Raw task data from the external system.
Returns:
Hash string for detecting changes.
"""
import hashlib
import json
# Create deterministic JSON representation
content = json.dumps(raw_task, sort_keys=True, default=str)
return hashlib.sha256(content.encode()).hexdigest()[:16]
