events

events

cocapi.events -- Event polling system for real-time Clash of Clans monitoring.

Async-only. Requires CocApi in async mode.

Usage::

from cocapi import CocApi, ApiConfig
from cocapi.events import EventStream, EventType

async with CocApi("token", config=ApiConfig(enable_caching=False)) as api:
    stream = EventStream(api)
    stream.watch_clans(["#ABC"], interval=60)
    stream.watch_wars(["#ABC"], interval=30)

    async with stream:
        async for event in stream:
            print(event.event_type, event.tag, event.changes)

PollingState

PollingState()

In-memory state for polled resources with optional JSON persistence.

Source code in cocapi/events/_state.py

def __init__(self) -> None:
    self._clan_snapshots: dict[str, dict[str, Any]] = {}
    self._member_snapshots: dict[str, list[dict[str, Any]]] = {}
    self._war_snapshots: dict[str, dict[str, Any]] = {}
    self._war_fsms: dict[str, WarStateMachine] = {}
    self._player_snapshots: dict[str, dict[str, Any]] = {}
    self._last_poll_times: dict[str, float] = {}

get_clan

get_clan(tag: str) -> dict[str, Any] | None

Return the last-polled clan snapshot, or None if not yet polled.

Source code in cocapi/events/_state.py

def get_clan(self, tag: str) -> dict[str, Any] | None:
    """Return the last-polled clan snapshot, or None if not yet polled."""
    return self._clan_snapshots.get(tag)

set_clan

set_clan(tag: str, data: dict[str, Any]) -> None

Store the latest clan snapshot for diffing on the next poll.

Source code in cocapi/events/_state.py

def set_clan(self, tag: str, data: dict[str, Any]) -> None:
    """Store the latest clan snapshot for diffing on the next poll."""
    self._clan_snapshots[tag] = data

get_members

get_members(tag: str) -> list[dict[str, Any]] | None

Return the last-polled member list for a clan, or None.

Source code in cocapi/events/_state.py

def get_members(self, tag: str) -> list[dict[str, Any]] | None:
    """Return the last-polled member list for a clan, or None."""
    return self._member_snapshots.get(tag)

set_members

set_members(
    tag: str, members: list[dict[str, Any]]
) -> None

Store the latest member list snapshot for a clan.

Source code in cocapi/events/_state.py

def set_members(self, tag: str, members: list[dict[str, Any]]) -> None:
    """Store the latest member list snapshot for a clan."""
    self._member_snapshots[tag] = members

get_war

get_war(tag: str) -> dict[str, Any] | None

Return the last-polled war snapshot, or None if not yet polled.

Source code in cocapi/events/_state.py

def get_war(self, tag: str) -> dict[str, Any] | None:
    """Return the last-polled war snapshot, or None if not yet polled."""
    return self._war_snapshots.get(tag)

set_war

set_war(tag: str, data: dict[str, Any]) -> None

Store the latest war snapshot for diffing on the next poll.

Source code in cocapi/events/_state.py

def set_war(self, tag: str, data: dict[str, Any]) -> None:
    """Store the latest war snapshot for diffing on the next poll."""
    self._war_snapshots[tag] = data

get_war_fsm

get_war_fsm(tag: str) -> WarStateMachine

Return the war state machine for a clan, creating one if needed.

Source code in cocapi/events/_state.py

def get_war_fsm(self, tag: str) -> WarStateMachine:
    """Return the war state machine for a clan, creating one if needed."""
    if tag not in self._war_fsms:
        self._war_fsms[tag] = WarStateMachine()
    return self._war_fsms[tag]

get_player

get_player(tag: str) -> dict[str, Any] | None

Return the last-polled player snapshot, or None if not yet polled.

Source code in cocapi/events/_state.py

def get_player(self, tag: str) -> dict[str, Any] | None:
    """Return the last-polled player snapshot, or None if not yet polled."""
    return self._player_snapshots.get(tag)

set_player

set_player(tag: str, data: dict[str, Any]) -> None

