"""Backend base class for chatom.
This module provides the base class that all backends must implement.
"""
import asyncio
from abc import abstractmethod
from collections.abc import AsyncIterator, Callable
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime
from functools import cached_property
from re import Pattern
from threading import Lock
from typing import (
Any,
ClassVar,
TypeVar,
Union,
cast,
)
from pydantic import Field
from ..base import (
Attachment,
BackendCapabilities,
BaseModel,
Channel,
ChannelRegistry,
Interaction,
Message,
Organization,
Presence,
User,
UserRegistry,
)
from ..format.variant import Format
__all__ = (
"Backend",
"BackendBase",
"SyncHelper",
)
# Type variable for backend subclasses
B = TypeVar("B", bound="BackendBase")
[docs]
class SyncHelper:
"""Helper class to run async methods synchronously.
This provides a convenient way to call async methods from sync code
by managing an event loop in a background thread. Uses __getattr__
to dynamically wrap any async method on the backend.
Example:
>>> backend = MyBackend()
>>> # Call async method synchronously
>>> user = backend.sync.lookup_user(id="123")
>>> # Any async method can be called:
>>> backend.sync.connect()
>>> backend.sync.send_message(channel_id="C123", content="Hello")
"""
def __init__(self, backend: "BackendBase") -> None:
"""Initialize the sync helper.
Args:
backend: The backend instance to wrap.
"""
self._backend = backend
self._executor: ThreadPoolExecutor | None = None
self._loop: asyncio.AbstractEventLoop | None = None
self._lock = Lock()
def _get_loop(self) -> asyncio.AbstractEventLoop:
"""Get or create the event loop for sync execution."""
if self._loop is None or self._loop.is_closed():
with self._lock:
if self._loop is None or self._loop.is_closed():
self._loop = asyncio.new_event_loop()
assert self._loop is not None
return self._loop
def _run_async(self, coro: Any) -> Any:
"""Run a coroutine synchronously.
Args:
coro: The coroutine to run.
Returns:
The result of the coroutine.
"""
loop = self._get_loop()
try:
return loop.run_until_complete(coro)
except RuntimeError:
# If we're already in an async context, use a thread
if self._executor is None:
self._executor = ThreadPoolExecutor(max_workers=1)
def run_in_new_loop() -> Any:
new_loop = asyncio.new_event_loop()
asyncio.set_event_loop(new_loop)
try:
return new_loop.run_until_complete(coro)
finally:
new_loop.close()
future = self._executor.submit(run_in_new_loop)
return future.result()
def __getattr__(self, name: str) -> Callable[..., Any]:
"""Dynamically create sync wrappers for async backend methods.
This method is called when accessing any attribute not found on SyncHelper.
It looks for a corresponding method on the backend and, if it's async,
returns a synchronous wrapper.
Args:
name: The method name to look up.
Returns:
A callable that wraps the async method synchronously.
Raises:
AttributeError: If the method doesn't exist on the backend.
"""
# Avoid infinite recursion for private attributes
if name.startswith("_"):
raise AttributeError(f"'{type(self).__name__}' has no attribute '{name}'")
# Get the method from the backend
method = getattr(self._backend, name, None)
if method is None:
raise AttributeError(f"'{type(self._backend).__name__}' has no attribute '{name}'")
# Check if it's a callable
if not callable(method):
raise AttributeError(f"'{name}' is not a method on '{type(self._backend).__name__}'") # noqa: TRY004
# Return a wrapper that runs the coroutine synchronously
def sync_wrapper(*args: Any, **kwargs: Any) -> Any:
result = method(*args, **kwargs)
# Check if it's a coroutine (async method)
if asyncio.iscoroutine(result):
return self._run_async(result)
return result
return sync_wrapper
[docs]
def close(self) -> None:
"""Clean up resources."""
if self._executor:
self._executor.shutdown(wait=False)
self._executor = None
if self._loop and not self._loop.is_closed():
self._loop.close()
self._loop = None
[docs]
class BackendBase(BaseModel):
"""Base class for all chat backend implementations.
This provides a unified interface for interacting with chat platforms.
All backends must implement the abstract methods.
Backends should be registered with the BackendRegistry via entry points
or by calling register_backend().
Attributes:
name: The backend identifier (e.g., 'slack', 'discord').
display_name: Human-readable name for the backend.
format: The preferred output format for this backend.
capabilities: The capabilities supported by this backend.
connected: Whether currently connected.
users: Registry of cached users.
channels: Registry of cached channels.
Example:
>>> class MyBackend(BackendBase):
... name = "my_backend"
... display_name = "My Backend"
... format = Format.MARKDOWN
...
... async def connect(self):
... # Implementation
... pass
"""
# Class-level attributes that define the backend
name: ClassVar[str] = ""
display_name: ClassVar[str] = ""
format: ClassVar[Format] = Format.MARKDOWN
# Regex for parsing user mentions out of raw incoming message text.
# Group 1 must capture the platform user ID. ``None`` means this
# backend does not embed user IDs inline (e.g. Telegram uses
# MessageEntity rather than inline markers), and mention translation
# should be skipped.
mention_pattern: ClassVar[Pattern[str] | None] = None
# Instance attributes
capabilities: BackendCapabilities | None = Field(
default=None,
description="The capabilities supported by this backend.",
)
connected: bool = Field(
default=False,
description="Whether currently connected.",
)
users: UserRegistry = Field(
default_factory=UserRegistry,
description="Registry of cached users.",
)
channels: ChannelRegistry = Field(
default_factory=ChannelRegistry,
description="Registry of cached channels.",
)
config: Any | None = Field(
default=None,
description="Backend-specific configuration. Subclasses override with their config type.",
)
_sync: SyncHelper | None = None
_presence_heartbeat_running: bool = False
_presence_heartbeat_task: asyncio.Task[None] | None = None
@cached_property
def sync(self) -> SyncHelper:
"""Get the sync helper for calling async methods synchronously.
Returns:
SyncHelper instance that wraps async methods.
Example:
>>> backend = MyBackend()
>>> backend.sync.connect() # Calls connect() synchronously
>>> user = backend.sync.lookup_user(id="123")
"""
return SyncHelper(self)
[docs]
def normalize_channel_id(self, channel_id: str) -> str:
"""Return a canonical form of a channel id for equality comparison.
Some platforms expose the same channel under multiple equivalent id
encodings. Backends where that happens should override this to return
a single canonical form so that equal channels compare equal. The
default returns the id unchanged.
Args:
channel_id: The channel id to canonicalize.
Returns:
The canonical channel id.
"""
return channel_id
# Connection methods
[docs]
@abstractmethod
async def connect(self) -> None:
"""Establish connection to the backend.
This should authenticate and establish a connection to the
chat platform. After successful connection, `connected` should
be set to True.
Raises:
ConnectionError: If connection fails.
"""
raise NotImplementedError("Subclass must implement connect()")
[docs]
@abstractmethod
async def disconnect(self) -> None:
"""Disconnect from the backend.
This should cleanly close the connection and release resources.
After disconnection, `connected` should be set to False.
"""
raise NotImplementedError("Subclass must implement disconnect()")
# User lookup methods
[docs]
async def lookup_user(
self,
*,
id: str | None = None,
name: str | None = None,
email: str | None = None,
handle: str | None = None,
) -> User | None:
"""Look up a user by any identifier.
First checks the local cache, then fetches from the backend
if not found. This method will attempt to use the backend's
fetch_user with whatever identifiers are provided.
Args:
id: User ID.
name: User name.
email: User email address.
handle: User handle/username.
Returns:
The user if found, None otherwise.
"""
# Try local registry first
user = self.users.lookup(id=id, name=name, email=email, handle=handle)
if user:
return user
# Try fetching from backend with whatever identifiers we have
if id or name or email or handle:
user = await self.fetch_user(id=id, name=name, email=email, handle=handle)
if user:
self.users.add(user)
return user
return None
[docs]
@abstractmethod
async def fetch_user(
self,
identifier: str | User | None = None,
*,
id: str | None = None,
name: str | None = None,
email: str | None = None,
handle: str | None = None,
) -> User | None:
"""Fetch a user from the backend.
This method accepts flexible input types for convenience:
- Pass a User object to validate/refresh it
- Pass an ID string as the first positional argument
- Use keyword arguments for lookup by name, email, or handle
The backend will attempt to resolve the user using the most
efficient method available for the platform.
Args:
identifier: A User object or user ID string.
id: User ID (alternative to positional identifier).
name: User name or display name to search for.
email: Email address to search for.
handle: Username/handle to search for.
Returns:
The user if found, None otherwise.
Example:
>>> # All of these work:
>>> user = await backend.fetch_user("U123456")
>>> user = await backend.fetch_user(id="U123456")
>>> user = await backend.fetch_user(name="John Doe")
>>> user = await backend.fetch_user(email="john@example.com")
>>> user = await backend.fetch_user(existing_user) # refresh
"""
raise NotImplementedError("Subclass must implement fetch_user()")
# Channel lookup methods
[docs]
async def lookup_channel(
self,
*,
id: str | None = None,
name: str | None = None,
) -> Channel | None:
"""Look up a channel by any identifier.
First checks the local cache, then fetches from the backend
if not found. This method will attempt to use the backend's
fetch_channel with whatever identifiers are provided.
Args:
id: Channel ID.
name: Channel name.
Returns:
The channel if found, None otherwise.
"""
# Try local registry first
channel = self.channels.lookup(id=id, name=name)
if channel:
return channel
# Try fetching from backend with whatever identifiers we have
if id or name:
channel = await self.fetch_channel(id=id, name=name)
if channel:
self.channels.add(channel)
return channel
return None
[docs]
async def lookup_room(
self,
*,
id: str | None = None,
name: str | None = None,
) -> Channel | None:
"""Look up a room by any identifier.
This is an alias for lookup_channel. Use whichever terminology
fits your platform (room for Symphony/Matrix, channel for Slack/Discord).
Args:
id: Room ID.
name: Room name.
Returns:
The room/channel if found, None otherwise.
"""
return await self.lookup_channel(id=id, name=name)
[docs]
@abstractmethod
async def fetch_channel(
self,
identifier: str | Channel | None = None,
*,
id: str | None = None,
name: str | None = None,
) -> Channel | None:
"""Fetch a channel from the backend.
This method accepts flexible input types for convenience:
- Pass a Channel object to validate/refresh it
- Pass an ID string as the first positional argument
- Use keyword arguments for lookup by name
The backend will attempt to resolve the channel using the most
efficient method available for the platform.
Args:
identifier: A Channel object or channel ID string.
id: Channel ID (alternative to positional identifier).
name: Channel name to search for.
Returns:
The channel if found, None otherwise.
Example:
>>> # All of these work:
>>> channel = await backend.fetch_channel("C123456")
>>> channel = await backend.fetch_channel(id="C123456")
>>> channel = await backend.fetch_channel(name="general")
>>> channel = await backend.fetch_channel(existing_channel) # refresh
"""
raise NotImplementedError("Subclass must implement fetch_channel()")
[docs]
async def fetch_room(
self,
identifier: str | Channel | None = None,
*,
id: str | None = None,
name: str | None = None,
) -> Channel | None:
"""Fetch a room from the backend.
This is an alias for fetch_channel. Use whichever terminology
fits your platform (room for Symphony/Matrix, channel for Slack/Discord).
Args:
identifier: A Channel object or room ID string.
id: Room ID (alternative to positional identifier).
name: Room name to search for.
Returns:
The room/channel if found, None otherwise.
"""
return await self.fetch_channel(identifier, id=id, name=name)
# Resolution methods for incomplete objects
[docs]
async def resolve_user(self, user: User) -> User:
"""Resolve an incomplete user to a complete one.
If the user is already complete (has all required fields), returns it as-is.
Otherwise, fetches the full user data from the backend.
Args:
user: A User object that may be incomplete.
Returns:
A complete User object with all fields populated.
Raises:
ValueError: If the user cannot be resolved (no id or name).
Example:
>>> # User from a message might only have id
>>> incomplete_user = message.author
>>> full_user = await backend.resolve_user(incomplete_user)
>>> print(full_user.email) # Now populated
"""
# If already complete and not marked incomplete, return as-is
if user.is_complete and not user.is_incomplete:
return user
# Try to resolve using available identifiers
if user.id:
resolved = await self.fetch_user(id=user.id)
elif user.name:
resolved = await self.fetch_user(name=user.name)
elif hasattr(user, "handle") and user.handle:
resolved = await self.fetch_user(handle=user.handle)
elif hasattr(user, "email") and user.email:
resolved = await self.fetch_user(email=user.email)
else:
raise ValueError("Cannot resolve user: no id, name, handle, or email available")
if resolved:
resolved.mark_complete()
return resolved
# If we couldn't resolve but have partial info, return original
return user
[docs]
async def resolve_channel(self, channel: Channel) -> Channel:
"""Resolve an incomplete channel to a complete one.
If the channel is already complete (has all required fields), returns it as-is.
Otherwise, fetches the full channel data from the backend.
Args:
channel: A Channel object that may be incomplete.
Returns:
A complete Channel object with all fields populated.
Raises:
ValueError: If the channel cannot be resolved (no id or name).
Example:
>>> # Channel from config might only have name
>>> incomplete_channel = Channel(name="general")
>>> full_channel = await backend.resolve_channel(incomplete_channel)
>>> print(full_channel.id) # Now populated
"""
# If already complete and not marked incomplete, return as-is
if channel.is_complete and not channel.is_incomplete:
return channel
# Try to resolve using available identifiers
if channel.id:
resolved = await self.fetch_channel(id=channel.id)
elif channel.name:
resolved = await self.fetch_channel(name=channel.name)
else:
raise ValueError("Cannot resolve channel: no id or name available")
if resolved:
resolved.mark_complete()
return resolved
# If we couldn't resolve but have partial info, return original
return channel
[docs]
async def resolve_room(self, room: Channel) -> Channel:
"""Resolve an incomplete room to a complete one.
This is an alias for resolve_channel. Use whichever terminology
fits your platform (room for Symphony/Matrix, channel for Slack/Discord).
Args:
room: A Channel object that may be incomplete.
Returns:
A complete Channel object with all fields populated.
"""
return await self.resolve_channel(room)
# Organization lookup methods
[docs]
async def fetch_organization(
self,
identifier: str | Organization | None = None,
*,
id: str | None = None,
name: str | None = None,
) -> Organization | None:
"""Fetch an organization from the backend.
An organization is the top-level container (guild, workspace, pod, etc.).
Args:
identifier: An Organization object or organization ID string.
id: Organization ID (alternative to positional identifier).
name: Organization name to search for.
Returns:
The organization if found, None otherwise.
Raises:
NotImplementedError: If the backend doesn't support organizations.
"""
raise NotImplementedError("This backend does not support organizations")
[docs]
async def list_organizations(self) -> list[Organization]:
"""List all organizations the bot has access to.
Returns:
List of organizations.
Raises:
NotImplementedError: If the backend doesn't support organizations.
"""
raise NotImplementedError("This backend does not support organizations")
[docs]
async def fetch_channel_members(
self,
identifier: str | Channel | None = None,
*,
id: str | None = None,
name: str | None = None,
) -> list[User]:
"""Fetch members of a channel.
Retrieves the list of users who are members of the specified channel.
This is useful for authorization checks, mention validation, or
building user interfaces.
This method accepts flexible input types for convenience:
- Pass a Channel object to use its ID
- Pass an ID string as the first positional argument
- Use keyword arguments for lookup by id or name
Args:
identifier: A Channel object or channel ID string.
id: Channel ID (alternative to positional identifier).
name: Channel name to search for.
Returns:
List of users who are members of the channel.
Raises:
NotImplementedError: If the backend doesn't support member listing.
Example:
>>> # All of these work:
>>> members = await backend.fetch_channel_members("C123")
>>> members = await backend.fetch_channel_members(id="C123")
>>> members = await backend.fetch_channel_members(name="general")
>>> members = await backend.fetch_channel_members(channel)
>>> for user in members:
... print(user.name)
"""
raise NotImplementedError("This backend does not support fetching channel members")
[docs]
async def fetch_room_members(
self,
identifier: str | Channel | None = None,
*,
id: str | None = None,
name: str | None = None,
) -> list[User]:
"""Fetch members of a room.
This is an alias for fetch_channel_members. Use whichever terminology
fits your platform.
Args:
identifier: A Channel object or room ID string.
id: Room ID (alternative to positional identifier).
name: Room name to search for.
Returns:
List of users who are members of the room.
"""
return await self.fetch_channel_members(identifier, id=id, name=name)
[docs]
async def resolve_message(self, message: Message) -> Message:
"""Resolve incomplete nested objects in a Message.
Resolves the message's author and channel if they are incomplete.
This is useful when a message is created with partial information
that needs to be filled in before sending.
Args:
message: The message to resolve.
Returns:
The message with resolved author and channel.
Example:
>>> msg = Message(
... content="Hello",
... channel=Channel(name="general"),
... author=User(email="john@example.com"),
... )
>>> resolved = await backend.resolve_message(msg)
>>> print(resolved.channel.id) # Now populated
"""
if message.author and message.author.is_incomplete:
message.author = await self.resolve_user(message.author)
if message.channel and message.channel.is_incomplete:
message.channel = await self.resolve_channel(message.channel)
return message
# Message methods
async def _resolve_channel_id(self, channel: str | Channel) -> str:
"""Helper to resolve a channel argument to an ID string.
Handles string IDs, complete Channel objects, and incomplete
Channel objects that need resolution. For DM channels with users
but no ID, creates or retrieves the DM channel.
Args:
channel: A channel ID string or Channel object.
Returns:
The channel ID string.
Raises:
ValueError: If the channel cannot be resolved.
"""
if isinstance(channel, str):
return channel
if channel.is_complete:
return channel.id
# Handle DM channels with users - create/get the DM
if channel.users and channel.is_dm:
dm_channel_id = await self.create_dm(cast(list[str | User], channel.users))
if dm_channel_id:
return dm_channel_id
raise ValueError(f"Failed to create DM channel with users: {[u.id for u in channel.users]}")
# Resolve incomplete channel by name or other identifiers
resolved = await self.resolve_channel(channel)
return resolved.id
async def _resolve_user_id(self, user: str | User) -> str:
"""Helper to resolve a user argument to an ID string.
Handles string IDs, complete User objects, and incomplete
User objects that need resolution.
Args:
user: A user ID string or User object.
Returns:
The user ID string.
Raises:
ValueError: If the user cannot be resolved.
"""
if isinstance(user, str):
return user
if user.is_complete:
return user.id
# Resolve incomplete user
resolved = await self.resolve_user(user)
return resolved.id
async def _resolve_message_id(self, message: str | Message, channel: str | Channel | None = None) -> tuple[str, str]:
"""Helper to resolve a message argument to (channel_id, message_id).
Handles string message IDs (requires channel), complete Message objects,
and incomplete Message objects.
Args:
message: A message ID string or Message object.
channel: Optional channel (required if message is a string).
Returns:
Tuple of (channel_id, message_id).
Raises:
ValueError: If the message/channel cannot be resolved.
"""
if isinstance(message, str):
if channel is None:
raise ValueError("channel is required when message is a string ID")
channel_id = await self._resolve_channel_id(channel)
return channel_id, message
# It's a Message object
if message.channel:
channel_id = await self._resolve_channel_id(message.channel)
elif channel:
channel_id = await self._resolve_channel_id(channel)
else:
raise ValueError("Message has no channel and no channel was provided")
return channel_id, message.id
[docs]
@abstractmethod
async def fetch_messages(
self,
channel: str | Channel,
limit: int = 100,
before: Union[str, "Message", datetime] | None = None,
after: Union[str, "Message", datetime] | None = None,
) -> list[Message]:
"""Fetch messages from a channel, newest-first.
Returns up to ``limit`` messages, ordered newest-to-oldest. ``before``
and ``after`` bound the range; each accepts a message id, a
:class:`Message`, or a timezone-aware :class:`~datetime.datetime`:
- ``after``: only messages at or after this point (lower bound).
- ``before``: only messages at or before this point (upper bound).
When a range is given, implementations page the underlying API to
cover the whole range without dropping messages (subject to
``limit``); when no range is given, they return the most recent
``limit`` messages.
Args:
channel: The channel to fetch messages from (ID string or Channel object).
limit: Maximum number of messages to return.
before: Upper bound — message id, Message, or datetime.
after: Lower bound — message id, Message, or datetime.
Returns:
List of messages, ordered newest-to-oldest.
Example:
>>> # Most recent 50:
>>> msgs = await backend.fetch_messages("C123", limit=50)
>>> # Everything in the last 30 minutes:
>>> from datetime import datetime, timedelta, timezone
>>> since = datetime.now(timezone.utc) - timedelta(minutes=30)
>>> msgs = await backend.fetch_messages("C123", after=since)
"""
raise NotImplementedError("Subclass must implement fetch_messages()")
[docs]
async def search_messages(
self,
query: str,
channel: str | Channel | None = None,
limit: int = 50,
**kwargs: Any,
) -> list[Message]:
"""Search for messages matching a query.
Searches message content across channels. Requires the MESSAGE_SEARCH
capability to be supported by the backend.
Args:
query: The search query string.
channel: Optional channel to limit search to (ID string or Channel object).
limit: Maximum number of results to return.
**kwargs: Additional platform-specific search options (e.g., from_user,
has_file, date range).
Returns:
List of messages matching the query.
Raises:
NotImplementedError: If the backend doesn't support MESSAGE_SEARCH capability.
Example:
>>> # Search all channels
>>> results = await backend.search_messages("important meeting")
>>> # Search specific channel
>>> results = await backend.search_messages("bug fix", channel="C123")
>>> # With filters
>>> results = await backend.search_messages("report", from_user="U123")
"""
from ..base.capabilities import Capability
if not self.capabilities or Capability.MESSAGE_SEARCH not in self.capabilities.capabilities:
raise NotImplementedError(f"{self.__class__.__name__} does not support message search")
# Default implementation - subclasses should override
raise NotImplementedError("Subclass must implement search_messages()")
[docs]
async def fetch_new_messages(
self,
channel: str | Channel,
after: str | None = None,
) -> list[Message]:
"""Fetch new messages from a channel.
This is a convenience method that fetches messages after a
specific point, typically used for getting updates.
Args:
channel: The channel to fetch messages from (ID string or Channel object).
after: Fetch messages after this message ID.
Returns:
List of new messages.
"""
return await self.fetch_messages(channel=channel, after=after)
@staticmethod
def _extract_thread_id(thread: Any) -> str | None:
"""Extract a thread ID from a ``thread=`` kwarg value.
Accepts ``None``, a ``str`` ID, a :class:`chatom.base.Thread`, or any
:class:`chatom.base.Message` (in which case the message's thread ID
is used if set, otherwise the message's own ID — matching the
"start a thread from this message" idiom).
Returns the thread ID as a string, or ``None`` if no thread was given.
"""
if thread is None:
return None
# Local imports to avoid tight coupling in module-scope types
from ..base import Message as _Msg
from ..base.thread import Thread as _Thread
if isinstance(thread, str):
return thread or None
if isinstance(thread, _Thread):
return thread.id or None
if isinstance(thread, _Msg):
return thread.thread_id or thread.id or None
# Duck-type fallback
return str(getattr(thread, "id", thread)) or None
@staticmethod
def _extract_reply_to_id(reply_to: Any) -> str | None:
"""Extract a message ID from a ``reply_to=`` kwarg value.
Accepts ``None``, a ``str`` ID, or a :class:`chatom.base.Message`.
"""
if reply_to is None:
return None
from ..base import Message as _Msg
if isinstance(reply_to, str):
return reply_to or None
if isinstance(reply_to, _Msg):
return reply_to.id or None
return str(getattr(reply_to, "id", reply_to)) or None
[docs]
@abstractmethod
async def send_message(
self,
channel: str | Channel,
content: str,
**kwargs: Any,
) -> Message:
"""Send a message to a channel.
Standardized optional kwargs (recognized by every backend):
- ``thread``: ``str | Thread | Message | None`` — send into an
existing thread. When a ``Message`` is passed, the message's
thread is used if set, otherwise the message itself becomes the
thread root. Backends translate this to their native concept
(Slack ``thread_ts``, Discord thread channel, Telegram
``message_thread_id``). Symphony has no thread concept and
silently ignores this.
- ``reply_to``: ``str | Message | None`` — reply referencing a
specific message (Discord ``reference=``, Telegram
``reply_to_message_id``, Slack ``thread_ts``). Symphony has no
native reply and silently ignores this.
Args:
channel: The channel to send to (ID string or Channel object).
content: The message content.
**kwargs: Additional platform-specific options (e.g., embeds,
attachments, ``thread``, ``reply_to``).
Returns:
The sent message.
Example:
>>> # All of these work:
>>> msg = await backend.send_message("C123", "Hello!")
>>> msg = await backend.send_message(Channel(id="C123"), "Hello!")
>>> msg = await backend.send_message(Channel(name="general"), "Hello!") # Resolves
>>> # Thread and reply:
>>> msg = await backend.send_message("C123", "In thread", thread=parent_msg)
>>> msg = await backend.send_message("C123", "Replying", reply_to=parent_msg)
"""
raise NotImplementedError("Subclass must implement send_message()")
[docs]
async def edit_message(
self,
message: str | Message,
content: str,
channel: str | Channel | None = None,
**kwargs: Any,
) -> Message:
"""Edit an existing message.
Args:
message: The message to edit (ID string or Message object).
content: The new message content.
channel: The channel containing the message (required if message is a string).
**kwargs: Additional platform-specific options.
Returns:
The edited message.
Raises:
NotImplementedError: If the backend doesn't support editing.
Example:
>>> # Edit using Message object
>>> edited = await backend.edit_message(msg, "Updated content")
>>> # Edit using IDs
>>> edited = await backend.edit_message("M123", "Updated", channel="C123")
"""
raise NotImplementedError("This backend does not support message editing")
[docs]
async def upload_file(
self,
channel: str | Channel,
data: bytes,
filename: str = "file",
content_type: str = "",
title: str = "",
content: str = "",
**kwargs: Any,
) -> Message:
"""Upload a file with binary data to a channel.
Backends should override this to use their native file upload API
(e.g. Slack ``files.uploadV2``, Discord ``File``, Telegram
``send_document`` / ``send_photo``, Symphony attachment API).
The default implementation falls back to ``send_message`` with
a text-only placeholder.
Args:
channel: The channel to upload to (ID string or Channel object).
data: Raw file bytes.
filename: Name of the file.
content_type: MIME type of the file.
title: Optional title for the upload.
content: Optional accompanying text message.
**kwargs: Additional platform-specific options.
Returns:
The sent message.
"""
raise NotImplementedError("This backend does not support file uploads")
[docs]
async def download_attachment(
self,
attachment: Attachment,
*,
message: Message | None = None,
) -> bytes:
"""Download the binary content of an attachment.
Returns the raw bytes of a file/image/document that was received in
chat. The default implementation returns ``attachment.data`` if it
is already populated, otherwise performs an HTTP ``GET`` against
``attachment.url``.
Backends override this to add platform authentication (Slack
``url_private`` bearer token, Telegram ``getFile``, Symphony
attachment API) where a plain public download is not possible.
Args:
attachment: The attachment to download. Must carry either
in-memory ``data``, a downloadable ``url``, or a platform
file ``id`` (for backends that resolve by ID).
message: The message the attachment belongs to. Some backends
(e.g. Symphony) require the message and channel context to
resolve the download.
Returns:
The raw file bytes.
Raises:
NotImplementedError: If the attachment cannot be resolved to a
downloadable source.
Example:
>>> for att in message.attachments:
... data = await backend.download_attachment(att, message=message)
... Path(att.filename).write_bytes(data)
"""
if attachment.data is not None:
return attachment.data
url = (attachment.url or "").strip()
if url:
return await self._download_url(url)
raise NotImplementedError(
f"{self.__class__.__name__} cannot download attachment {attachment.id or attachment.filename!r}: no data or url available"
)
async def _download_url(self, url: str, headers: dict | None = None) -> bytes:
"""Download bytes from an ``http(s)`` URL in a worker thread.
Only ``http`` and ``https`` schemes are allowed to avoid local-file
and other SSRF vectors (e.g. ``file://``).
Args:
url: The URL to download.
headers: Optional request headers (e.g. an auth bearer token).
Returns:
The downloaded bytes.
"""
import urllib.request
from urllib.parse import urlparse
scheme = urlparse(url).scheme.lower()
if scheme not in ("http", "https"):
raise ValueError(f"Refusing to download non-http(s) URL: {url!r}")
def _get() -> bytes:
req = urllib.request.Request(url, headers=headers or {})
with urllib.request.urlopen(req, timeout=30) as resp:
return resp.read()
loop = asyncio.get_event_loop()
return await loop.run_in_executor(None, _get)
[docs]
async def delete_message(
self,
message: str | Message,
channel: str | Channel | None = None,
) -> None:
"""Delete a message.
Args:
message: The message to delete (ID string or Message object).
channel: The channel containing the message (required if message is a string).
Raises:
NotImplementedError: If the backend doesn't support deletion.
Example:
>>> # Delete using Message object
>>> await backend.delete_message(msg)
>>> # Delete using IDs
>>> await backend.delete_message("M123", channel="C123")
"""
raise NotImplementedError("This backend does not support message deletion")
[docs]
async def reply_in_thread(
self,
message: Message,
content: str,
**kwargs: Any,
) -> Message:
"""Reply to a message in its thread.
This is a convenience method that sends a reply in the thread
of the given message. If the message is not in a thread, it
creates a new thread from that message.
This method simplifies thread-based conversations for bot developers
by abstracting the platform-specific details of thread handling.
Args:
message: The message to reply to. Thread context is extracted
from message.thread_id or message.id.
content: The reply content.
**kwargs: Additional platform-specific options (e.g., embeds,
attachments).
Returns:
The sent reply message.
Raises:
NotImplementedError: If the backend doesn't support threading.
Example:
>>> # Reply in a thread
>>> async def on_message(message: Message):
... reply = await backend.reply_in_thread(
... message,
... "Thanks for your message!"
... )
"""
# Default implementation uses send_message with thread_id
# Get the thread ID: either the message's thread or start a new thread
thread_id = message.thread_id or message.id
channel = message.channel
if not channel and not message.channel_id:
raise ValueError("Cannot reply: message has no channel information")
# Use channel object if available, otherwise use channel_id
target_channel: str | Channel = channel if channel else message.channel_id
return await self.send_message(
channel=target_channel,
content=content,
thread_id=thread_id,
**kwargs,
)
[docs]
async def forward_message(
self,
message: Message,
to_channel: str | Channel,
*,
include_attribution: bool = True,
prefix: str | None = None,
**kwargs: Any,
) -> Message:
"""Forward a message to another channel.
This method forwards a message from one channel to another,
optionally including attribution about the original source.
The forwarded message can include the original content,
attachments, and optionally embeds.
Different backends handle forwarding differently:
- Some have native forwarding (Discord has reference links)
- Others simulate it by re-sending with attribution
Args:
message: The message to forward.
to_channel: The destination channel (ID string or Channel object).
include_attribution: If True (default), include information about
the original message source (author, channel, time).
prefix: Optional text to prepend to the forwarded message.
**kwargs: Additional platform-specific options (e.g., embeds,
thread_id for threading).
Returns:
The forwarded message in the destination channel.
Raises:
NotImplementedError: If the backend doesn't support forwarding.
ValueError: If the message or channel cannot be resolved.
Example:
>>> # Forward a message to another channel
>>> forwarded = await backend.forward_message(
... incoming_message,
... to_channel="C456789",
... )
>>>
>>> # Forward with custom prefix
>>> forwarded = await backend.forward_message(
... incoming_message,
... to_channel=Channel(name="alerts"),
... prefix="⚠️ Escalated: ",
... )
>>>
>>> # Forward without attribution
>>> forwarded = await backend.forward_message(
... incoming_message,
... to_channel="C456789",
... include_attribution=False,
... )
"""
raise NotImplementedError("This backend does not support message forwarding")
# Real-time message streaming
[docs]
async def listen(
self,
channel: str | Channel | None = None,
skip_own: bool = True,
) -> AsyncIterator[Message]:
"""Listen for incoming messages in real-time.
This async generator yields Message objects as they arrive from
the chat platform. It abstracts away the platform-specific
details of real-time messaging (WebSockets, Socket Mode, datafeed, etc.).
This is an alias for stream_messages() with skip_history=True.
Args:
channel: Optional channel to filter messages to a specific
channel (ID string or Channel object). If None, yields
messages from all channels the bot has access to.
skip_own: If True (default), skip messages sent by the bot itself.
Yields:
Message: Each message as it arrives.
Raises:
NotImplementedError: If the backend doesn't support streaming.
ConnectionError: If the real-time connection fails.
Example:
>>> async for message in backend.listen():
... print(f"Received: {message.content}")
... if "!ping" in message.content:
... await backend.reply_in_thread(message, "Pong!")
"""
async for message in self.stream_messages(
channel=channel,
skip_own=skip_own,
skip_history=True,
):
yield message
[docs]
async def stream_messages(
self,
channel: str | Channel | None = None,
skip_own: bool = True,
skip_history: bool = True,
) -> AsyncIterator[Message]:
"""Stream incoming messages in real-time.
This async generator yields Message objects as they arrive from
the chat platform. It abstracts away the platform-specific
details of real-time messaging (WebSockets, Socket Mode, datafeed, etc.).
The stream continues until the generator is closed or an error occurs.
Args:
channel: Optional channel to filter messages to a specific
channel (ID string or Channel object). If None, yields
messages from all channels the bot has access to.
skip_own: If True (default), skip messages sent by the bot itself.
skip_history: If True (default), skip messages that existed before
the stream started. Only yields new messages.
Yields:
Message: Each message as it arrives.
Raises:
NotImplementedError: If the backend doesn't support streaming.
ConnectionError: If the real-time connection fails.
Example:
>>> async for message in backend.stream_messages():
... print(f"Received: {message.content}")
... if message.mentions_bot:
... await backend.reply_in_thread(message, "Hello!")
>>>
>>> # Filter to a specific channel
>>> async for message in backend.stream_messages(channel="C123"):
... await process_message(message)
>>>
>>> # Filter using Channel object
>>> async for message in backend.stream_messages(channel=Channel(name="general")):
... await process_message(message)
"""
raise NotImplementedError("This backend does not support message streaming")
# This is needed for type checking, but won't be reached
yield
[docs]
async def stream_interactions(
self,
channel: str | Channel | None = None,
) -> AsyncIterator[Interaction]:
"""Stream incoming component interactions in real-time.
Yields :class:`~chatom.base.Interaction` objects each time a user
clicks a button, picks from a select menu, or submits a modal
that was sent via this backend.
Backends that don't natively support interactive components, or
where interaction streaming hasn't been implemented yet, should
leave this as ``NotImplementedError``.
Args:
channel: Optional channel filter.
Yields:
Interaction: Each component interaction as it arrives.
Raises:
NotImplementedError: If the backend doesn't support
interaction streaming.
Example:
>>> async for event in backend.stream_interactions():
... if event.action_id == "confirm":
... await handle_confirm(event)
"""
raise NotImplementedError("This backend does not support interaction streaming")
yield # pragma: no cover
[docs]
async def read_messages(
self,
channel: str | Channel,
limit: int = 100,
before: str | None = None,
after: str | None = None,
) -> AsyncIterator[Message]:
"""Read message history from a channel as an async iterator.
This is a convenience async generator that yields messages from
a channel's history one at a time. It wraps fetch_messages for
easier iteration.
Args:
channel: The channel to read messages from (ID string or Channel object).
limit: Maximum number of messages to read.
before: Read messages before this message ID (for pagination).
after: Read messages after this message ID (for pagination).
Yields:
Message: Each message from the history.
Example:
>>> async for message in backend.read_messages("C123", limit=50):
... print(f"{message.author.name}: {message.content}")
"""
messages = await self.fetch_messages(
channel=channel,
limit=limit,
before=before,
after=after,
)
for message in messages:
yield message
[docs]
async def read_thread(
self,
channel: str | Channel,
thread_id: str,
limit: int = 100,
) -> AsyncIterator[Message]:
"""Read messages from a thread.
This async generator yields messages from a thread/conversation
one at a time. Threads are platform-specific:
- Slack: message thread (thread_ts)
- Discord: thread channel
- Symphony: reply chain
Args:
channel: The parent channel containing the thread (ID string or Channel object).
thread_id: The thread identifier (message ID that started the thread).
limit: Maximum number of messages to read.
Yields:
Message: Each message from the thread.
Raises:
NotImplementedError: If the backend doesn't support threads.
Example:
>>> async for message in backend.read_thread("C123", "1234567890.123456"):
... print(f"{message.author.name}: {message.content}")
"""
raise NotImplementedError("This backend does not support reading threads")
yield
[docs]
async def create_thread(
self,
channel: str | Channel,
message_id: str,
name: str,
**kwargs: Any,
) -> Channel:
"""Create a thread from a message.
Creates a new thread attached to the specified message.
This is primarily used for Discord-style threads where a
thread is a separate channel derived from a message.
For Slack-style threads that are implicit (just replies to
a message), use reply_in_thread() instead.
Args:
channel: The channel containing the message (ID string or Channel object).
message_id: The ID of the message to create a thread from.
name: The name/title for the thread.
**kwargs: Additional platform-specific options:
- auto_archive_duration: Discord thread archive duration in minutes.
- type: Discord thread type (public, private, announcement).
- reason: Audit log reason.
Returns:
The created thread as a Channel object.
Raises:
NotImplementedError: If the backend doesn't support thread creation.
Example:
>>> thread = await backend.create_thread(
... channel="C123",
... message_id="M456",
... name="Discussion Thread",
... )
>>> await backend.send_message(thread.id, "First message in thread!")
"""
raise NotImplementedError("This backend does not support thread creation")
[docs]
async def reply_to_message(
self,
channel: str | Channel,
message_id: str,
content: str,
**kwargs: Any,
) -> Message:
"""Reply to a specific message.
Sends a reply that references the original message. How this
is displayed depends on the platform:
- Slack: In thread (if thread support) or as inline reply
- Discord: As a reply with reference link
- Symphony: As a reply to the message
For thread-based replies, consider using reply_in_thread() instead
which handles the thread context automatically.
Args:
channel: The channel containing the message (ID string or Channel object).
message_id: The ID of the message to reply to.
content: The reply content.
**kwargs: Additional platform-specific options.
Returns:
The sent reply message.
Raises:
NotImplementedError: If the backend doesn't support replies.
Example:
>>> reply = await backend.reply_to_message(
... channel="C123",
... message_id="M456",
... content="Thanks for your message!",
... )
"""
# Default implementation: try to use reply_in_thread with a reconstructed message
message = Message(
id=message_id,
content="",
channel=Channel(id=await self._resolve_channel_id(channel)) if isinstance(channel, Channel) else Channel(id=channel),
)
return await self.reply_in_thread(message, content, **kwargs)
[docs]
async def get_bot_info(self) -> User | None:
"""Get information about the connected bot user.
Returns the User object representing the bot/service account
that is currently connected. This is useful for checking if
messages mention the bot.
Returns:
The bot's User object, or None if not available.
Raises:
NotImplementedError: If the backend doesn't support this.
Example:
>>> bot = await backend.get_bot_info()
>>> print(f"Connected as: {bot.name} ({bot.id})")
"""
raise NotImplementedError("This backend does not support get_bot_info")
# Presence methods
[docs]
async def set_presence(
self,
status: str,
status_text: str | None = None,
**kwargs: Any,
) -> None:
"""Set the current user's presence status.
Args:
status: The presence status (e.g., 'online', 'away', 'dnd').
status_text: Optional status message/text.
**kwargs: Additional platform-specific options.
Raises:
NotImplementedError: If the backend doesn't support presence.
"""
raise NotImplementedError("This backend does not support presence")
[docs]
async def get_presence(self, user: str | User) -> Presence | None:
"""Get a user's presence status.
Args:
user: The user to get presence for (ID string or User object).
Returns:
The user's presence, or None if not available.
Raises:
NotImplementedError: If the backend doesn't support presence.
Example:
>>> # All of these work:
>>> presence = await backend.get_presence("U123")
>>> presence = await backend.get_presence(User(id="U123"))
>>> presence = await backend.get_presence(User(email="john@example.com")) # Resolves
"""
raise NotImplementedError("This backend does not support presence")
[docs]
def start_presence_heartbeat(
self,
interval_seconds: int = 60,
status: str = "online",
status_text: str | None = None,
) -> None:
"""Start an automatic presence heartbeat.
This method periodically sets the user's presence to keep
the bot appearing online. This is useful for platforms that
require regular presence updates or for bots that need to
maintain an active status.
The heartbeat runs in the background and can be stopped with
stop_presence_heartbeat().
Args:
interval_seconds: How often to send presence updates (default 60).
status: The presence status to set (default 'online').
status_text: Optional status message/text to display.
Example:
>>> # Start keeping the bot online
>>> backend.start_presence_heartbeat(60, "online", "Ready to help!")
>>>
>>> # Later, stop the heartbeat
>>> backend.stop_presence_heartbeat()
"""
# Stop any existing heartbeat
self.stop_presence_heartbeat()
async def _heartbeat_loop() -> None:
while self._presence_heartbeat_running:
try:
await self.set_presence(status, status_text)
except Exception: # noqa: BLE001, S110
# Ignore errors, just keep trying
pass
await asyncio.sleep(interval_seconds)
self._presence_heartbeat_running = True
self._presence_heartbeat_task = asyncio.create_task(_heartbeat_loop())
[docs]
def stop_presence_heartbeat(self) -> None:
"""Stop the automatic presence heartbeat.
Cancels any running presence heartbeat task started by
start_presence_heartbeat().
Example:
>>> backend.stop_presence_heartbeat()
"""
self._presence_heartbeat_running = False
if self._presence_heartbeat_task is not None:
self._presence_heartbeat_task.cancel()
self._presence_heartbeat_task = None
@property
def is_presence_heartbeat_active(self) -> bool:
"""Check if the presence heartbeat is currently running.
Returns:
bool: True if the heartbeat is active.
"""
return self._presence_heartbeat_running
# Reaction methods
[docs]
async def add_reaction(
self,
message: str | Message,
emoji: str,
channel: str | Channel | None = None,
) -> None:
"""Add a reaction to a message.
Args:
message: The message to react to (ID string or Message object).
emoji: The emoji to add (name or unicode).
channel: The channel containing the message (required if message is a string).
Raises:
NotImplementedError: If the backend doesn't support reactions.
Example:
>>> # React using Message object
>>> await backend.add_reaction(msg, "👍")
>>> # React using IDs
>>> await backend.add_reaction("M123", "👍", channel="C123")
"""
raise NotImplementedError("This backend does not support reactions")
[docs]
async def remove_reaction(
self,
message: str | Message,
emoji: str,
channel: str | Channel | None = None,
) -> None:
"""Remove a reaction from a message.
Args:
message: The message to remove reaction from (ID string or Message object).
emoji: The emoji to remove (name or unicode).
channel: The channel containing the message (required if message is a string).
Raises:
NotImplementedError: If the backend doesn't support reactions.
Example:
>>> # Remove reaction using Message object
>>> await backend.remove_reaction(msg, "👍")
>>> # Remove reaction using IDs
>>> await backend.remove_reaction("M123", "👍", channel="C123")
"""
raise NotImplementedError("This backend does not support reactions")
# Channel/Room management methods
[docs]
async def create_dm(
self,
users: list[str | User],
) -> str | None:
"""Create a direct message (DM) or instant message (IM) channel.
Creates a private conversation with one or more users.
For single users, this creates a 1:1 DM. For multiple users,
this may create a group DM/MIM depending on the platform.
Args:
users: List of users to include in the DM (ID strings or User objects).
Returns:
The channel/stream ID of the created DM, or None if failed.
Raises:
NotImplementedError: If the backend doesn't support DM creation.
Example:
>>> # Create DM with user IDs
>>> dm_id = await backend.create_dm(["U123", "U456"])
>>> # Create DM with User objects
>>> dm_id = await backend.create_dm([User(id="U123")])
>>> # Create DM with incomplete User (will resolve)
>>> dm_id = await backend.create_dm([User(email="john@example.com")])
"""
raise NotImplementedError("This backend does not support DM creation")
[docs]
async def create_im(
self,
users: list[str | User],
) -> str | None:
"""Create an instant message (IM) channel.
This is an alias for create_dm. Use whichever terminology
fits your platform (IM for Symphony, DM for Discord/Slack).
Args:
users: List of users to include in the IM (ID strings or User objects).
Returns:
The channel/stream ID of the created IM, or None if failed.
"""
return await self.create_dm(users)
[docs]
async def send_dm(
self,
user: str | User,
content: str,
**kwargs: Any,
) -> Message:
"""Send a direct message to a user.
This is a convenience method that creates a DM channel with the user
if needed, then sends the message. It simplifies the common pattern of:
dm_id = await backend.create_dm([user])
await backend.send_message(dm_id, content)
Args:
user: The user to send to (ID string or User object).
content: The message content.
**kwargs: Additional platform-specific options passed to send_message.
Returns:
The sent message.
Raises:
ValueError: If the DM channel could not be created.
NotImplementedError: If the backend doesn't support DM creation.
Example:
>>> # Send DM using user ID
>>> msg = await backend.send_dm("U123", "Hello!")
>>> # Send DM using User object
>>> msg = await backend.send_dm(user, "Hello!")
>>> # With additional options
>>> msg = await backend.send_dm(user, "Check this!", thread_id="T123")
"""
# Normalize to list for create_dm
user_list = [user]
# Create or get existing DM channel
dm_channel_id = await self.create_dm(user_list)
if not dm_channel_id:
user_id = user.id if isinstance(user, User) else user
raise ValueError(f"Failed to create DM channel with user {user_id}")
# Send the message
return await self.send_message(
channel=dm_channel_id,
content=content,
**kwargs,
)
[docs]
async def create_channel(
self,
name: str,
description: str = "",
public: bool = True,
**kwargs: Any,
) -> str | None:
"""Create a new channel.
Creates a channel/room for group communication.
Args:
name: The channel name.
description: Optional channel description/purpose.
public: Whether the channel is public (default True).
**kwargs: Additional platform-specific options:
- read_only: Whether the channel is read-only (Symphony).
- topic: Channel topic (Slack).
- category_id: Category to create under (Discord).
Returns:
The channel ID of the created channel, or None if failed.
Raises:
NotImplementedError: If the backend doesn't support channel creation.
"""
raise NotImplementedError("This backend does not support channel creation")
[docs]
async def create_room(
self,
name: str,
description: str = "",
public: bool = True,
**kwargs: Any,
) -> str | None:
"""Create a new room.
This is an alias for create_channel. Use whichever terminology
fits your platform (room for Symphony/Matrix, channel for Slack/Discord).
Args:
name: The room name.
description: Optional room description.
public: Whether the room is public (default True).
**kwargs: Additional platform-specific options.
Returns:
The room/stream ID of the created room, or None if failed.
"""
return await self.create_channel(name, description, public, **kwargs)
[docs]
async def join_channel(
self,
channel: str | Channel,
**kwargs: Any,
) -> None:
"""Join a channel.
Makes the bot/user a member of the specified channel.
Args:
channel: The channel to join (ID string or Channel object).
**kwargs: Additional platform-specific options:
- key: Channel password/key (IRC).
- invite_code: Invite code (Discord).
Raises:
NotImplementedError: If the backend doesn't support joining channels.
Example:
>>> # Join using channel ID
>>> await backend.join_channel("C123")
>>> # Join using Channel object
>>> await backend.join_channel(Channel(name="general"))
"""
raise NotImplementedError("This backend does not support joining channels")
[docs]
async def join_room(
self,
room: str | Channel,
**kwargs: Any,
) -> None:
"""Join a room.
This is an alias for join_channel. Use whichever terminology
fits your platform.
Args:
room: The room to join (ID string or Channel object).
**kwargs: Additional platform-specific options.
"""
return await self.join_channel(room, **kwargs)
[docs]
async def leave_channel(
self,
channel: str | Channel,
**kwargs: Any,
) -> None:
"""Leave a channel.
Removes the bot/user from the specified channel.
Args:
channel: The channel to leave (ID string or Channel object).
**kwargs: Additional platform-specific options:
- message: Part message (IRC).
Raises:
NotImplementedError: If the backend doesn't support leaving channels.
Example:
>>> # Leave using channel ID
>>> await backend.leave_channel("C123")
>>> # Leave using Channel object
>>> await backend.leave_channel(Channel(id="C123", name="general"))
"""
raise NotImplementedError("This backend does not support leaving channels")
[docs]
async def leave_room(
self,
room: str | Channel,
**kwargs: Any,
) -> None:
"""Leave a room.
This is an alias for leave_channel. Use whichever terminology
fits your platform.
Args:
room: The room to leave (ID string or Channel object).
**kwargs: Additional platform-specific options.
"""
return await self.leave_channel(room, **kwargs)
# Extended messaging methods
[docs]
async def send_action(
self,
target: str | Channel | User,
action: str,
) -> None:
"""Send an action/emote message.
Sends an action message (like IRC's /me command).
On IRC this is a CTCP ACTION. On other platforms,
this may be formatted as italicized text or similar.
Args:
target: The channel or user to send to (ID string, Channel, or User).
action: The action text (e.g., "waves hello").
Raises:
NotImplementedError: If the backend doesn't support actions.
Example:
>>> # Send action to channel
>>> await backend.send_action(Channel(name="#general"), "waves hello")
>>> # Send action to user
>>> await backend.send_action(User(id="U123"), "waves hello")
"""
raise NotImplementedError("This backend does not support action messages")
[docs]
async def send_notice(
self,
target: str | Channel | User,
text: str,
) -> None:
"""Send a notice message.
Sends a notice (typically displayed differently from regular messages).
On IRC this is a NOTICE. Other platforms may not distinguish notices.
Args:
target: The channel or user to send to (ID string, Channel, or User).
text: The notice text.
Raises:
NotImplementedError: If the backend doesn't support notices.
Example:
>>> # Send notice to channel
>>> await backend.send_notice(Channel(name="#general"), "Server maintenance")
>>> # Send notice to user
>>> await backend.send_notice(User(id="U123"), "You have been warned")
"""
raise NotImplementedError("This backend does not support notice messages")
# Mention methods
[docs]
def mention_user(self, user: User) -> str:
"""Format a user mention for this backend.
Args:
user: The user to mention.
Returns:
The formatted mention string.
"""
from ..base.mention import mention_user_for_backend
return mention_user_for_backend(user, self.__class__.name)
[docs]
def mention_channel(self, channel: Channel) -> str:
"""Format a channel mention for this backend.
Args:
channel: The channel to mention.
Returns:
The formatted mention string.
"""
from ..base.mention import mention_channel_for_backend
return mention_channel_for_backend(channel, self.__class__.name)
[docs]
def channel_link(self, channel: str | Channel) -> str:
"""Generate a clickable channel link/mention for this backend.
This is a convenience method that accepts either a channel ID string
or a Channel object and returns the appropriate platform-specific
channel reference.
Args:
channel: A channel ID string or Channel object.
Returns:
The formatted channel link/mention string.
Example:
>>> # From channel ID
>>> link = backend.channel_link("C123")
>>> # From Channel object
>>> link = backend.channel_link(channel)
>>> # Use in message
>>> await backend.send_message(ch, f"Join us in {backend.channel_link('general')}")
"""
if isinstance(channel, str):
# Create a minimal Channel object with just the ID
channel = Channel(id=channel)
return self.mention_channel(channel)
[docs]
def mention_here(self) -> str:
"""Format an @here mention for this backend.
Notifies all users who are currently active/online in the channel.
Subclasses should override this with platform-specific format.
Returns:
The formatted @here mention string.
"""
return "@here"
[docs]
def mention_everyone(self) -> str:
"""Format an @everyone mention for this backend.
Notifies all members of the channel/server.
Subclasses should override this with platform-specific format.
Returns:
The formatted @everyone mention string.
"""
return "@everyone"
[docs]
def mention_channel_all(self) -> str:
"""Format an @channel mention for this backend.
Notifies all members of the current channel (Slack-specific concept).
For platforms without this distinction, defaults to @everyone.
Subclasses should override this with platform-specific format.
Returns:
The formatted @channel mention string.
"""
return self.mention_everyone()
def __repr__(self) -> str:
return f"{self.__class__.__name__}(connected={self.connected})"
# Alias for convenience
Backend = BackendBase