Storage

Where updates wait between “received from the server” and “handled”, plus the failed updates and the pinned keys. The guide is Storage and failed updates.

class MemoryStorage(*, max_failed=1000)

Keeps the updates in memory: the default storage.

Everything is lost when the process ends, so Storage.durable is False: the bot confirms an update to the server only once its handlers have finished. A crash means the server sends some updates again, never that one is lost. The price is speed: long polling waits for a whole batch to be handled before it asks for more, and a webhook handles one update at a time, because the server waits for each answer.

It sees only its own process. Two processes that poll the same bot are not detected (the bot only logs a warning when polls keep coming back empty), and the pins last only as long as the process. A bot stopped and started again in the same process finds its pending and failed updates and its pins where it left them.

max_failed

How many failed updates the list keeps. When one more fails, the one with the lowest updateId is dropped.

Type:

int

Sets up an empty storage.

Parameters:

max_failed (int) – How many failed updates to keep, 1000 by default.

open(bot_id)

Binds the storage to a bot.

For the bot it was opened for before, it keeps everything it holds. For another bot it starts again from nothing, rather than mix two queues.

Parameters:

bot_id (int) – The bot’s user ID, read from its token.

close()

Does nothing: the data stays in memory, for the next open in this process.

try_lock()

Takes the right to run the bot’s handlers, if no bot that uses this object holds it.

The right lives in this object, so another process, or another MemoryStorage, can’t see it. While it is taken the answer is False, even to the bot that took it.

Returns:

True if the right was free and is now taken, False if it was taken already.

Return type:

bool

unlock()

Frees the right taken by try_lock, whoever took it.

The right is not tied to a bot: a bot that never got it frees it too when it stops.

accepted_max()

Reads the high-water mark, as Storage.accepted_max describes.

Returns:

The highest updateId accepted, or None before the first one.

Return type:

int | None

accept(updates)

Keeps the updates it has not seen before, as Storage.accept describes.

Parameters:

updates (Sequence[Mapping[str, Any]]) – The updates, as the server sent them.

Returns:

The new updates, sorted by updateId.

Raises:

StorageError – An update has no integer updateId. Nothing is kept.

Return type:

list[Mapping[str, Any]]

pending()

Lists the pending updates: accepted, and neither finished nor failed.

Returns:

The pending updates, oldest first.

Return type:

list[Mapping[str, Any]]

begin(update_id)

Counts one more handler run of a pending update, as Storage.begin describes.

Parameters:

update_id (int) – The update about to run.

Returns:

The number of runs, this one included, or 0 when the update is not pending.

Return type:

int

finish(update_id)

Forgets a pending update: its handlers are done.

Parameters:

update_id (int) – The update that is done.

fail(update_id, error)

Moves a pending update to the failed list, with the error and the time.

Then, if the list holds more than max_failed updates, the oldest are dropped.

Parameters:
failed(limit=100)

Lists the failed updates, oldest first.

Parameters:

limit (int) – The most to return.

Returns:

Up to limit failed updates, in updateId order.

Return type:

list[FailedUpdate]

requeue_failed(update_ids=None)

Moves failed updates back to the pending list, to run again.

Parameters:

update_ids (Sequence[int] | None) – The updates to move. None moves all of them; an ID that is not in the failed list is skipped.

Returns:

How many updates were moved.

Return type:

int

discard_failed(update_ids=None)

Deletes failed updates.

Parameters:

update_ids (Sequence[int] | None) – The updates to delete. None deletes all of them; an ID that is not in the failed list is skipped.

Returns:

How many updates were deleted.

Return type:

int

get_pin(user_id)

Reads the key fingerprint pinned for a person.

Parameters:

user_id (int) – The person’s user ID.

Returns:

The pinned fingerprint, or None if none is pinned yet.

Return type:

str | None

set_pin(user_id, fingerprint)

Pins a key fingerprint for a person, in upper case, replacing any earlier pin.

Parameters:

  • user_id (int) – The person’s user ID.

  • fingerprint (str) – The key’s fingerprint, in hexadecimal.

class SQLiteStorage(path, *, synchronous='FULL', timeout=10.0, max_failed=10_000)

Keeps the updates in a SQLite database file, so they survive restarts and crashes.