Store the latest player snapshot for diffing on the next poll.

Source code in cocapi/events/_state.py

def set_player(self, tag: str, data: dict[str, Any]) -> None:
    """Store the latest player snapshot for diffing on the next poll."""
    self._player_snapshots[tag] = data

should_poll

should_poll(resource_key: str, interval: float) -> bool

Check if enough time has passed since the last poll.

Source code in cocapi/events/_state.py

def should_poll(self, resource_key: str, interval: float) -> bool:
    """Check if enough time has passed since the last poll."""
    last = self._last_poll_times.get(resource_key, 0.0)
    return (time.time() - last) >= interval

mark_polled

mark_polled(resource_key: str) -> None

Record the current time as the last poll time for a resource.

Source code in cocapi/events/_state.py

def mark_polled(self, resource_key: str) -> None:
    """Record the current time as the last poll time for a resource."""
    self._last_poll_times[resource_key] = time.time()

save

save(path: Path) -> None

Persist state to JSON file for restart recovery.

Source code in cocapi/events/_state.py

def save(self, path: Path) -> None:
    """Persist state to JSON file for restart recovery."""
    path = _safe_resolve(path)
    data = {
        "clans": self._clan_snapshots,
        "members": self._member_snapshots,
        "wars": self._war_snapshots,
        "war_states": {t: fsm.state.value for t, fsm in self._war_fsms.items()},
        "players": self._player_snapshots,
        "last_poll_times": self._last_poll_times,
        "saved_at": time.time(),
    }
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(data, default=str), encoding="utf-8")

load

load(path: Path) -> bool

Load state from JSON file. Returns True if loaded successfully.

Source code in cocapi/events/_state.py

def load(self, path: Path) -> bool:
    """Load state from JSON file. Returns True if loaded successfully."""
    path = _safe_resolve(path)
    if not path.exists():
        return False
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
        self._clan_snapshots = data.get("clans", {})
        self._member_snapshots = data.get("members", {})
        self._war_snapshots = data.get("wars", {})
        for tag, state_str in data.get("war_states", {}).items():
            try:
                self._war_fsms[tag] = WarStateMachine(WarState(state_str))
            except ValueError:
                self._war_fsms[tag] = WarStateMachine()
        self._player_snapshots = data.get("players", {})
        self._last_poll_times = data.get("last_poll_times", {})
        return True
    except Exception as e:
        logger.warning("Failed to load polling state: %s", e)
        return False

EventStream

EventStream(
    api: CocApi,
    queue_size: int = 1000,
    persist_path: Path | str | None = None,
)

Real-time event stream for Clash of Clans resources.

Usage as async generator::

async with CocApi("token") as api:
    stream = EventStream(api)
    stream.watch_clans(["#ABC"], interval=60)

    async with stream:
        async for event in stream:
            print(event.event_type, event.changes)

Usage with callbacks::

@stream.on(EventType.MEMBER_JOINED)
async def on_join(event):
    print(f"{event.metadata['member_name']} joined!")

await stream.run()

PARAMETER	DESCRIPTION
api	A CocApi instance in async mode (inside async with). TYPE: CocApi
queue_size	Max events buffered before backpressure (0 = unlimited). TYPE: int DEFAULT: 1000
persist_path	Optional path for state persistence across restarts. TYPE: Path | str | None DEFAULT: None

Initialize the event stream.

PARAMETER	DESCRIPTION
api	A CocApi instance in async mode (inside async with). TYPE: CocApi
queue_size	Max events buffered before backpressure (0 = unlimited). TYPE: int DEFAULT: 1000
persist_path	Optional path for state persistence across restarts. TYPE: Path | str | None DEFAULT: None

RAISES	DESCRIPTION
RuntimeError	If the CocApi instance is not in async mode.

Source code in cocapi/events/_stream.py

