35 lines
1.1 KiB
Python
35 lines
1.1 KiB
Python
|
"""Provider-neutral, read-only busy-time interface."""
|
||
|
|
from typing import Protocol
|
||
|
|
|
||
|
|
from app.core.database import execute_query
|
||
|
|
|
||
|
|
|
||
|
|
class BusyTimeProvider(Protocol):
|
||
|
|
def get_busy(self, user_ids: list[int], starts_at, ends_at) -> list[dict]: ...
|
||
|
|
def statuses(self, user_ids: list[int]) -> list[dict]: ...
|
||
|
|
|
||
|
|
|
||
|
|
class DatabaseBusyTimeProvider:
|
||
|
|
"""Cache adapter used by Outlook/Google sync jobs without exposing meeting data."""
|
||
|
|
|
||
|
|
def get_busy(self, user_ids, starts_at, ends_at):
|
||
|
|
if not user_ids:
|
||
|
|
return []
|
||
|
|
return execute_query(
|
||
|
|
"""SELECT user_id, starts_at, ends_at, provider FROM planner_external_busy
|
||
|
|
WHERE user_id = ANY(%s) AND starts_at < %s AND ends_at > %s""",
|
||
|
|
(user_ids, ends_at, starts_at),
|
||
|
|
) or []
|
||
|
|
|
||
|
|
def statuses(self, user_ids):
|
||
|
|
if not user_ids:
|
||
|
|
return []
|
||
|
|
return execute_query(
|
||
|
|
"""SELECT user_id, provider, status, last_success_at, last_error_at
|
||
|
|
FROM planner_integrations WHERE user_id = ANY(%s)""",
|
||
|
|
(user_ids,),
|
||
|
|
) or []
|
||
|
|
|
||
|
|
|
||
|
|
busy_time_provider: BusyTimeProvider = DatabaseBusyTimeProvider()
|