Client#

Air Raid Alert API client module.

class alertapi.impl.client.APIClient(access_token: str)[source]#

Alert API client.

Api client for Air Raid Alert API.

Parameters: access_token (builtins.str) – An access token to the Air Raid Alert API. Can be obtained here

Example

import asyncio

import alertapi

    async def main() -> None:
        client = alertapi.APIClient(access_token='...')
        print(await client.fetch_states())


    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())

async fetch_state(state: Union[Literal[StateConverter.STATES], snowflakes.Snowflake]) → states.State[source]#

Fetch state entity from Alert API

Parameters

state (Union[Literal[converters.StateConverter.STATES], snowflakes.Snowflake]) – State for search.

Returns

Deserialied state entity.

Return type

alertapi.states.State

Raises

alertapi.errors.StateNotFound –

If specified state does not exists.

async fetch_states(state: snowflakes.Snowflake) → states.State[source]#

async fetch_states(with_alert: bool) → tuple[states.State]

async fetch_states(limit: int) → tuple[states.State]

Fetch all state entities from Alert API.

Parameters

state (Optional[alertapi.snowflakes.Snowflake]) – State for search. If specified, returns state object with information.
with_alert (Optional[builtins.bool]) – Fetch states with active/inactive alert.
limit (Optional[builtins.int]) – Limit of states. Defaults to 25.

Returns

alertapi.states.State – Deserialied state entity if state is specified.
builtins.tuple[alertapi.states.State] – Tuple of deserialised state entities.

Raises

alertapi.errors.StateNotFound –

If specified state does not exists.

async is_alert(state: Union[Literal[StateConverter.STATES], snowflakes.Snowflake]) → bool[source]#

Check whether active alert in specified state or not.

Parameters

state (Union[Literal[converters.StateConverter.STATES], snowflakes.Snowflake]) – State for search.

Returns

builtins.True if alert is active.
builtins.False if alert is inactive.

Return type

builtins.bool

Raises

alertapi.errors.StateNotFound –

If specified state does not exists.

async static_map() → images.Image[source]#

Fetch static map of states.

Returns: Deserialized Image object.
Return type: alertapi.images.Image

class alertapi.impl.client.GatewayClient(access_token: str)[source]#

Gateway Alert API client.

Parameters

access_token (builtins.str) –

An access token to the Air Raid Alert API. Can be obtained here

Example

import alertapi

client = alertapi.GatewayClient(access_token='...')

@client.listen(alertapi.ClientConnectedEvent)
async def on_client_connected(event: alertapi.ClientConnectedEvent) -> None:
    states = await event.api.fetch_states()
    print(states)


@client.listen(alertapi.StateUpdateEvent)
async def on_state_update(event: alertapi.StateUpdateEvent) -> None:
    print('State updated': event.state)


client.connect()

connect() → None[source]#: Connect client to Air Raid Alert API endpoint.

listen(event_type: Type[base_events.Event]) → Callable[source]#

Generate a decorator to subscribe a callback to an event type.

Parameters: event_type (Type[alertapi.events.base_events.Event]) – The event type to subscribe to.
Returns: A decorator for a coroutine function that passes it to EventManager.subscribe before returning the function reference.
Return type: Callable

subscribe(event_type: Type[base_events.Event], callback: Callable) → None[source]#

Subscribe a given callback to a given event type.

Parameters

event_type (Type[alertapi.events.base_events.Event]) – The event type to listen for. This will also listen for any subclasses of the given type.
callback (Callable) – Must be a coroutine function to invoke. This should consume an instance of the given event.

Example

The following demonstrates subscribing a callback to state update event.

from alertapi.events.base_events import StateUpdateEvent

async def on_state_update(event):
    ...

client.subscribe(StateUpdateEvent, on_state_update)