SignalBot

LOGGER_NAME module-attribute

LOGGER_NAME = 'signalbot'

The logger name used by signalbot.

MIN_SIGNAL_CLI_REST_API_VERSION module-attribute

MIN_SIGNAL_CLI_REST_API_VERSION = Version('0.95.0')

The minimum required version of signal-cli-rest-api for this version of signalbot.

SignalBot

SignalBot(config: Config | Mapping | Path | str)

SignalBot is the main class for the bot. It provides methods to register commands, start the bot, and interact with messages.

Attributes:

Name	Type	Description
config	Config	The configuration for the bot.
commands	CommandList	A list of registered commands with their filters. Only available after .start() is called and init_task is done.
groups	list	A list of groups the bot is a member of. Only available after .start() is called and init_task is done.
storage	SQLiteStorage | RedisStorage	The storage backend used by the bot.
scheduler	AsyncIOScheduler	The scheduler for running scheduled tasks.
init_task	None | Task	The initialization async task for the bot. Only available after .start() is called.

Parameters:

Name	Type	Description	Default
config	Config | Mapping | Path | str	the configuration for the bot.	required

Example config:

{
    signal_service: "127.0.0.1:8080",
    phone_number: "+49123456789"
}

Source code in src/signalbot/bot.py

def __init__(self, config: Config | Mapping | Path | str) -> None:
    """Initilization for the SignalBot.

    Args:
        config: the configuration for the bot.

    Example config:
    ```python
    {
        signal_service: "127.0.0.1:8080",
        phone_number: "+49123456789"
    }
    ```
    """
    self._logger = logging.getLogger(LOGGER_NAME)

    self.config = load_config(config)

    self._commands_to_be_registered: CommandList = []  # populated by .register()
    self.commands: CommandList = []  # populated by .start()

    self.groups = []  # populated by .start()
    self._groups_by_id = {}
    self._groups_by_internal_id = {}
    self._groups_by_name = defaultdict(list)

    self.init_task: None | asyncio.Task = None

    try:
        self._signal = SignalAPI(
            self.config.signal_service,
            self.config.phone_number,
            self.config.download_attachments,
            self.config.connection_mode,
        )
    except KeyError:
        raise SignalBotError("Could not initialize SignalAPI with given config")  # noqa: B904, EM101, TRY003

    try:
        self._event_loop = asyncio.get_event_loop()
    except RuntimeError:
        self._event_loop = asyncio.new_event_loop()
        asyncio.set_event_loop(self._event_loop)

    self._q = asyncio.Queue()

    self._produce_tasks: set[asyncio.Task] = set()
    self._consume_tasks: set[asyncio.Task] = set()

    try:
        self.scheduler = AsyncIOScheduler(event_loop=self._event_loop)
    except Exception as e:  # noqa: BLE001
        raise SignalBotError(f"Could not initialize scheduler: {e}")  # noqa: B904, EM102, TRY003

    if isinstance(self.config.storage, SQLiteConfig):
        self.storage = SQLiteStorage(
            self.config.storage.sqlite_db,
            check_same_thread=self.config.storage.check_same_thread,
        )
        self._logger.info("sqlite storage initilized")
    elif isinstance(self.config.storage, RedisConfig):
        self.storage = RedisStorage(
            self.config.storage.redis_host, self.config.storage.redis_port
        )
        self._logger.info("redis storage initilized")
    elif isinstance(self.config.storage, InMemoryConfig):
        self.storage = SQLiteStorage()
        self._logger.info("in-memory storage initilized")
    else:
        self.storage = SQLiteStorage()
        self._logger.warning(
            " Using in-memory storage."
            " Restarting will delete the storage!"
            " Add storage: {'type': 'in-memory'}"
            " to the config to silence this error.",
        )

delete_attachment async

delete_attachment(attachment_filename: str) -> None

Delete an attachment from local storage.

Parameters:

Name	Type	Description	Default
attachment_filename	str	File name to delete.	required

Source code in src/signalbot/bot.py

async def delete_attachment(self, attachment_filename: str) -> None:
    """Delete an attachment from local storage.

    Args:
        attachment_filename: File name to delete.
    """
    await self._signal.delete_attachment(attachment_filename)

poll async

poll(
    receiver: str,
    question: str,
    answers: list[str],
    *,
    allow_multiple_selections: bool = False,
) -> int

Create a poll.

Parameters:

Name	Type	Description	Default
receiver	str	The recipient of the message.	required
question	str	The poll question.	required
answers	list[str]	List of answer options for the poll.	required
allow_multiple_selections	bool	Whether multiple answers can be selected.	False