def __init__(
    self,
    api: CocApi,
    queue_size: int = 1000,
    persist_path: Path | str | None = None,
) -> None:
    """Initialize the event stream.

    Args:
        api: A CocApi instance in async mode (inside ``async with``).
        queue_size: Max events buffered before backpressure (0 = unlimited).
        persist_path: Optional path for state persistence across restarts.

    Raises:
        RuntimeError: If the CocApi instance is not in async mode.
    """
    if not api.async_mode:
        raise RuntimeError(
            "EventStream requires CocApi in async mode. "
            "Use 'async with CocApi(token) as api:'"
        )
    self._api = api
    self._state = PollingState()
    self._queue: asyncio.Queue[Event] = asyncio.Queue(maxsize=queue_size)
    self._watchers: list[
        ClanWatcher | WarWatcher | PlayerWatcher | MaintenanceWatcher
    ] = []
    self._callbacks: dict[EventType | None, list[EventCallback]] = {}
    self._running = False
    self._persist_path = Path(persist_path) if persist_path else None

    if self._persist_path:
        self._state.load(self._persist_path)

watch_clans

watch_clans(
    tags: list[str],
    interval: float = 60.0,
    track_members: bool = True,
) -> EventStream

Register clans for polling.

PARAMETER	DESCRIPTION
tags	Clan tags to watch. TYPE: list[str]
interval	Seconds between polls (default 60). TYPE: float DEFAULT: 60.0
track_members	Track member joins/leaves/updates (default True). TYPE: bool DEFAULT: True

Source code in cocapi/events/_stream.py

def watch_clans(
    self,
    tags: list[str],
    interval: float = 60.0,
    track_members: bool = True,
) -> EventStream:
    """Register clans for polling.

    Args:
        tags: Clan tags to watch.
        interval: Seconds between polls (default 60).
        track_members: Track member joins/leaves/updates (default True).
    """
    watcher = ClanWatcher(
        api=self._api,
        state=self._state,
        queue=self._queue,
        clan_tags=tags,
        interval=interval,
        track_members=track_members,
    )
    self._watchers.append(watcher)
    return self

watch_wars

watch_wars(
    tags: list[str], interval: float = 30.0
) -> EventStream

Register clans for war state polling.

PARAMETER	DESCRIPTION
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/_stream.py

def watch_wars(
    self,
    tags: list[str],
    interval: float = 30.0,
) -> EventStream:
    """Register clans for war state polling.

    Args:
        tags: Clan tags to watch for war updates.
        interval: Seconds between polls (default 30).
    """
    watcher = WarWatcher(
        api=self._api,
        state=self._state,
        queue=self._queue,
        clan_tags=tags,
        interval=interval,
    )
    self._watchers.append(watcher)
    return self

watch_players

watch_players(
    tags: list[str],
    interval: float = 120.0,
    include_fields: frozenset[str] | None = None,
    exclude_fields: frozenset[str] | None = None,
) -> EventStream

Register players for polling.

PARAMETER	DESCRIPTION
tags	Player tags to watch. 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/_stream.py

def watch_players(
    self,
    tags: list[str],
    interval: float = 120.0,
    include_fields: frozenset[str] | None = None,
    exclude_fields: frozenset[str] | None = None,
) -> EventStream:
    """Register players for polling.

    Args:
        tags: Player tags to watch.
        interval: Seconds between polls (default 120).
        include_fields: Only report changes to these fields.
        exclude_fields: Ignore changes to these fields.
    """
    watcher = PlayerWatcher(
        api=self._api,
        state=self._state,
        queue=self._queue,
        player_tags=tags,
        interval=interval,
        include_fields=include_fields,
        exclude_fields=exclude_fields,
    )
    self._watchers.append(watcher)
    return self

watch_maintenance

watch_maintenance(
    interval: float = 30.0, probe_tag: str = "#JY9J2Y99"
) -> EventStream

Enable API maintenance detection.

Polls a known player endpoint to detect 503 maintenance responses. Emits MAINTENANCE_START and MAINTENANCE_END events.