Storage.durable is True: the bot confirms an update to the server as soon as it is written here, without waiting for its handlers. A crash before they finish loses nothing: the update is still in the file and runs at the next start. Bot(storage="bot.db") is the same as Bot(storage=SQLiteStorage("bot.db")).

How it works:

  • The database is in WAL mode, with synchronous=FULL by default: an update is confirmed to the server as soon as it is written, so a write that a power cut could undo would be an update lost.

  • Writes of several statements run in one BEGIN IMMEDIATE transaction. Worker processes that share the file wait for each other there, so two of them never accept the same update.

  • An operating system lock on a file next to the database, <path>.<bot id>.lock (for example bot.db.42.lock), lets one process at a time run a bot’s handlers. The system releases it when the process dies, even in a crash. A second process that polls the same bot gets PollingConflictError; webhook workers share the file, and the one that holds the lock runs the handlers.

  • One file can hold several bots: each bot’s rows are kept apart by its ID, and each bot has its own lock file.

  • The failed list keeps the newest max_failed updates per bot, each error cut to 4000 characters.

  • The methods that read or write updates and pins raise StorageError while the storage is not open: before open, or after close.

On Linux and macOS the database and the lock file are created readable by their owner only (0600), and opening an existing database that other users can read logs a warning. Keep the file on a local disk: SQLite locks are not reliable on network shares (NFS, SMB).

path

The database file.

Type:

Path

max_failed

How many failed updates the list keeps per bot. When one more fails, the one with the lowest updateId is dropped.

Type:

int

Warning

Updates are stored as the server sent them. From encrypted chats they are armored OpenPGP messages, and nothing decrypted is ever written; from readable chats they are plain text. Protect the file and its backups like the messages it holds. To copy it while the bot runs, use SQLite’s own backup (sqlite3 bot.db ".backup copy.db").

Sets up the storage. Nothing touches the file until the bot starts and calls open.

Parameters:

  • path (str | PathLike[str]) – The database file. It is created, with its folder, if it does not exist.

  • synchronous (str) – SQLite’s synchronous setting: "FULL" (the default), "EXTRA" or "NORMAL", in upper or lower case. Keep "FULL": with "NORMAL" a power cut can lose the last update the server was told is safe.

  • timeout (float) – How many seconds to wait when another process holds the database’s write lock.

  • max_failed (int) – How many failed updates to keep per bot, 10 000 by default.

Raises:

ValueError – synchronous is not one of the three values.

open(bot_id)

Opens the database file for a bot, and creates it if it does not exist.

A new file is created with its folder, readable by its owner only on Linux and macOS. Opening an existing file that other users can read logs a warning. The connection uses WAL mode and the synchronous setting given to the constructor. Opened again for another bot, the storage releases the first bot’s lock and works on the new bot’s rows.

Parameters:

bot_id (int) – The bot’s user ID, read from its token.

Raises:

  • StorageError – The file can’t be opened as a SQLite database.

  • OSError – The file or its folder can’t be created.

close()

Releases the lock and closes the database. open can open it again later.

try_lock()

Takes the operating system lock on <path>.<bot id>.lock.

The lock file is created next to the database if needed, and the process ID is written in it for lock_holder. Asking again while holding the lock returns True. Two SQLiteStorage objects on the same file conflict even within one process.

Returns:

True if this storage holds the lock now, False if someone else holds it.

Raises:

StorageError – The storage was never opened.

Return type:

bool

unlock()

Releases the file lock, if this storage holds it. The storage stays open.

lock_holder()

Reads the process ID written in the lock file, for error messages.

A process writes its ID there when it takes the lock, and the ID stays after the lock is released: it names the current holder only while the lock is taken.

Returns:

The process ID as text, or None before open or when the file is empty or can’t be read.

Return type:

str | None

require_lock()

Takes the lock, or raises an error that says which process has it.

The bot calls it when it starts long polling with this storage.

Raises:
accepted_max()

Reads the high-water mark kept in the file, as Storage.accepted_max describes.

Returns:

The highest updateId accepted, or None before the first one.

Return type:

int | None

accept(updates)

Writes the updates it has not seen before, in one BEGIN IMMEDIATE transaction.

Worker processes that share the file wait for each other here, so two of them never accept the same update. The rest is as Storage.accept describes.

