Saltar a contenido

Bot y routers

bostie.Bot

Bot(*, intents: Intents = DEFAULT, dev_guilds: Iterable[int] = (), auto_sync: bool = True, contexts: Iterable[InteractionContextType] | None = None, integration_types: Iterable[ApplicationIntegrationType] | None = None, translate: Translate | None = None, auto_defer: float | None = 2.5, cache: bool = True, max_messages: int = 0, members: bool = False, chunk_members: bool = True, presences: bool = False, max_dm_channels: int = 0, guild_ready_timeout: float = 2.0, large_threshold: int = 50, allowed_mentions: AllowedMentions = DEFAULT_ALLOWED_MENTIONS, status: Status = ONLINE, activities: Iterable[ActivityDict] = (), afk: bool = False, idle_since: datetime | None = None, http_settings: HTTPSettings = DEFAULT_HTTP, api_url: str | None = None, voice: Defaults = DEFAULT_VOICE, shard_count: int | Literal['auto'] = 'auto', shard_ids: Iterable[int] | Literal['all'] = 'all', before_identify: Callable[[int], Awaitable[object]] | None = None, session_file: str | PathLike[str] | None = None, shutdown_grace: float = 0.0, owner_ids: Iterable[int] | None = None)

Bases: Router

A Discord bot: commands, components, listeners, REST and its gateway shards.

commands instance-attribute

commands: dict[CommandKey, Command | Group] = {}

context_menus instance-attribute

context_menus: dict[MenuKey, ContextMenuCommand | EntryPointCommand] = {}

listeners instance-attribute

listeners: dict[str, list[Listener]] = {}

component_routes instance-attribute

component_routes: dict[str, ComponentRoute[...]] = {}

modal_routes instance-attribute

modal_routes: dict[str, ComponentRoute[...]] = {}

tasks instance-attribute

tasks: dict[TaskCallback, Task] = {}

user class-attribute instance-attribute

user: CurrentUser | None = None

application_id class-attribute instance-attribute

application_id: int | None = None

cache instance-attribute

cache = Cache(max_messages, members=members, presences=presences, max_dm_channels=max_dm_channels)

voice instance-attribute

voice = VoiceManager(self, voice, lambda: None if self._bridge is None else self._bridge.core)

lookup instance-attribute

lookup = Lookup(self)

auto_defer instance-attribute

auto_defer = auto_defer

bot property

bot: Bot

This bot, so code written against a Router works on the bot too.

http property

http: HTTPClient

The REST client, for any endpoint; it exists once start() runs.

rest property

rest: REST

Discord's REST operations by resource; they exist once start() runs.

gateway property

gateway: GatewayClient

The gateway connection, for any opcode; it exists once start() runs.

latency property

latency: float | None

Mean seconds from a heartbeat to its ACK over the shards, or None.

intents property

intents: Intents

The gateway intents the bot was created with.

owner_ids property

owner_ids: frozenset[int]

Who checks.is_owner lets through: the given ids, else the app's owners.

is_running property

is_running: bool

Whether start() or open() runs the bot, until close() ends it.

is_ready property

is_ready: bool

Whether Started was sent since the last start(): every shard was ready.

on_error

on_error(handler: H) -> H
on_error(exception_type: type[E]) -> Callable[[ErrorHook[E]], ErrorHook[E]]
on_error(first: type[A], second: type[B]) -> Callable[[ErrorHook[A | B]], ErrorHook[A | B]]
on_error(*exception_types: type[BaseException]) -> Callable[[ErrorHook[Any]], ErrorHook[Any]]
on_error(*targets: Any) -> Any

Register async def hook(ctx, error) for the errors of what this covers.

Use @x.on_error for every Exception, or name the exception types.

command

command(callback: CommandCallback) -> Command
command(callback: None = None, /, *, name: str | None = None, description: str | None = None, name_localizations: Mapping[str, str] | None = None, description_localizations: Mapping[str, str] | None = None, ephemeral: bool | None = None, auto_defer: float | Unset | None = UNSET, default_permissions: Permissions | None = None, nsfw: bool | None = None, guilds: Iterable[int] = (), contexts: Iterable[InteractionContextType] | None = None, integration_types: Iterable[ApplicationIntegrationType] | None = None, checks: Iterable[Check] = (), extra: Mapping[str, Any] | None = None) -> Callable[[CommandCallback], Command]
command(callback: CommandCallback | None = None, /, *, name: str | None = None, description: str | None = None, name_localizations: Mapping[str, str] | None = None, description_localizations: Mapping[str, str] | None = None, ephemeral: bool | None = None, auto_defer: float | Unset | None = UNSET, default_permissions: Permissions | None = None, nsfw: bool | None = None, guilds: Iterable[int] = (), contexts: Iterable[InteractionContextType] | None = None, integration_types: Iterable[ApplicationIntegrationType] | None = None, checks: Iterable[Check] = (), extra: Mapping[str, Any] | None = None) -> Command | Callable[[CommandCallback], Command]