PARAMETER	DESCRIPTION
interval	Seconds between probes (default 30). TYPE: float DEFAULT: 30.0
probe_tag	Player tag to probe (default #JY9J2Y99). TYPE: str DEFAULT: '#JY9J2Y99'

Source code in cocapi/events/_stream.py

def watch_maintenance(
    self,
    interval: float = 30.0,
    probe_tag: str = "#JY9J2Y99",
) -> EventStream:
    """Enable API maintenance detection.

    Polls a known player endpoint to detect 503 maintenance responses.
    Emits ``MAINTENANCE_START`` and ``MAINTENANCE_END`` events.

    Args:
        interval: Seconds between probes (default 30).
        probe_tag: Player tag to probe (default ``#JY9J2Y99``).
    """
    watcher = MaintenanceWatcher(
        api=self._api,
        state=self._state,
        queue=self._queue,
        interval=interval,
        probe_tag=probe_tag,
    )
    self._watchers.append(watcher)
    return self

on

on(
    event_type: EventType | None = None,
) -> Callable[[EventCallback], EventCallback]

Decorator to register an event callback.

PARAMETER	DESCRIPTION
event_type	Filter for specific event type. None matches all. TYPE: EventType | None DEFAULT: None

Source code in cocapi/events/_stream.py

def on(
    self,
    event_type: EventType | None = None,
) -> Callable[[EventCallback], EventCallback]:
    """Decorator to register an event callback.

    Args:
        event_type: Filter for specific event type. None matches all.
    """

    def decorator(func: EventCallback) -> EventCallback:
        self._callbacks.setdefault(event_type, []).append(func)
        return func

    return decorator

add_callback

add_callback(
    callback: EventCallback,
    event_type: EventType | None = None,
) -> None

Register an event callback programmatically.

Source code in cocapi/events/_stream.py

def add_callback(
    self,
    callback: EventCallback,
    event_type: EventType | None = None,
) -> None:
    """Register an event callback programmatically."""
    self._callbacks.setdefault(event_type, []).append(callback)

start async

start() -> None

Start all registered watchers.

Source code in cocapi/events/_stream.py

async def start(self) -> None:
    """Start all registered watchers."""
    self._running = True
    for watcher in self._watchers:
        watcher.start()

stop async

stop() -> None

Stop all watchers and persist state if configured.

Source code in cocapi/events/_stream.py

async def stop(self) -> None:
    """Stop all watchers and persist state if configured."""
    self._running = False
    for watcher in self._watchers:
        await watcher.stop()
    if self._persist_path:
        self._state.save(self._persist_path)

run async

run() -> None

Run the stream with callback dispatch until stopped.

Call stream.stop() from a callback or signal handler to exit.

Source code in cocapi/events/_stream.py

async def run(self) -> None:
    """Run the stream with callback dispatch until stopped.

    Call ``stream.stop()`` from a callback or signal handler to exit.
    """
    async with self:
        async for event in self:
            await self._dispatch(event)

Change dataclass

Change(field: str, old_value: Any, new_value: Any)

A single field-level change between two snapshots.

Event dataclass

Event(
    event_type: EventType,
    tag: str,
    timestamp: float = time(),
    old_data: dict[str, Any] | None = None,
    new_data: dict[str, Any] | None = None,
    changes: tuple[Change, ...] = (),
    metadata: dict[str, Any] = dict(),
)

An event produced by the polling system.

ATTRIBUTE	DESCRIPTION
event_type	The type of event. TYPE: EventType
tag	The clan or player tag this event relates to. TYPE: str
timestamp	Unix timestamp when the event was created. TYPE: float
old_data	Full previous snapshot (None on first poll). TYPE: dict[str, Any] | None
new_data	Full current snapshot (None for MEMBER_LEFT). TYPE: dict[str, Any] | None
changes	Tuple of field-level changes detected. TYPE: tuple[Change, ...]
metadata	Extra context (member_tag, war_state_from, etc.). TYPE: dict[str, Any]

EventType

Bases: Enum

All event types produced by the polling system.

WarState

Bases: Enum

War state machine states matching the CoC API.

WarStateMachine

WarStateMachine(initial_state: WarState = NOT_IN_WAR)

Tracks war state transitions for a single clan.

The CoC API returns war state as a string in clan_current_war(). This FSM detects when the state actually changes (vs. the same state being reported on consecutive polls).

Source code in cocapi/events/_war_fsm.py

def __init__(self, initial_state: WarState = WarState.NOT_IN_WAR) -> None:
    self.state = initial_state

transition

transition(raw_state: str) -> WarState | None

Attempt a state transition.

PARAMETER	DESCRIPTION
raw_state	The state field from clan_current_war() response. TYPE: str

RETURNS	DESCRIPTION
WarState | None	The new WarState if a transition occurred, or None if the state
WarState | None	is unchanged or the raw_state is unrecognized.

Source code in cocapi/events/_war_fsm.py

def transition(self, raw_state: str) -> WarState | None:
    """Attempt a state transition.

    Args:
        raw_state: The ``state`` field from ``clan_current_war()`` response.

    Returns:
        The new WarState if a transition occurred, or None if the state
        is unchanged or the raw_state is unrecognized.
    """
    try:
        new_state = WarState(raw_state)
    except ValueError:
        return None

    if new_state == self.state:
        return None

    self.state = new_state
    return new_state

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

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

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

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)

