Bot とルーター¶
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.
context_menus
instance-attribute
¶
context_menus: dict[MenuKey, ContextMenuCommand | EntryPointCommand] = {}
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)
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.
owner_ids
property
¶
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_types: type[BaseException]) -> Callable[[ErrorHook[Any]], ErrorHook[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(first: type[AnyEvent], second: type[AnyEvent], third: type[AnyEvent], /, *more: type[AnyEvent]) -> Callable[[L], L]
listen(listener: L) -> L
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
¶
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
¶
Add a router's commands, listeners, routes and tasks, including later ones.
A str names a module to import, whose router is added.
exclude
¶
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
¶
The component route for a custom_id and its placeholder values, or None.
match_modal
¶
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
¶
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 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.
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.
context_menus
instance-attribute
¶
context_menus: dict[MenuKey, ContextMenuCommand | EntryPointCommand] = {}
on_error
¶
on_error(handler: H) -> H
on_error(*exception_types: type[BaseException]) -> Callable[[ErrorHook[Any]], ErrorHook[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(first: type[AnyEvent], second: type[AnyEvent], third: type[AnyEvent], /, *more: type[AnyEvent]) -> Callable[[L], L]
listen(listener: L) -> L
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
¶
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
¶
Add a router's commands, listeners, routes and tasks, including later ones.
A str names a module to import, whose router is added.
exclude
¶
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
¶
The component route for a custom_id and its placeholder values, or None.
match_modal
¶
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.