Register a slash command, as @router.command or @router.command(...).

group

group(name: str, description: str, *, name_localizations: Mapping[str, str] | None = None, description_localizations: Mapping[str, str] | None = None, default_permissions: Permissions | None = None, nsfw: bool | None = None, guilds: Iterable[int] = (), contexts: Iterable[InteractionContextType] | None = None, integration_types: Iterable[ApplicationIntegrationType] | None = None, checks: Iterable[Check] = (), extra: Mapping[str, Any] | None = None) -> Group

Register a slash command made of subcommands.

user_command

user_command(name: str, *, name_localizations: Mapping[str, str] | None = None, ephemeral: bool | None = None, auto_defer: float | Unset | None = UNSET, default_permissions: Permissions | None = None, nsfw: bool | None = None, guilds: Iterable[int] = (), contexts: Iterable[InteractionContextType] | None = None, integration_types: Iterable[ApplicationIntegrationType] | None = None, checks: Iterable[Check] = (), extra: Mapping[str, Any] | None = None) -> Callable[[CommandCallback], ContextMenuCommand]

Register async def f(ctx, user: User) in the context menu of users.

message_command

message_command(name: str, *, name_localizations: Mapping[str, str] | None = None, ephemeral: bool | None = None, auto_defer: float | Unset | None = UNSET, default_permissions: Permissions | None = None, nsfw: bool | None = None, guilds: Iterable[int] = (), contexts: Iterable[InteractionContextType] | None = None, integration_types: Iterable[ApplicationIntegrationType] | None = None, checks: Iterable[Check] = (), extra: Mapping[str, Any] | None = None) -> Callable[[CommandCallback], ContextMenuCommand]

Register async def f(ctx, message: Message) in messages' context menu.

entry_point

entry_point(name: str, *, description: str | None = None, name_localizations: Mapping[str, str] | None = None, description_localizations: Mapping[str, str] | None = None, ephemeral: bool | None = None, auto_defer: float | Unset | None = UNSET, default_permissions: Permissions | None = None, nsfw: bool | None = None, contexts: Iterable[InteractionContextType] | None = None, integration_types: Iterable[ApplicationIntegrationType] | None = None, checks: Iterable[Check] = (), extra: Mapping[str, Any] | None = None) -> Callable[[CommandCallback], EntryPointCommand]

Register async def f(ctx) as the entry point that opens the app's Activity.

The app answers it, with ctx.launch_activity() or otherwise.

listen

listen(event: type[E]) -> Callable[[EventHandler[E]], EventHandler[E]]
listen(first: type[A], second: type[B]) -> Callable[[EventHandler[A | B]], EventHandler[A | B]]
listen(first: type[AnyEvent], second: type[AnyEvent], third: type[AnyEvent], /, *more: type[AnyEvent]) -> Callable[[L], L]
listen(listener: L) -> L
listen(*targets: Any) -> Any

Register a listener for the events in its annotation, or the given ones.

remove_listener

remove_listener(listener: ListenerCallback) -> None

Stop calling listener for every event it listens to here.

listen_raw

listen_raw(*names: str) -> Callable[[L], L]

Register a listener that gets a RawEvent for each of these dispatch names.

on_component

on_component(template: str, *, ephemeral: bool | None = None, auto_defer: float | Unset | None = UNSET, checks: Iterable[Check] = ()) -> Callable[[Callable[P, Coroutine[Any, Any, None]]], ComponentRoute[P]]

Route buttons and selects whose custom_id fits template.

on_modal

on_modal(template: str, *, ephemeral: bool | None = None, auto_defer: float | Unset | None = UNSET, checks: Iterable[Check] = ()) -> Callable[[Callable[P, Coroutine[Any, Any, None]]], ComponentRoute[P]]

Route the submits of modals whose custom_id fits template.

every

every(*, seconds: float | None = None, time: time | Iterable[time] = ()) -> Callable[[TaskCallback], Task]

Run async def task() while the bot is ready, every seconds or daily.

time names the times of day; one without tzinfo is in UTC. The decorator returns the Task, which starts, stops and reschedules it.

check