Parameters:

updates (Sequence[Mapping[str, Any]]) – The updates, as the server sent them.

Returns:

The new updates, sorted by updateId.

Raises:

StorageError – SQLite could not write them, one has no integer updateId, or the storage is not open. Nothing is written.

Return type:

list[Mapping[str, Any]]

pending()

Lists the pending updates: accepted, and neither finished nor failed.

Returns:

The pending updates, oldest first.

Return type:

list[Mapping[str, Any]]

count_pending()

Counts the pending updates with one query, without reading them.

Returns:

How many updates are accepted and neither finished nor failed.

Return type:

int

pending_ids()

Lists the IDs of the pending updates, oldest first, without reading the updates.

Returns:

The updateId of each pending update, in ascending order.

Return type:

list[int]

load(update_ids)

Reads the pending updates with the given IDs.

Parameters:

update_ids (Sequence[int]) – The IDs to read. An ID that is not pending any more is left out.

Returns:

The pending updates among update_ids, oldest first.

Return type:

list[Mapping[str, Any]]

begin(update_id)

Counts one more handler run of a pending update, in the file.

The count is kept with the update, so it survives a crash: see Storage.begin.

Parameters:

update_id (int) – The update about to run.

Returns:

The number of runs, this one included, or 0 when the update is not pending.

Return type:

int

finish(update_id)

Deletes a pending update: its handlers are done.

Parameters:

update_id (int) – The update that is done.

fail(update_id, error)

Moves a pending update to the failed list, with the error cut to 4000 characters.

In the same transaction, the oldest failed updates beyond max_failed are deleted.

Parameters:
Raises:

StorageError – SQLite could not write it. Nothing changes.

failed(limit=100)

Lists the failed updates, oldest first.

Parameters:

limit (int) – The most to return.

Returns:

Up to limit failed updates, in updateId order.

Return type:

list[FailedUpdate]

requeue_failed(update_ids=None)

Moves failed updates back to the pending list, in one transaction.

Parameters:

update_ids (Sequence[int] | None) – The updates to move. None moves all of them; an ID that is not in the failed list is skipped.

Returns:

How many updates were moved.

Return type:

int

discard_failed(update_ids=None)

Deletes failed updates from the file.

Parameters:

update_ids (Sequence[int] | None) – The updates to delete. None deletes all of them; an ID that is not in the failed list is skipped.

Returns:

How many updates were deleted.

Return type:

int

get_pin(user_id)

Reads the key fingerprint pinned for a person.

Parameters:

user_id (int) – The person’s user ID.

Returns:

The pinned fingerprint, or None if none is pinned yet.

Return type:

str | None

set_pin(user_id, fingerprint)

Pins a key fingerprint for a person, in upper case, replacing any earlier pin.

Parameters:

  • user_id (int) – The person’s user ID.

  • fingerprint (str) – The key’s fingerprint, in hexadecimal.

class Storage

The interface every storage implements: subclass it to keep updates somewhere else.

The bot hands each update it receives to its storage before a handler sees it, and asks the storage what is left to do at every start. MemoryStorage and SQLiteStorage implement it. To keep updates in PostgreSQL or Redis, set durable and write the abstract methods below: each one says what it must do and when the bot calls it. count_pending, pending_ids and load have defaults built on pending; override them when loading every pending update at once is slow.

Updates are plain mappings of the server’s JSON. The key that matters here is updateId, a number that grows with each new update of the bot.

Rules for every method:

  • It is synchronous and thread safe. The bot calls it from its event loop and from worker threads, sometimes at the same time. With durable False the calls for each update run on the event loop itself, so they must be quick and must not block.

  • It works on the bot given to open only. One store may hold several bots, kept apart by their ID.

  • It raises StorageError when the store fails. The bot recovers where it can: a batch that accept could not store is fetched again by long polling, or answered 503 on a webhook so that the server sends it again, and an update whose run could not be counted or recorded stays pending and runs again later.

When the bot starts, it calls open, try_lock and, for long polling, accepted_max. While it runs, it calls accept, pending_ids, load, begin, and finish or fail. When it stops, it calls unlock and close. The other methods serve Bot methods such as Bot.failed_updates, and key pinning.

durable