Returns:

Type	Description
int	The timestamp the poll was created.

Source code in src/signalbot/bot.py

async def poll(
    self,
    receiver: str,
    question: str,
    answers: list[str],
    *,
    allow_multiple_selections: bool = False,
) -> int:
    """Create a poll.

    Args:
        receiver: The recipient of the message.
        question: The poll question.
        answers: List of answer options for the poll.
        allow_multiple_selections: Whether multiple answers can be selected.

    Returns:
        The timestamp the poll was created.
    """
    receiver = self._resolve_receiver(receiver)

    resp = await self._signal.poll(
        receiver,
        question,
        answers,
        allow_multiple_selections=allow_multiple_selections,
    )
    resp_payload = await resp.json()
    timestamp = int(resp_payload["timestamp"])
    self._logger.info("[Bot] New poll created:\n%s", question)

    return timestamp

react async

react(message: Message, emoji: str) -> None

React to a message with an emoji.

Parameters:

Name	Type	Description	Default
message	Message	The message to react to.	required
emoji	str	Emoji reaction value.	required

Source code in src/signalbot/bot.py

async def react(self, message: Message, emoji: str) -> None:
    """React to a message with an emoji.

    Args:
        message: The message to react to.
        emoji: Emoji reaction value.
    """
    # TODO: check that emoji is really an emoji  # noqa: TD002, TD003
    recipient = message.recipient()
    recipient = self._resolve_receiver(recipient)
    target_author = message.source
    timestamp = message.timestamp
    await self._signal.react(recipient, emoji, target_author, timestamp)
    self._logger.info(f"[Bot] New reaction: {emoji}")  # noqa: G004

receipt async

receipt(
    message: Message,
    receipt_type: Literal["read", "viewed"],
) -> None

Send a read or viewed receipt for a message if supported.

Parameters:

Name	Type	Description	Default
message	Message	The message to acknowledge.	required
receipt_type	Literal['read', 'viewed']	Receipt type to send.	required

Source code in src/signalbot/bot.py

async def receipt(
    self,
    message: Message,
    receipt_type: Literal["read", "viewed"],
) -> None:
    """Send a read or viewed receipt for a message if supported.

    Args:
        message: The message to acknowledge.
        receipt_type: Receipt type to send.
    """
    if message.group is not None:
        self._logger.warning("[Bot] Receipts are not supported for groups")
        return

    recipient = self._resolve_receiver(message.recipient())
    await self._signal.receipt(recipient, receipt_type, message.timestamp)
    self._logger.info(f"[Bot] Receipt: {receipt_type}")  # noqa: G004

register

register(
    command: Command,
    contacts: list[str] | bool = True,
    groups: list[str] | bool = True,
    f: Callable[[Message], bool] | None = None,
) -> None

Register a command with optional contact/group filters.

Parameters:

Name	Type	Description	Default
command	Command	Command instance to register.	required
contacts	list[str] | bool	Allowed contacts or True for all.	True
groups	list[str] | bool	Allowed groups or True for all.	True
f	Callable[[Message], bool] | None	Optional function to further filter messages.	None

Source code in src/signalbot/bot.py

def register(
    self,
    command: Command,
    contacts: list[str] | bool = True,  # noqa: FBT001, FBT002
    groups: list[str] | bool = True,  # noqa: FBT001, FBT002
    f: Callable[[Message], bool] | None = None,
) -> None:
    """Register a command with optional contact/group filters.

    Args:
        command: Command instance to register.
        contacts: Allowed contacts or True for all.
        groups: Allowed groups or True for all.
        f: Optional function to further filter messages.
    """
    command.bot = self
    command.setup()
    self._commands_to_be_registered.append((command, contacts, groups, f))

remote_delete async

remote_delete(receiver: str, timestamp: int) -> int

Delete a previously sent message.

Parameters:

Name	Type	Description	Default
receiver	str	Recipient identifier.	required
timestamp	int	Timestamp of the message to delete.	required

Returns:

Type	Description
int	The timestamp of the delete action.

Source code in src/signalbot/bot.py

async def remote_delete(self, receiver: str, timestamp: int) -> int:
    """Delete a previously sent message.

    Args:
        receiver: Recipient identifier.
        timestamp: Timestamp of the message to delete.

    Returns:
        The timestamp of the delete action.
    """
    receiver = self._resolve_receiver(receiver)

    resp = await self._signal.remote_delete(
        receiver,
        timestamp=timestamp,
    )
    resp_payload = await resp.json()
    ret_timestamp = int(resp_payload["timestamp"])
    self._logger.info(f"[Bot] Deleted message with timestamp {timestamp}")  # noqa: G004

    return ret_timestamp