check(predicate: C) -> C

Register a check that runs before every handler of this router.

A check takes the context and returns or awaits a bool; False refuses the run with CheckFailure, and it may raise a CheckFailure of its own.

before_invoke

before_invoke(hook: H) -> H

Register async def hook(ctx), run after the checks of each handler here.

after_invoke

after_invoke(hook: H) -> H

Register async def hook(ctx), run after each handler here, even failed.

include

include(router: Router | str) -> None

Add a router's commands, listeners, routes and tasks, including later ones.

A str names a module to import, whose router is added.

exclude

exclude(router: Router | str) -> None

Remove an included router and all it added; a str names its module.

Python forgets an excluded module, so including it again imports it anew.

reload

reload(name: str) -> None

Exclude a module's router, import the module anew and include its router.

When that fails, the old router stays included and the error is raised.

match_component

match_component(custom_id: str) -> tuple[ComponentRoute[...], dict[str, str | int]] | None

The component route for a custom_id and its placeholder values, or None.

match_modal

match_modal(custom_id: str) -> tuple[ComponentRoute[...], dict[str, str | int]] | None

The modal route for a custom_id and its placeholder values, or None.

find_error_handlers

find_error_handlers(error: BaseException) -> Iterator[ErrorHook[Any]]

Hooks for error: this router's nearest match first, then each parent's.

walk_commands

walk_commands() -> Iterator[Command]

Every command with a callback, subcommands included, in order.

run

run(token: str | None = None, *, loop_factory: Callable[[], AbstractEventLoop] | None = None) -> None

Run the bot until it stops; Ctrl+C and SIGTERM close it cleanly.

Logging goes to stderr at INFO unless the root logger is already configured.

start async

start(token: str | None = None) -> None

Connect, sync commands and handle events until close() or a fatal error.

The token defaults to the BOSTIE_TOKEN environment variable.

open async

open(token: str | None = None) -> Application

Start REST, send Starting, sync commands and run tasks, with no gateway.

close async

close() -> None

Disconnect, cancel running handlers after shutdown_grace, close the bridge.

The shards end their sessions with code 1000, or save them to the session file.

wait_until_ready async

wait_until_ready() -> None

Wait until every shard is ready; raise StateError if the bot stops first.

wait_for async

wait_for(event: type[E], *, check: Callable[[E], bool] | None = None) -> E

Wait for the next event that passes check.

An exception from check is raised here; bound the wait with asyncio.timeout().

stream async

stream(event: type[E], *, check: Callable[[E], bool] | None = None) -> AsyncGenerator[AsyncIterator[E]]

Queue each event passing check for async for while the block runs.

wait_for_component async

wait_for_component(*custom_ids: str, check: Callable[[ComponentContext], bool] | None = None, ephemeral: bool = False, auto_defer: float | Unset | None = UNSET) -> ComponentContext

Wait for a use of a component with one of custom_ids, or of any.

The context answers like a route's, and no route runs for that use. An exception from check is raised here; bound the wait with asyncio.timeout().

wait_for_modal async

wait_for_modal(*custom_ids: str, check: Callable[[ModalContext], bool] | None = None, ephemeral: bool = False, auto_defer: float | Unset | None = UNSET) -> ModalContext

Wait for the submit of a modal with one of custom_ids, or of any.

The context answers like a route's, and no route runs for that submit.

dispatch

dispatch(event: _LocalEvent) -> None

Send an event of your own to its listeners, waiters and streams.

update_presence async

update_presence(*, status: Status | None = None, activities: Iterable[ActivityDict] | None = None, afk: bool | None = None, idle_since: datetime | Unset | None = UNSET) -> None

Change what the bot shows on every shard, and on sessions it starts later.

request_members async

request_members(guild_id: int, *, query: str | None = None, limit: int | None = None, user_ids: Iterable[int] | None = None, presences: bool = False, timeout: float = REQUEST_TIMEOUT) -> RequestedMembers

Ask the gateway for members of a guild and gather every chunk of the answer.

Without query or user_ids it asks for every member. timeout bounds the wait for each chunk, 30 s by default, and raises TimeoutError when it ends.

request_soundboard_sounds async

request_soundboard_sounds(guild_id: int) -> tuple[SoundboardSound, ...]

Ask the gateway for a guild's soundboard sounds; TimeoutError after 30 s.

request_channel_info async

request_channel_info(guild_id: int, *, fields: Iterable[Literal['status', 'voice_start_time']] = _CHANNEL_INFO_FIELDS) -> tuple[VoiceChannelInfo, ...]