True only if accept has saved the updates for good when it returns. The bot then confirms them to the server right away, before their handlers run, and makes the calls for each update from worker threads. With False, the default, it confirms an update only after its handlers finished, as with MemoryStorage. Set it on your class.

Type:

bool

Example

A storage of your own, given to the bot like the built-in ones:

from pgpcbot import Bot, Storage


class PostgresStorage(Storage):
    durable = True  # accept() commits before it returns

    def open(self, bot_id):
        ...

    # ... and every other abstract method


bot = Bot(storage=PostgresStorage())
abstractmethod open(bot_id)

Prepares the storage for one bot.

Called at every start, before the bot uses the storage for anything else. Everything after it (the high-water mark, the pending and failed updates, the pins, the lock) belongs to this bot only: a store shared by several bots keeps them apart by this ID. After close, the bot may call it again when it starts again.

Bot.failed_updates, Bot.retry_failed, Bot.discard_failed and Bot.trust_key call the storage directly, so on a bot that is not running they reach a storage that is not open. SQLiteStorage then raises StorageError.

Parameters:

bot_id (int) – The bot’s user ID, read from its token.

Raises:

StorageError – The store can’t be opened. The bot does not start.

abstractmethod close()

Releases what the storage holds: connections, open files, the lock.

Called when the bot stops, after unlock. Its errors are ignored.

abstractmethod try_lock()

Takes the right to run this bot’s handlers, if nobody else holds it.

Only one process at a time may run a bot’s handlers, and this is how the bot makes sure. Called at start, after open. Long polling does not start without the right: the bot raises PollingConflictError. A webhook process that does not get it keeps accepting deliveries (with a durable storage it only stores them, for the process that holds the right) and asks again every 5 seconds.

The right must end by itself when the process that holds it dies, even in a crash, or the bot could never start again after one. An operating system file lock does that.

Returns:

True if the caller holds the right now, False if someone else has it.

Return type:

bool

abstractmethod unlock()

Gives back the right taken by try_lock.

Called when the bot stops, before close, even if the right was never taken: then it must do nothing. Its errors are ignored.

abstractmethod accepted_max()

Reads the high-water mark: the highest updateId ever accepted for this bot.

The mark never goes down, not even when the updates are finished, failed or discarded. Long polling reads it at start: with a durable storage, its first request confirms to the server every update up to the mark.

Returns:

The highest updateId that accept stored, or None before the first one.

Return type:

int | None

abstractmethod accept(updates)

Stores the updates it has not seen before and returns them, oldest first.

Called with every batch of a long poll and every webhook delivery. An update is new when its updateId is above accepted_max; anything at or below the mark is a repeat and is skipped, as is a second copy of an ID within the batch. The batch can come in any order.

It must be atomic: either every new update is stored and the mark moves to the highest of them, or nothing changes and it raises. With Storage.durable True, the updates must be saved for good when it returns, because the bot tells the server it has them right after.

Only updateId is sure to be there. With skip_pending=True the bot stores {"updateId": n, "kind": "skipped"} for each update it throws away, and finishes them at once, so that a late repeat is still recognised.

Parameters:

updates (Sequence[Mapping[str, Any]]) – The updates, as the server sent them.

Returns:

The new updates, sorted by updateId. These are the ones the handlers will see.

Raises:

StorageError – The updates could not be stored, or one has no integer updateId. A long poll then does not confirm the batch and fetches it again; a webhook answers 503. Any other exception stops a polling bot.

Return type:

list[Mapping[str, Any]]

abstractmethod pending()

Lists the pending updates: accepted, and neither finished nor failed.

The bot does not call it directly: the defaults of pending_ids, load and count_pending build on it.

Returns:

The pending updates, oldest first, as accept stored them.

Return type:

list[Mapping[str, Any]]

abstractmethod begin(update_id)

Counts one more handler run of a pending update.

Called right before the handlers of an update run. The count is how the bot spots an update that kills the process every time: after max_handler_runs runs that never finished, it moves the update to the failed list instead of running it again. A durable storage must keep the count with the update, so that it survives a crash. After requeue_failed the count starts again from zero.

Parameters:

update_id (int) – The update about to run.

Returns:

The number of runs, this one included, so 1 the first time. 0 when the update is not pending any more, because it finished or failed in the meantime; the bot then does not run it.

