_watchers

_watchers

Polling watchers that detect changes and produce events.

BaseWatcher

BaseWatcher(
    api: CocApi,
    state: PollingState,
    queue: Queue[Event],
    interval: float,
)

Base class for all watchers.

Subclasses implement _poll_once() which fetches data, diffs it, and returns a list of events.

Initialize the watcher.

PARAMETER	DESCRIPTION
api	CocApi instance in async mode. TYPE: CocApi
state	Shared polling state for snapshot storage. TYPE: PollingState
queue	Event queue to push detected changes to. TYPE: Queue[Event]
interval	Seconds between poll cycles. TYPE: float

Source code in cocapi/events/_watchers.py

def __init__(
    self,
    api: CocApi,
    state: PollingState,
    queue: asyncio.Queue[Event],
    interval: float,
) -> None:
    """Initialize the watcher.

    Args:
        api: CocApi instance in async mode.
        state: Shared polling state for snapshot storage.
        queue: Event queue to push detected changes to.
        interval: Seconds between poll cycles.
    """
    self._api = api
    self._state = state
    self._queue = queue
    self._interval = interval
    self._task: asyncio.Task[None] | None = None
    self._running = False

interval property writable

interval: float

Current poll interval in seconds.

start

start() -> None

Start the polling loop as an asyncio task.

Source code in cocapi/events/_watchers.py

def start(self) -> None:
    """Start the polling loop as an asyncio task."""
    if self._task is not None:
        return
    self._running = True
    self._task = asyncio.create_task(self._loop())

stop async

stop() -> None

Stop the polling loop and cancel the background task.

Source code in cocapi/events/_watchers.py

async def stop(self) -> None:
    """Stop the polling loop and cancel the background task."""
    self._running = False
    if self._task is not None:
        self._task.cancel()
        try:
            await self._task
        except asyncio.CancelledError:
            pass  # Expected: we just cancelled this task
        self._task = None

ClanWatcher

ClanWatcher(
    api: CocApi,
    state: PollingState,
    queue: Queue[Event],
    clan_tags: list[str],
    interval: float = 60.0,
    track_members: bool = True,
)

Bases: BaseWatcher

Polls clan data and member lists.

Initialize the clan watcher.

PARAMETER	DESCRIPTION
api	CocApi instance in async mode. TYPE: CocApi
state	Shared polling state. TYPE: PollingState
queue	Event queue for detected changes. TYPE: Queue[Event]
clan_tags	Clan tags to poll. TYPE: list[str]
interval	Seconds between polls (default 60). TYPE: float DEFAULT: 60.0
track_members	Whether to track member join/leave/update events. TYPE: bool DEFAULT: True

Source code in cocapi/events/_watchers.py

def __init__(
    self,
    api: CocApi,
    state: PollingState,
    queue: asyncio.Queue[Event],
    clan_tags: list[str],
    interval: float = 60.0,
    track_members: bool = True,
) -> None:
    """Initialize the clan watcher.

    Args:
        api: CocApi instance in async mode.
        state: Shared polling state.
        queue: Event queue for detected changes.
        clan_tags: Clan tags to poll.
        interval: Seconds between polls (default 60).
        track_members: Whether to track member join/leave/update events.
    """
    super().__init__(api, state, queue, interval)
    self._clan_tags = list(clan_tags)
    self._track_members = track_members

WarWatcher

WarWatcher(
    api: CocApi,
    state: PollingState,
    queue: Queue[Event],
    clan_tags: list[str],
    interval: float = 30.0,
)

Bases: BaseWatcher

Polls current war and tracks state transitions and new attacks.

Initialize the war watcher.

PARAMETER	DESCRIPTION
api	CocApi instance in async mode. TYPE: CocApi
state	Shared polling state. TYPE: PollingState
queue	Event queue for detected changes. TYPE: Queue[Event]
clan_tags	Clan tags to watch for war updates. TYPE: list[str]
interval	Seconds between polls (default 30). TYPE: float DEFAULT: 30.0

Source code in cocapi/events/_watchers.py