Ask the gateway for the voice status and start time of a guild's channels.

sync_commands async

sync_commands() -> list[SyncResult]

Overwrite the commands on Discord now and return what differed.

command_mention async

command_mention(name: str, *, guild_id: int | None = None) -> str

</name:id>, shown as a clickable /name; name may end in subcommands.

In guild_id, a command registered there wins over a global one. Without a sync since the start, it asks Discord for the ids once per scope.

bostie.Router

Router(*, ephemeral: bool | None = None, auto_defer: float | Unset | None = UNSET, default_permissions: Permissions | None = None, nsfw: bool | None = None, contexts: Iterable[InteractionContextType] | None = None, integration_types: Iterable[ApplicationIntegrationType] | None = None)

Bases: ErrorHooks

Commands, listeners, routes and error hooks; a bot is a router.

Set what its commands and routes inherit where they set nothing.

commands instance-attribute

commands: dict[CommandKey, Command | Group] = {}

context_menus instance-attribute

context_menus: dict[MenuKey, ContextMenuCommand | EntryPointCommand] = {}

listeners instance-attribute

listeners: dict[str, list[Listener]] = {}

component_routes instance-attribute

component_routes: dict[str, ComponentRoute[...]] = {}

modal_routes instance-attribute

modal_routes: dict[str, ComponentRoute[...]] = {}

tasks instance-attribute

tasks: dict[TaskCallback, Task] = {}

bot property

bot: Bot

The bot this router is included in; StateError until then.

on_error

on_error(handler: H) -> H
on_error(exception_type: type[E]) -> Callable[[ErrorHook[E]], ErrorHook[E]]
on_error(first: type[A], second: type[B]) -> Callable[[ErrorHook[A | B]], ErrorHook[A | B]]
on_error(*exception_types: type[BaseException]) -> Callable[[ErrorHook[Any]], ErrorHook[Any]]
on_error(*targets: Any) -> Any

Register async def hook(ctx, error) for the errors of what this covers.

Use @x.on_error for every Exception, or name the exception types.

command

command(callback: CommandCallback) -> Command
command(callback: None = None, /, *, name: str | None = None, description: str | None = None, name_localizations: Mapping[str, str] | None = None, description_localizations: Mapping[str, str] | None = None, ephemeral: bool | None = None, auto_defer: float | Unset | None = UNSET, default_permissions: Permissions | None = None, nsfw: bool | None = None, guilds: Iterable[int] = (), contexts: Iterable[InteractionContextType] | None = None, integration_types: Iterable[ApplicationIntegrationType] | None = None, checks: Iterable[Check] = (), extra: Mapping[str, Any] | None = None) -> Callable[[CommandCallback], Command]
command(callback: CommandCallback | None = None, /, *, name: str | None = None, description: str | None = None, name_localizations: Mapping[str, str] | None = None, description_localizations: Mapping[str, str] | None = None, ephemeral: bool | None = None, auto_defer: float | Unset | None = UNSET, default_permissions: Permissions | None = None, nsfw: bool | None = None, guilds: Iterable[int] = (), contexts: Iterable[InteractionContextType] | None = None, integration_types: Iterable[ApplicationIntegrationType] | None = None, checks: Iterable[Check] = (), extra: Mapping[str, Any] | None = None) -> Command | Callable[[CommandCallback], Command]

Register a slash command, as @router.command or @router.command(...).

group

group(name: str, description: str, *, name_localizations: Mapping[str, str] | None = None, description_localizations: Mapping[str, str] | None = None, default_permissions: Permissions | None = None, nsfw: bool | None = None, guilds: Iterable[int] = (), contexts: Iterable[InteractionContextType] | None = None, integration_types: Iterable[ApplicationIntegrationType] | None = None, checks: Iterable[Check] = (), extra: Mapping[str, Any] | None = None) -> Group

Register a slash command made of subcommands.

user_command

user_command(name: str, *, name_localizations: Mapping[str, str] | None = None, ephemeral: bool | None = None, auto_defer: float | Unset | None = UNSET, default_permissions: Permissions | None = None, nsfw: bool | None = None, guilds: Iterable[int] = (), contexts: Iterable[InteractionContextType] | None = None, integration_types: Iterable[ApplicationIntegrationType] | None = None, checks: Iterable[Check] = (), extra: Mapping[str, Any] | None = None) -> Callable[[CommandCallback], ContextMenuCommand]

Register async def f(ctx, user: User) in the context menu of users.

message_command