Return type:

int

abstractmethod finish(update_id)

Forgets a pending update: its handlers are done.

Called when the handlers return, or when an error handler took their error. An ID that is not pending must be ignored. If it raises, the update stays pending and runs again later: the bot handles updates at least once.

Parameters:

update_id (int) – The update that is done.

abstractmethod fail(update_id, error)

Moves a pending update to the failed list, with the error.

Called when the handlers raised and no error handler took the error, and when an update ran max_handler_runs times without ever finishing (see begin). The update leaves the pending list, and failed returns it with the error and the time. An ID that is not pending must be ignored. Anyone who can write to the bot can make updates fail, so keep the list bounded: the built-in storages keep the newest max_failed.

Parameters:
abstractmethod failed(limit=100)

Lists the failed updates, oldest first.

Bot.failed_updates calls it.

Parameters:

limit (int) – The most to return.

Returns:

Up to limit failed updates, in updateId order.

Return type:

list[FailedUpdate]

abstractmethod requeue_failed(update_ids=None)

Moves failed updates back to the pending list, to run again.

Bot.retry_failed calls it, and a running bot then picks them up at once; otherwise they run at the next start. Their run count starts again from zero.

Parameters:

update_ids (Sequence[int] | None) – The updates to move. None moves all of them; an ID that is not in the failed list is skipped.

Returns:

How many updates were moved.

Return type:

int

abstractmethod discard_failed(update_ids=None)

Deletes failed updates for good.

Bot.discard_failed calls it.

Parameters:

update_ids (Sequence[int] | None) – The updates to delete. None deletes all of them; an ID that is not in the failed list is skipped.

Returns:

How many updates were deleted.

Return type:

int

abstractmethod get_pin(user_id)

Reads the key fingerprint pinned for a person.

With pin_keys=True the bot calls it each time it checks a person’s key: before it seals a message to them, and when it checks the signature of a message they sent.

Parameters:

user_id (int) – The person’s user ID.

Returns:

The pinned fingerprint, or None if none is pinned yet.

Return type:

str | None

abstractmethod set_pin(user_id, fingerprint)

Pins a key fingerprint for a person, replacing any earlier pin.

Called with pin_keys=True the first time the bot sees a person’s key, and by Bot.trust_key. The built-in storages keep it in upper case; the bot compares fingerprints without regard to case.

Parameters:

  • user_id (int) – The person’s user ID.

  • fingerprint (str) – The key’s fingerprint, in hexadecimal: 40 characters for a v4 key, 64 for a v6 key.

count_pending()

Counts the pending updates.

The bot itself does not need it: it is there for your own checks, such as a report at start.

Returns:

How many updates are accepted and neither finished nor failed.

Return type:

int

pending_ids()

Lists the IDs of the pending updates, oldest first.

The bot calls it when it looks for work: at start if it got the right from try_lock, or later when it gets it; after requeue_failed; and, in a webhook process with a durable storage, every 2 seconds while it holds the right, to pick up what other worker processes stored.

Returns:

The updateId of each pending update, in ascending order.

Return type:

list[int]

load(update_ids)

Reads the pending updates with the given IDs.

Called right after pending_ids, with the IDs that are not running yet, as many as the bot has room for.

Parameters:

update_ids (Sequence[int]) – The IDs to read. An ID that is not pending any more is left out.

Returns:

The pending updates among update_ids, oldest first.

Return type:

list[Mapping[str, Any]]

class FailedUpdate(update_id, payload, error, failed_at)

An update whose handlers failed, kept with the error.

Bot.failed_updates returns them, oldest first. A storage of your own builds them in its Storage.failed: the fields are also the constructor’s arguments, in this order.

update_id

The update’s updateId.

Type:

int

payload

The update as the server sent it. From an encrypted chat it holds the armored message: nothing decrypted is kept.

Type:

Mapping[str, Any]

error

What went wrong: the exception’s type and message, and the file and line where it was raised, such as "KeyError: 'city' at bot.py:31". For an update from an encrypted chat the message is left out, because a handler’s exception could quote the decrypted text; the messages of pgpcbot’s own exceptions are kept. An update set aside because the process kept stopping while it ran says "gave up: ...".

Type:

str

failed_at

When it failed, in UTC.

Type:

datetime