send async

send(
    receiver: str,
    text: str,
    *,
    base64_attachments: list | None = None,
    link_preview: LinkPreview | None = None,
    quote_author: str | None = None,
    quote_mentions: list | None = None,
    quote_message: str | None = None,
    quote_timestamp: int | None = None,
    mentions: list[dict[str, Any]] | None = None,
    edit_timestamp: int | None = None,
    text_mode: str | None = None,
    view_once: bool = False,
) -> int

Send or edit a message.

Parameters:

Name	Type	Description	Default
receiver	str	The recipient of the message.	required
text	str	The content of the message.	required
base64_attachments	list | None	List of attachments encoded in base64.	None
link_preview	LinkPreview | None	Link previews to be sent with the message.	None
quote_author	str | None	The author of the quoted message, required if quote_message is set.	None
quote_mentions	list | None	List of mentioned users in the quoted message, required if quote_message is set.	None
quote_message	str | None	The content of the quoted message, required if quote_message is set.	None
quote_timestamp	int | None	The timestamp of the quoted message, required if quote_message is set.	None
mentions	list[dict[str, Any]] | None	List of dictionary of mentions, it has the format [{ "author": "uuid" , "start": 0, "length": 1 }].	None
edit_timestamp	int | None	The timestamp of the message to edit, if not set a new message will be sent.	None
text_mode	str | None	The text mode of the message, can be "normal" or "styled".	None
view_once	bool	Whether the message should be view once or not.	False

Returns:

Type	Description
int	The timestamp of the sent or edited message.

Source code in src/signalbot/bot.py

async def send(  # noqa: PLR0913
    self,
    receiver: str,
    text: str,
    *,
    base64_attachments: list | None = None,
    link_preview: LinkPreview | None = None,
    quote_author: str | None = None,
    quote_mentions: list | None = None,
    quote_message: str | None = None,
    quote_timestamp: int | None = None,
    mentions: (
        list[dict[str, Any]] | None
    ) = None,  # [{ "author": "uuid" , "start": 0, "length": 1 }]
    edit_timestamp: int | None = None,
    text_mode: str | None = None,
    view_once: bool = False,
) -> int:
    """Send or edit a message.

    Args:
        receiver: The recipient of the message.
        text: The content of the message.
        base64_attachments: List of attachments encoded in base64.
        link_preview: Link previews to be sent with the message.
        quote_author: The author of the quoted message, required if quote_message is
            set.
        quote_mentions: List of mentioned users in the quoted message, required if
            quote_message is set.
        quote_message: The content of the quoted message, required if quote_message
            is set.
        quote_timestamp: The timestamp of the quoted message, required if
            quote_message is set.
        mentions: List of dictionary of mentions, it has the format
            `[{ "author": "uuid" , "start": 0, "length": 1 }]`.
        edit_timestamp: The timestamp of the message to edit, if not set a new
            message will be sent.
        text_mode: The text mode of the message, can be "normal" or "styled".
        view_once: Whether the message should be view once or not.

    Returns:
        The timestamp of the sent or edited message.
    """
    receiver = self._resolve_receiver(receiver)
    link_preview_raw = link_preview.model_dump() if link_preview else None

    resp = await self._signal.send(
        receiver,
        text,
        base64_attachments=base64_attachments,
        link_preview=link_preview_raw,
        quote_author=quote_author,
        quote_mentions=quote_mentions,
        quote_message=quote_message,
        quote_timestamp=quote_timestamp,
        mentions=mentions,
        text_mode=text_mode,
        edit_timestamp=edit_timestamp,
        view_once=view_once,
    )
    resp_payload = await resp.json()
    timestamp = int(resp_payload["timestamp"])
    self._logger.info(f"[Bot] New message {timestamp} sent:\n{text}")  # noqa: G004

    return timestamp

signal_cli_rest_api_mode async

signal_cli_rest_api_mode() -> str

Return the signal-cli-rest-api mode.

Source code in src/signalbot/bot.py

async def signal_cli_rest_api_mode(self) -> str:
    """Return the signal-cli-rest-api mode."""
    return await self._signal.get_signal_cli_rest_api_mode()

signal_cli_rest_api_version async

signal_cli_rest_api_version() -> str

Return the signal-cli-rest-api version.

Source code in src/signalbot/bot.py

