"""Base class for data sources."""
from abc import ABC, abstractmethod
from datetime import date, datetime
from typing import Any
from daily_briefing.models import ActionItem, CalendarEvent, MeetingSummary, TravelInfo
class DataSource(ABC):
"""Abstract base class for data sources."""
def __init__(self, name: str, enabled: bool = True):
"""Initialize the data source.
Args:
name: Display name for the source
enabled: Whether the source is enabled
"""
self.name = name
self.enabled = enabled
self._cache: dict[str, Any] = {}
self._cache_time: datetime | None = None
@abstractmethod
async def get_events(
self, start_date: date, end_date: date
) -> list[CalendarEvent]:
"""Get calendar events for a date range.
Args:
start_date: Start of date range
end_date: End of date range
Returns:
List of calendar events
"""
pass
async def get_meetings(
self, start_date: date, end_date: date
) -> list[MeetingSummary]:
"""Get meeting summaries for a date range.
Default implementation returns empty list.
Args:
start_date: Start of date range
end_date: End of date range
Returns:
List of meeting summaries
"""
return []
async def get_action_items(self) -> list[ActionItem]:
"""Get pending action items.
Default implementation returns empty list.
Returns:
List of action items
"""
return []
async def get_travel_info(
self, start_date: date, end_date: date
) -> list[TravelInfo]:
"""Get travel information for a date range.
Default implementation returns empty list.
Args:
start_date: Start of date range
end_date: End of date range
Returns:
List of travel info
"""
return []
@abstractmethod
async def health_check(self) -> dict[str, Any]:
"""Check if the data source is healthy.
Returns:
Health status dictionary
"""
pass