diff_dicts

diff_dicts(
    old: dict[str, Any],
    new: dict[str, Any],
    *,
    include_fields: frozenset[str] | None = None,
    exclude_fields: frozenset[str] | None = None,
) -> list[Change]

Shallow diff of two dicts, returning a list of Change objects.

Compares top-level keys only. Nested dicts/lists are compared by equality.

PARAMETER	DESCRIPTION
old	Previous snapshot. TYPE: dict[str, Any]
new	Current snapshot. TYPE: dict[str, Any]
include_fields	If set, only diff these fields. TYPE: frozenset[str] | None DEFAULT: None
exclude_fields	If set, skip these fields. TYPE: frozenset[str] | None DEFAULT: None

RETURNS	DESCRIPTION
list[Change]	List of Change objects for fields that differ.

Source code in cocapi/events/_diff.py

def diff_dicts(
    old: dict[str, Any],
    new: dict[str, Any],
    *,
    include_fields: frozenset[str] | None = None,
    exclude_fields: frozenset[str] | None = None,
) -> list[Change]:
    """Shallow diff of two dicts, returning a list of Change objects.

    Compares top-level keys only. Nested dicts/lists are compared by equality.

    Args:
        old: Previous snapshot.
        new: Current snapshot.
        include_fields: If set, only diff these fields.
        exclude_fields: If set, skip these fields.

    Returns:
        List of Change objects for fields that differ.
    """
    changes: list[Change] = []
    all_keys = set(old.keys()) | set(new.keys())

    for key in sorted(all_keys):
        if include_fields is not None and key not in include_fields:
            continue
        if exclude_fields is not None and key in exclude_fields:
            continue

        old_val = old.get(key)
        new_val = new.get(key)

        if old_val != new_val:
            changes.append(Change(field=key, old_value=old_val, new_value=new_val))

    return changes

diff_member_tags

diff_member_tags(
    old_members: list[dict[str, Any]],
    new_members: list[dict[str, Any]],
) -> tuple[
    list[dict[str, Any]],
    list[dict[str, Any]],
    list[tuple[dict[str, Any], dict[str, Any]]],
]

Compare member lists by tag to detect joins, leaves, and updates.

PARAMETER	DESCRIPTION
old_members	Previous member list. TYPE: list[dict[str, Any]]
new_members	Current member list. TYPE: list[dict[str, Any]]