async def signal_cli_rest_api_version(self) -> str:
    """Return the signal-cli-rest-api version."""
    return await self._signal.get_signal_cli_rest_api_version()

start

start(run_forever: bool = True) -> None

Start the bot event loop and scheduler.

Parameters:

Name	Type	Description	Default
run_forever	bool	Whether to start the event loop or only add the task to it.	True

Source code in src/signalbot/bot.py

def start(self, run_forever: bool = True) -> None:  # noqa: FBT001, FBT002
    """Start the bot event loop and scheduler.

    Args:
        run_forever: Whether to start the event loop or only add the task to it.
    """
    self.init_task = self._event_loop.create_task(
        self._rerun_on_exception(self._async_post_init),
    )

    if run_forever:
        self.scheduler.start()

        self._event_loop.run_forever()

start_typing async

start_typing(receiver: str) -> None

Send a typing indicator to a receiver.

Parameters:

Name	Type	Description	Default
receiver	str	Message recipient.	required

Source code in src/signalbot/bot.py

async def start_typing(self, receiver: str) -> None:
    """Send a typing indicator to a receiver.

    Args:
        receiver: Message recipient.
    """
    receiver = self._resolve_receiver(receiver)
    await self._signal.start_typing(receiver)

stop_typing async

stop_typing(receiver: str) -> None

Stop a typing indicator for a receiver.

Parameters:

Name	Type	Description	Default
receiver	str	Message recipient.	required

Source code in src/signalbot/bot.py

async def stop_typing(self, receiver: str) -> None:
    """Stop a typing indicator for a receiver.

    Args:
        receiver: Message recipient.
    """
    receiver = self._resolve_receiver(receiver)
    await self._signal.stop_typing(receiver)

update_contact async

update_contact(
    receiver: str,
    expiration_in_seconds: int | None = None,
    name: str | None = None,
) -> None

Update a contact's metadata.

Parameters:

Name	Type	Description	Default
receiver	str	Contact identifier.	required
expiration_in_seconds	int | None	Expiration timer in seconds.	None
name	str | None	Contact display name.	None

Source code in src/signalbot/bot.py

async def update_contact(
    self,
    receiver: str,
    expiration_in_seconds: int | None = None,
    name: str | None = None,
) -> None:
    """Update a contact's metadata.

    Args:
        receiver: Contact identifier.
        expiration_in_seconds: Expiration timer in seconds.
        name: Contact display name.
    """
    receiver = self._resolve_receiver(receiver)
    await self._signal.update_contact(
        receiver,
        expiration_in_seconds=expiration_in_seconds,
        name=name,
    )

update_group async

update_group(
    group_id: str,
    base64_avatar: str | None = None,
    description: str | None = None,
    expiration_in_seconds: int | None = None,
    name: str | None = None,
) -> None

Update a group's metadata.

Parameters:

Name	Type	Description	Default
group_id	str	Group identifier or name.	required
base64_avatar	str | None	Base64-encoded avatar.	None
description	str | None	Group description.	None
expiration_in_seconds	int | None	Expiration timer in seconds.	None
name	str | None	Group display name.	None

Source code in src/signalbot/bot.py

async def update_group(
    self,
    group_id: str,
    base64_avatar: str | None = None,
    description: str | None = None,
    expiration_in_seconds: int | None = None,
    name: str | None = None,
) -> None:
    """Update a group's metadata.

    Args:
        group_id: Group identifier or name.
        base64_avatar: Base64-encoded avatar.
        description: Group description.
        expiration_in_seconds: Expiration timer in seconds.
        name: Group display name.
    """
    group_id = self._resolve_receiver(group_id)
    await self._signal.update_group(
        group_id,
        base64_avatar=base64_avatar,
        description=description,
        expiration_in_seconds=expiration_in_seconds,
        name=name,
    )

enable_console_logging

enable_console_logging(level: int = WARNING) -> None

Enable console logging for the signalbot logger.

Parameters:

Name	Type	Description	Default
level	int	Logging level for the logger.	WARNING

Source code in src/signalbot/bot.py

def enable_console_logging(level: int = logging.WARNING) -> None:
    """Enable console logging for the signalbot logger.

    Args:
        level: Logging level for the logger.
    """
    handler = logging.StreamHandler()

    formatter = logging.Formatter(
        "%(asctime)s %(name)s [%(levelname)s] - %(funcName)s - %(message)s"
    )
    handler.setFormatter(formatter)

    logger = logging.getLogger(LOGGER_NAME)
    logger.addHandler(handler)
    logger.setLevel(level)