def __init__(
    self,
    api: CocApi,
    state: PollingState,
    queue: asyncio.Queue[Event],
    clan_tags: list[str],
    interval: float = 30.0,
) -> None:
    """Initialize the war watcher.

    Args:
        api: CocApi instance in async mode.
        state: Shared polling state.
        queue: Event queue for detected changes.
        clan_tags: Clan tags to watch for war updates.
        interval: Seconds between polls (default 30).
    """
    super().__init__(api, state, queue, interval)
    self._clan_tags = list(clan_tags)

PlayerWatcher

PlayerWatcher(
    api: CocApi,
    state: PollingState,
    queue: Queue[Event],
    player_tags: list[str],
    interval: float = 120.0,
    include_fields: frozenset[str] | None = None,
    exclude_fields: frozenset[str] | None = None,
)

Bases: BaseWatcher

Polls player data and detects field changes.

Emits granular events for upgrades (troops, spells, heroes, equipment, townhall, builder hall) as well as the generic PLAYER_UPDATED event for any other top-level field change.

Initialize the player watcher.

PARAMETER	DESCRIPTION
api	CocApi instance in async mode. TYPE: CocApi
state	Shared polling state. TYPE: PollingState
queue	Event queue for detected changes. TYPE: Queue[Event]
player_tags	Player tags to poll. TYPE: list[str]
interval	Seconds between polls (default 120). TYPE: float DEFAULT: 120.0
include_fields	Only report changes to these fields. TYPE: frozenset[str] | None DEFAULT: None
exclude_fields	Ignore changes to these fields. TYPE: frozenset[str] | None DEFAULT: None

Source code in cocapi/events/_watchers.py

def __init__(
    self,
    api: CocApi,
    state: PollingState,
    queue: asyncio.Queue[Event],
    player_tags: list[str],
    interval: float = 120.0,
    include_fields: frozenset[str] | None = None,
    exclude_fields: frozenset[str] | None = None,
) -> None:
    """Initialize the player watcher.

    Args:
        api: CocApi instance in async mode.
        state: Shared polling state.
        queue: Event queue for detected changes.
        player_tags: Player tags to poll.
        interval: Seconds between polls (default 120).
        include_fields: Only report changes to these fields.
        exclude_fields: Ignore changes to these fields.
    """
    super().__init__(api, state, queue, interval)
    self._player_tags = list(player_tags)
    self._include_fields = include_fields
    self._exclude_fields = exclude_fields

MaintenanceWatcher

MaintenanceWatcher(
    api: CocApi,
    state: PollingState,
    queue: Queue[Event],
    interval: float = 30.0,
    probe_tag: str = "#JY9J2Y99",
)

Bases: BaseWatcher

Detects API maintenance windows by polling a known endpoint.

Emits MAINTENANCE_START when the API starts returning errors (HTTP 503 or connection failures) and MAINTENANCE_END when it recovers. Uses a configurable probe tag (defaults to a well-known player #JY9J2Y99).

Initialize the maintenance watcher.

PARAMETER	DESCRIPTION
api	CocApi instance in async mode. TYPE: CocApi
state	Shared polling state. TYPE: PollingState
queue	Event queue for detected changes. TYPE: Queue[Event]
interval	Seconds between probes (default 30). TYPE: float DEFAULT: 30.0
probe_tag	Player tag to probe for availability. TYPE: str DEFAULT: '#JY9J2Y99'

Source code in cocapi/events/_watchers.py

def __init__(
    self,
    api: CocApi,
    state: PollingState,
    queue: asyncio.Queue[Event],
    interval: float = 30.0,
    probe_tag: str = "#JY9J2Y99",
) -> None:
    """Initialize the maintenance watcher.

    Args:
        api: CocApi instance in async mode.
        state: Shared polling state.
        queue: Event queue for detected changes.
        interval: Seconds between probes (default 30).
        probe_tag: Player tag to probe for availability.
    """
    super().__init__(api, state, queue, interval)
    self._probe_tag = probe_tag
    self._in_maintenance = False
    self._maintenance_start_time: float | None = None