RETURNS	DESCRIPTION
list[dict[str, Any]]	Tuple of (joined, left, updated) where:
list[dict[str, Any]]	joined: list of new member dicts
list[tuple[dict[str, Any], dict[str, Any]]]	left: list of old member dicts no longer present
tuple[list[dict[str, Any]], list[dict[str, Any]], list[tuple[dict[str, Any], dict[str, Any]]]]	updated: list of (old_member, new_member) tuples where data changed

Source code in cocapi/events/_diff.py

def diff_member_tags(
    old_members: list[dict[str, Any]],
    new_members: list[dict[str, Any]],
) -> tuple[
    list[dict[str, Any]],
    list[dict[str, Any]],
    list[tuple[dict[str, Any], dict[str, Any]]],
]:
    """Compare member lists by tag to detect joins, leaves, and updates.

    Args:
        old_members: Previous member list.
        new_members: Current member list.

    Returns:
        Tuple of (joined, left, updated) where:
        - joined: list of new member dicts
        - left: list of old member dicts no longer present
        - updated: list of (old_member, new_member) tuples where data changed
    """
    old_by_tag: dict[str, dict[str, Any]] = {m["tag"]: m for m in old_members}
    new_by_tag: dict[str, dict[str, Any]] = {m["tag"]: m for m in new_members}

    old_tags = set(old_by_tag.keys())
    new_tags = set(new_by_tag.keys())

    joined = [new_by_tag[t] for t in sorted(new_tags - old_tags)]
    left = [old_by_tag[t] for t in sorted(old_tags - new_tags)]

    updated: list[tuple[dict[str, Any], dict[str, Any]]] = []
    for tag in sorted(old_tags & new_tags):
        if old_by_tag[tag] != new_by_tag[tag]:
            updated.append((old_by_tag[tag], new_by_tag[tag]))

    return joined, left, updated

diff_named_list

diff_named_list(
    old_items: list[dict[str, Any]],
    new_items: list[dict[str, Any]],
    key: str = "name",
    compare_field: str = "level",
) -> list[tuple[str, Any, Any]]

Diff two lists of dicts keyed by key, comparing compare_field.

Used for troops, spells, heroes, and equipment which all share a {"name": str, "level": int} structure.

PARAMETER	DESCRIPTION
old_items	Previous list snapshot. TYPE: list[dict[str, Any]]
new_items	Current list snapshot. TYPE: list[dict[str, Any]]
key	Dict key used to identify items (default "name"). TYPE: str DEFAULT: 'name'
compare_field	Dict key whose value is compared (default "level"). TYPE: str DEFAULT: 'level'

RETURNS	DESCRIPTION
list[tuple[str, Any, Any]]	List of (name, old_value, new_value) for items where
list[tuple[str, Any, Any]]	compare_field changed. New items appear as (name, None, value).

Source code in cocapi/events/_diff.py

def diff_named_list(
    old_items: list[dict[str, Any]],
    new_items: list[dict[str, Any]],
    key: str = "name",
    compare_field: str = "level",
) -> list[tuple[str, Any, Any]]:
    """Diff two lists of dicts keyed by *key*, comparing *compare_field*.

    Used for troops, spells, heroes, and equipment which all share a
    ``{"name": str, "level": int}`` structure.

    Args:
        old_items: Previous list snapshot.
        new_items: Current list snapshot.
        key: Dict key used to identify items (default ``"name"``).
        compare_field: Dict key whose value is compared (default ``"level"``).

    Returns:
        List of ``(name, old_value, new_value)`` for items where
        *compare_field* changed.  New items appear as ``(name, None, value)``.
    """
    old_by_key: dict[str, dict[str, Any]] = {item[key]: item for item in old_items}
    new_by_key: dict[str, dict[str, Any]] = {item[key]: item for item in new_items}

    changes: list[tuple[str, Any, Any]] = []
    for name in sorted(new_by_key):
        new_val = new_by_key[name].get(compare_field)
        if name in old_by_key:
            old_val = old_by_key[name].get(compare_field)
            if old_val != new_val:
                changes.append((name, old_val, new_val))
        else:
            # Newly unlocked
            changes.append((name, None, new_val))

    return changes