timka123
77b26757bf
- Refactored TimerPage component to improve layout and user experience. - Updated theme configuration to utilize design tokens for better maintainability. - Introduced new tokens for colors and typography to align with design principles. - Simplified state management and improved timer functionality. - Enhanced accessibility and responsiveness of UI elements.
111 lines
3.8 KiB
Python
111 lines
3.8 KiB
Python
"""Builds the configured sources and pulls them all, independently.
|
|
|
|
Independence is the whole point: an expired laundry cookie or a Jira outage
|
|
costs you that one source and shows up as a warning beside it, never as an
|
|
empty task list.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from typing import Any, Callable
|
|
|
|
from app.settings_schema import AppSettings
|
|
from app.sources.base import SourceError, SourceResult, TodoItem
|
|
from app.sources.google_tasks import GoogleTasksSource
|
|
from app.sources.jira import JiraSource
|
|
from app.sources.laundry import LaundrySource
|
|
from app.sources.todoist import TodoistSource
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
SOURCE_LABELS = {
|
|
"jira": "Jira",
|
|
"todoist": "Todoist",
|
|
"google_tasks": "Google Tasks",
|
|
"laundry": "Laundry",
|
|
"local": "My list",
|
|
}
|
|
|
|
|
|
def build_sources(settings: AppSettings) -> dict[str, Any]:
|
|
"""Instantiate every enabled source. A misconfigured one is skipped, loudly."""
|
|
timeout = settings.sources.request_timeout_seconds
|
|
builders: list[tuple[str, Callable[[], Any]]] = [
|
|
("jira", lambda: JiraSource(settings.sources.jira, timeout=timeout)),
|
|
("todoist", lambda: TodoistSource(settings.sources.todoist, timeout=timeout)),
|
|
(
|
|
"google_tasks",
|
|
lambda: GoogleTasksSource(settings.sources.google_tasks, timeout=timeout),
|
|
),
|
|
("laundry", lambda: LaundrySource(settings.sources.laundry, timeout=timeout)),
|
|
]
|
|
enabled = {
|
|
"jira": settings.sources.jira.enabled,
|
|
"todoist": settings.sources.todoist.enabled,
|
|
"google_tasks": settings.sources.google_tasks.enabled,
|
|
"laundry": settings.sources.laundry.enabled,
|
|
}
|
|
|
|
sources: dict[str, Any] = {}
|
|
for name, build in builders:
|
|
if not enabled[name]:
|
|
continue
|
|
try:
|
|
sources[name] = build()
|
|
except SourceError as exc:
|
|
log.warning("source %s is enabled but unusable: %s", name, exc)
|
|
return sources
|
|
|
|
|
|
def _close(source: Any) -> None:
|
|
closer = getattr(source, "close", None)
|
|
if callable(closer):
|
|
try:
|
|
closer()
|
|
except Exception: # pragma: no cover
|
|
log.debug("closing %s failed", type(source).__name__, exc_info=True)
|
|
|
|
|
|
def fetch_all(settings: AppSettings) -> list[SourceResult]:
|
|
"""Fetch every enabled source in parallel; never raises."""
|
|
sources = build_sources(settings)
|
|
if not sources:
|
|
return []
|
|
|
|
def run(name: str, source: Any) -> SourceResult:
|
|
try:
|
|
items: list[TodoItem] = source.fetch()
|
|
return SourceResult(source=name, items=items)
|
|
except SourceError as exc:
|
|
return SourceResult.failed(name, str(exc))
|
|
except Exception as exc: # pragma: no cover - defensive
|
|
log.exception("source %s blew up", name)
|
|
return SourceResult.failed(name, f"unexpected error: {exc}")
|
|
|
|
try:
|
|
with ThreadPoolExecutor(max_workers=len(sources)) as pool:
|
|
futures = [pool.submit(run, name, src) for name, src in sources.items()]
|
|
return [future.result() for future in futures]
|
|
finally:
|
|
for source in sources.values():
|
|
_close(source)
|
|
|
|
|
|
def check_source(settings: AppSettings, name: str) -> tuple[bool, str]:
|
|
"""Verify one source's credentials — what the settings page's Test button calls."""
|
|
sources = build_sources(settings)
|
|
source = sources.get(name)
|
|
if source is None:
|
|
return False, f"{SOURCE_LABELS.get(name, name)} is not enabled or not configured."
|
|
try:
|
|
return True, source.check()
|
|
except SourceError as exc:
|
|
return False, str(exc)
|
|
except Exception as exc: # pragma: no cover
|
|
return False, f"unexpected error: {exc}"
|
|
finally:
|
|
for built in sources.values():
|
|
_close(built)
|