message_command(name: str, *, name_localizations: Mapping[str, str] | None = None, ephemeral: bool | None = None, auto_defer: float | Unset | None = UNSET, default_permissions: Permissions | None = None, nsfw: bool | None = None, guilds: Iterable[int] = (), contexts: Iterable[InteractionContextType] | None = None, integration_types: Iterable[ApplicationIntegrationType] | None = None, checks: Iterable[Check] = (), extra: Mapping[str, Any] | None = None) -> Callable[[CommandCallback], ContextMenuCommand]

Register async def f(ctx, message: Message) in messages' context menu.

entry_point

entry_point(name: str, *, description: str | None = None, name_localizations: Mapping[str, str] | None = None, description_localizations: Mapping[str, str] | None = None, ephemeral: bool | None = None, auto_defer: float | Unset | None = UNSET, default_permissions: Permissions | None = None, nsfw: bool | None = None, contexts: Iterable[InteractionContextType] | None = None, integration_types: Iterable[ApplicationIntegrationType] | None = None, checks: Iterable[Check] = (), extra: Mapping[str, Any] | None = None) -> Callable[[CommandCallback], EntryPointCommand]

Register async def f(ctx) as the entry point that opens the app's Activity.

The app answers it, with ctx.launch_activity() or otherwise.

listen

listen(event: type[E]) -> Callable[[EventHandler[E]], EventHandler[E]]
listen(first: type[A], second: type[B]) -> Callable[[EventHandler[A | B]], EventHandler[A | B]]
listen(first: type[AnyEvent], second: type[AnyEvent], third: type[AnyEvent], /, *more: type[AnyEvent]) -> Callable[[L], L]
listen(listener: L) -> L
listen(*targets: Any) -> Any

Register a listener for the events in its annotation, or the given ones.

remove_listener

remove_listener(listener: ListenerCallback) -> None

Stop calling listener for every event it listens to here.

listen_raw

listen_raw(*names: str) -> Callable[[L], L]

Register a listener that gets a RawEvent for each of these dispatch names.

on_component

on_component(template: str, *, ephemeral: bool | None = None, auto_defer: float | Unset | None = UNSET, checks: Iterable[Check] = ()) -> Callable[[Callable[P, Coroutine[Any, Any, None]]], ComponentRoute[P]]

Route buttons and selects whose custom_id fits template.

on_modal

on_modal(template: str, *, ephemeral: bool | None = None, auto_defer: float | Unset | None = UNSET, checks: Iterable[Check] = ()) -> Callable[[Callable[P, Coroutine[Any, Any, None]]], ComponentRoute[P]]

Route the submits of modals whose custom_id fits template.

every

every(*, seconds: float | None = None, time: time | Iterable[time] = ()) -> Callable[[TaskCallback], Task]

Run async def task() while the bot is ready, every seconds or daily.

time names the times of day; one without tzinfo is in UTC. The decorator returns the Task, which starts, stops and reschedules it.

check

check(predicate: C) -> C

Register a check that runs before every handler of this router.

A check takes the context and returns or awaits a bool; False refuses the run with CheckFailure, and it may raise a CheckFailure of its own.

before_invoke

before_invoke(hook: H) -> H

Register async def hook(ctx), run after the checks of each handler here.

after_invoke

after_invoke(hook: H) -> H

Register async def hook(ctx), run after each handler here, even failed.

include

include(router: Router | str) -> None

Add a router's commands, listeners, routes and tasks, including later ones.

A str names a module to import, whose router is added.

exclude

exclude(router: Router | str) -> None

Remove an included router and all it added; a str names its module.

Python forgets an excluded module, so including it again imports it anew.

reload

reload(name: str) -> None

Exclude a module's router, import the module anew and include its router.

When that fails, the old router stays included and the error is raised.

match_component

match_component(custom_id: str) -> tuple[ComponentRoute[...], dict[str, str | int]] | None

The component route for a custom_id and its placeholder values, or None.

match_modal

match_modal(custom_id: str) -> tuple[ComponentRoute[...], dict[str, str | int]] | None

The modal route for a custom_id and its placeholder values, or None.

find_error_handlers

find_error_handlers(error: BaseException) -> Iterator[ErrorHook[Any]]

Hooks for error: this router's nearest match first, then each parent's.

walk_commands

walk_commands() -> Iterator[Command]

Every command with a callback, subcommands included, in order.

bostie.router.Listener dataclass

Listener(callback: ListenerCallback, event: type[AnyEvent] | None)

A registered listener; event is the class to decode, or None for RawEvent.

callback instance-attribute

callback: ListenerCallback

event instance-attribute

event: type[AnyEvent] | None