Updates and events

A comment, a like, a post finishing its encoding, somebody removing your app - every one of these arrives as the same Update, whichever network it came from and whether the network pushed it to you or socialchimp had to check on a timer. Your code cannot tell the two ways apart, and does not need to.

Update

Update dataclass

Update(
    id: str,
    kind: UpdateKind,
    platform: str,
    connection_id: str,
    created_at: datetime,
    raw: RawData = dict(),
    kind_name: str = "",
    envelope: RawData = dict(),
    actor: Person | None = None,
    post_id: str | None = None,
    about_post_id: str | None = None,
    thread_root_id: str | None = None,
    conversation_id: str | None = None,
)

Something that happened on a social network.

The same object whether the network pushed it to us or we found it by checking on a timer.

Attributes:

Name	Type	Description
id	str	The network's identifier for this update. Used to spot the same update arriving twice.
kind	UpdateKind	What happened.
platform	str	Which network it happened on, for example "facebook".
connection_id	str	Which of your connections it concerns, so you know whose account this is without another lookup.
created_at	datetime	When it happened, according to the network. Always has a timezone.
raw	RawData	The one thing that happened, in the network's own untouched words, for anything we did not model. A handler reads this straight - it is this update's own change and nothing else.
kind_name	str	The word the network used. The same as kind's own value for anything we recognise; for UNKNOWN this is where the network's original word is kept.
envelope	RawData	The message this arrived in, where the network wraps things up. Meta puts several changes in one message and names the page and the time out there rather than on each change, so that is what this holds. Empty for a network that sends one thing on its own, and for an update found by asking.
actor	Person | None	Who did it, when the network says. None when it is not known or does not apply.
post_id	str | None	The thing that happened, as a post id - the reply, the mention, or the message itself. None when this update is not about one particular post.
about_post_id	str | None	The connected account's own post this concerns - the one that was liked, reposted or replied to. None when there is no such post, or the network did not say.
thread_root_id	str | None	The top of the thread this sits in, when that is known without an extra request. None otherwise - Mastodon leaves this None unless it is asked for, because finding it needs a call of its own.
conversation_id	str | None	Which conversation this concerns, for a MESSAGE_RECEIVED update. None when it does not apply, or the network did not say.

from_network classmethod

from_network(
    *,
    update_id: str,
    kind_name: str,
    platform: str,
    connection_id: str,
    created_at: datetime,
    raw: RawData | None = None,
    envelope: RawData | None = None,
    actor: Person | None = None,
    post_id: str | None = None,
    about_post_id: str | None = None,
    thread_root_id: str | None = None,
    conversation_id: str | None = None,
) -> Update

Build an update from a word a network gave us.

This is what platform files use. It never fails on a word we do not know: the update comes back as UNKNOWN with the word kept.

Parameters:

Name	Type	Description	Default
update_id	str	The network's identifier for this update.	required
kind_name	str	The network's word for what happened, already translated into socialchimp's vocabulary by the platform file.	required
platform	str	Which network it happened on.	required
connection_id	str	Which of your connections it concerns.	required
created_at	datetime	When it happened. Must have a timezone.	required
raw	RawData | None	The one thing that happened, untouched. Pass the change itself, not the message it came in - a handler should not have to hunt through a list for its own change.	None
envelope	RawData | None	The message it arrived in, where the network wraps things up and puts the account and the time out there.	None
actor	Person | None	Who did it, when the network says.	None
post_id	str | None	The thing that happened, as a post id.	None
about_post_id	str | None	The connected account's own post this concerns.	None
thread_root_id	str | None	The top of the thread this sits in, when known without an extra request.	None
conversation_id	str | None	Which conversation this concerns, for a message.	None

Returns:

Type	Description
Update	The update, ready to deliver.

Source code in src/socialchimp/events.py

@classmethod
def from_network(
    cls,
    *,
    update_id: str,
    kind_name: str,
    platform: str,
    connection_id: str,
    created_at: datetime,
    raw: RawData | None = None,
    envelope: RawData | None = None,
    actor: Person | None = None,
    post_id: str | None = None,
    about_post_id: str | None = None,
    thread_root_id: str | None = None,
    conversation_id: str | None = None,
) -> Update:
    """Build an update from a word a network gave us.

    This is what platform files use. It never fails on a word we do not
    know: the update comes back as `UNKNOWN` with the word kept.

    Args:
        update_id: The network's identifier for this update.
        kind_name: The network's word for what happened, already
            translated into socialchimp's vocabulary by the platform file.
        platform: Which network it happened on.
        connection_id: Which of your connections it concerns.
        created_at: When it happened. Must have a timezone.
        raw: The one thing that happened, untouched. Pass the change
            itself, not the message it came in - a handler should not
            have to hunt through a list for its own change.
        envelope: The message it arrived in, where the network wraps
            things up and puts the account and the time out there.
        actor: Who did it, when the network says.
        post_id: The thing that happened, as a post id.
        about_post_id: The connected account's own post this concerns.
        thread_root_id: The top of the thread this sits in, when known
            without an extra request.
        conversation_id: Which conversation this concerns, for a
            message.

    Returns:
        The update, ready to deliver.
    """
    return cls(
        id=update_id,
        kind=UpdateKind.from_name(kind_name),
        platform=platform,
        connection_id=connection_id,
        created_at=created_at,
        raw=raw if raw is not None else {},
        kind_name=kind_name,
        envelope=envelope if envelope is not None else {},
        actor=actor,
        post_id=post_id,
        about_post_id=about_post_id,
        thread_root_id=thread_root_id,
        conversation_id=conversation_id,
    )

UpdateKind

Bases: Enum

What happened.

The values are the words socialchimp uses on the wire. A network's own word for the same thing is translated by its platform file, so your handlers only ever see these.

Anything we do not recognise becomes UNKNOWN, with the network's original word kept on Update.kind_name. Networks add new kinds without warning, and an app that only cares about comments should keep working the day one appears.

COMMENT_CREATED class-attribute instance-attribute

COMMENT_CREATED = 'comment_created'

Someone commented on a post.

COMMENT_DELETED class-attribute instance-attribute

COMMENT_DELETED = 'comment_deleted'

A comment was removed, by its author or by a moderator.

POST_PUBLISHED class-attribute instance-attribute

POST_PUBLISHED = 'post_published'

A post the network was still working on is now live.

YouTube and TikTok keep working after they accept an upload, so this can arrive long after publish() returned.

POST_FAILED class-attribute instance-attribute

POST_FAILED = 'post_failed'

A post the network was still working on will never go live.

POST_DELETED class-attribute instance-attribute

POST_DELETED = 'post_deleted'

A post was removed, by its author or by a moderator.

POST_DRAFTED class-attribute instance-attribute

POST_DRAFTED = 'post_drafted'

The network put the post in somebody's drafts for them to finish.

TikTok can do this instead of posting straight away. Nothing is wrong, and nothing more will happen until a person taps a button.

REACTION_ADDED class-attribute instance-attribute

REACTION_ADDED = 'reaction_added'

Someone liked, favourited or reacted to a post.

MENTION class-attribute instance-attribute

MENTION = 'mention'

Someone named this account in a post of their own.

CONNECTION_REVOKED class-attribute instance-attribute

CONNECTION_REVOKED = 'connection_revoked'

The person took your app's access away.

Delete the connection when you see this. Its token has already stopped working, and Meta will not tell you twice.

REVIEW_CREATED class-attribute instance-attribute

REVIEW_CREATED = 'review_created'

Someone left a review.

REVIEW_UPDATED class-attribute instance-attribute

REVIEW_UPDATED = 'review_updated'

Someone changed a review they had already left - its rating, its words, or both.

QUESTION_CREATED class-attribute instance-attribute

QUESTION_CREATED = 'question_created'

Someone asked a question.

ANSWER_CREATED class-attribute instance-attribute

ANSWER_CREATED = 'answer_created'

Someone answered a question - not necessarily the business itself.

REPOST_ADDED class-attribute instance-attribute

REPOST_ADDED = 'repost_added'

Someone reposted, boosted or reblogged a post.

Mastodon and Bluesky reposts used to arrive as REACTION_ADDED; from 0.8.0 they have this kind of their own instead.

MESSAGE_RECEIVED class-attribute instance-attribute

MESSAGE_RECEIVED = 'message_received'

A direct message arrived. Update.conversation_id names which conversation, where the network says.

FOLLOWED class-attribute instance-attribute

FOLLOWED = 'followed'

Someone started following this account.

Used to arrive as UNKNOWN; from 0.8.0 it has this kind of its own.

UNKNOWN class-attribute instance-attribute

UNKNOWN = 'unknown'

Something we have no name for yet. Look at raw to see what it was.

from_name classmethod

from_name(name: str) -> UpdateKind

Turn a word from the wire into a kind, without ever failing.

Parameters:

Name	Type	Description	Default
name	str	The word socialchimp uses for this kind of update.	required

Returns:

Type	Description
UpdateKind	The matching kind, or UNKNOWN if there is no match.

Source code in src/socialchimp/events.py

@classmethod
def from_name(cls, name: str) -> UpdateKind:
    """Turn a word from the wire into a kind, without ever failing.

    Args:
        name: The word socialchimp uses for this kind of update.

    Returns:
        The matching kind, or `UNKNOWN` if there is no match.
    """
    try:
        return cls(name)
    except ValueError:
        return cls.UNKNOWN

Receiving pushed updates (webhooks)

verify_hmac_sha256

verify_hmac_sha256(
    body: bytes,
    headers: Mapping[str, str],
    *,
    secret: str,
    header_name: str = "X-Hub-Signature-256",
    prefix: str = "sha256=",
) -> None

Check a signed body against the secret only you and the network know.

This is how Meta signs what it sends to Facebook, Instagram, Threads and WhatsApp apps, and several other networks copy it.

It takes the raw bytes and a plain mapping of headers on purpose. It must never be handed a framework's request object, because by the time one of those has parsed the JSON the original bytes are gone. Read the body, check it here, and only then parse it - re-encoding a parsed body changes the spacing and the key order, and the signature is over the exact bytes that were sent. Frameworks that parse the body for you are the single most common reason a correct signature appears to fail.

Parameters:

Name	Type	Description	Default
body	bytes	The request body, exactly as it arrived. Not a parsed and re-encoded copy of it.	required
headers	Mapping[str, str]	The request's headers. Case does not matter.	required
secret	str	The secret agreed with the network. Meta calls this the app secret.	required
header_name	str	Which header carries the signature.	'X-Hub-Signature-256'
prefix	str	What the network puts in front of the hex digits. Pass "" for a network that sends the digits on their own.	'sha256='

Raises:

Type	Description
SignatureError	If the header is missing, malformed, or does not match. Answer 401 and stop.

Source code in src/socialchimp/events.py

def verify_hmac_sha256(
    body: bytes,
    headers: Mapping[str, str],
    *,
    secret: str,
    header_name: str = "X-Hub-Signature-256",
    prefix: str = "sha256=",
) -> None:
    """Check a signed body against the secret only you and the network know.

    This is how Meta signs what it sends to Facebook, Instagram, Threads and
    WhatsApp apps, and several other networks copy it.

    It takes the raw bytes and a plain mapping of headers on purpose. It must
    never be handed a framework's request object, because by the time one of
    those has parsed the JSON the original bytes are gone. Read the body,
    check it here, and only then parse it - re-encoding a parsed body changes
    the spacing and the key order, and the signature is over the exact bytes
    that were sent. Frameworks that parse the body for you are the single most
    common reason a correct signature appears to fail.

    Args:
        body: The request body, exactly as it arrived. Not a parsed and
            re-encoded copy of it.
        headers: The request's headers. Case does not matter.
        secret: The secret agreed with the network. Meta calls this the app
            secret.
        header_name: Which header carries the signature.
        prefix: What the network puts in front of the hex digits. Pass `""`
            for a network that sends the digits on their own.

    Raises:
        SignatureError: If the header is missing, malformed, or does not
            match. Answer 401 and stop.
    """
    sent = _header(headers, header_name)
    if sent is None:
        message = (
            f"This request has no {header_name} header, so there is nothing "
            f"to check it against. Refusing it."
        )
        raise SignatureError(message)

    if not sent.startswith(prefix):
        message = (
            f"The {header_name} header does not start with {prefix!r}, so it "
            f"is not a signature we know how to check. Refusing it."
        )
        raise SignatureError(message)

    expected = prefix + hmac.new(secret.encode(), body, hashlib.sha256).hexdigest()

    # Compared this way so that how long the comparison takes says nothing
    # about how much of the signature was right.
    if not hmac.compare_digest(sent, expected):
        message = (
            "The signature does not match the body. Either the body was "
            "changed on the way here, or it was signed with a different "
            "secret. If you are sure the secret is right, check that nothing "
            "parsed and rebuilt the body before it reached this function."
        )
        raise SignatureError(message)

verify_shared_secret

verify_shared_secret(
    headers: Mapping[str, str],
    *,
    secret: str,
    header_name: str = "X-Telegram-Bot-Api-Secret-Token",
) -> None

Check a header that simply repeats a secret back to us.

Telegram works this way: you give it a secret when you set up the URL, and it sends that same secret with everything it posts to you.

This proves less than a signature does. It says the sender knows the secret; it says nothing about the body being unchanged. Only serve such a URL over HTTPS, and keep the secret long and random - secrets.token_hex makes a good one.

Parameters:

Name	Type	Description	Default
headers	Mapping[str, str]	The request's headers. Case does not matter.	required
secret	str	The secret you gave the network.	required
header_name	str	Which header carries it.	'X-Telegram-Bot-Api-Secret-Token'

Raises:

Type	Description
SignatureError	If the header is missing or holds something else.

Source code in src/socialchimp/events.py

def verify_shared_secret(
    headers: Mapping[str, str],
    *,
    secret: str,
    header_name: str = "X-Telegram-Bot-Api-Secret-Token",
) -> None:
    """Check a header that simply repeats a secret back to us.

    Telegram works this way: you give it a secret when you set up the URL, and
    it sends that same secret with everything it posts to you.

    This proves less than a signature does. It says the sender knows the
    secret; it says nothing about the body being unchanged. Only serve such a
    URL over HTTPS, and keep the secret long and random - `secrets.token_hex`
    makes a good one.

    Args:
        headers: The request's headers. Case does not matter.
        secret: The secret you gave the network.
        header_name: Which header carries it.

    Raises:
        SignatureError: If the header is missing or holds something else.
    """
    sent = _header(headers, header_name)
    if sent is None:
        message = (
            f"This request has no {header_name} header, so there is nothing "
            f"to check it against. Refusing it."
        )
        raise SignatureError(message)

    if not hmac.compare_digest(sent, secret):
        message = (
            f"The secret in {header_name} does not match the one we agreed "
            f"with the network. Refusing it."
        )
        raise SignatureError(message)

check_not_too_old

check_not_too_old(
    sent_at: datetime | float,
    *,
    allowed_age_seconds: float = DEFAULT_ALLOWED_AGE_SECONDS,
    now: datetime | None = None,
) -> None

Refuse a request that was signed too long ago.

A signature stays correct forever. Anyone who gets hold of one request - from a log file, a proxy, a screenshot of a debug page - can send that exact request again next year and the signature will still check out. Refusing anything old closes that off, so run this alongside the signature check, not instead of it.

Only works if the time itself is covered by the signature. Networks that sign a timestamp header, such as Discord, cover it; where a network puts the time inside the body, the body is what was signed, so it counts.

Parameters:

Name	Type	Description	Default
sent_at	datetime | float	When the network says it sent this. Either a datetime with a timezone or plain seconds since 1970, which is what most networks send.	required
allowed_age_seconds	float	How old a request may be. Five minutes by default, which leaves room for clocks that disagree a little.	DEFAULT_ALLOWED_AGE_SECONDS
now	datetime | None	The current time. Only useful in tests.	None

Raises:

Type	Description
ConfigError	If sent_at is a datetime with no timezone.
SignatureError	If the request is older than allowed.

Source code in src/socialchimp/events.py

def check_not_too_old(
    sent_at: datetime | float,
    *,
    allowed_age_seconds: float = DEFAULT_ALLOWED_AGE_SECONDS,
    now: datetime | None = None,
) -> None:
    """Refuse a request that was signed too long ago.

    A signature stays correct forever. Anyone who gets hold of one request -
    from a log file, a proxy, a screenshot of a debug page - can send that
    exact request again next year and the signature will still check out.
    Refusing anything old closes that off, so run this alongside the
    signature check, not instead of it.

    Only works if the time itself is covered by the signature. Networks that
    sign a timestamp header, such as Discord, cover it; where a network puts
    the time inside the body, the body is what was signed, so it counts.

    Args:
        sent_at: When the network says it sent this. Either a datetime with a
            timezone or plain seconds since 1970, which is what most networks
            send.
        allowed_age_seconds: How old a request may be. Five minutes by
            default, which leaves room for clocks that disagree a little.
        now: The current time. Only useful in tests.

    Raises:
        ConfigError: If `sent_at` is a datetime with no timezone.
        SignatureError: If the request is older than allowed.
    """
    if isinstance(sent_at, datetime):
        moment = sent_at
    else:
        moment = datetime.fromtimestamp(sent_at, UTC)

    require_timezone(moment, "sent_at")

    against = now if now is not None else datetime.now(UTC)
    age = (against - moment).total_seconds()
    if age > allowed_age_seconds:
        message = (
            f"This request was signed {age:.0f} seconds ago, and we only "
            f"accept requests up to {allowed_age_seconds:.0f} seconds old. "
            f"An old request with a correct signature can be sent again by "
            f"anyone who copied it, so we refuse it."
        )
        raise SignatureError(message)

answer_setup_check

answer_setup_check(
    params: Mapping[str, str], *, expected_token: str
) -> str

Answer the one-off GET that Meta sends when you point it at a URL.

Before Meta will send you anything it asks your URL a question: it does a GET with a token you chose and a challenge. Echo the challenge back as plain text and the URL starts working. Get it wrong and Meta says the URL could not be verified, without saying why.

Parameters:

Name	Type	Description	Default
params	Mapping[str, str]	The query values from the GET, such as Django's request.GET or FastAPI's request.query_params.	required
expected_token	str	The token you typed into Meta's dashboard. Its own forms call this the verify token.	required

Returns:

Type	Description
str	The challenge. Send it back as the whole body, with a 200 and a
str	content type of text/plain.

Raises:

Type	Description
SignatureError	If this is not a setup check, or the token is wrong. Answer 403 and send nothing back.

Source code in src/socialchimp/events.py

def answer_setup_check(
    params: Mapping[str, str],
    *,
    expected_token: str,
) -> str:
    """Answer the one-off GET that Meta sends when you point it at a URL.

    Before Meta will send you anything it asks your URL a question: it does a
    GET with a token you chose and a challenge. Echo the challenge back as
    plain text and the URL starts working. Get it wrong and Meta says the URL
    could not be verified, without saying why.

    Args:
        params: The query values from the GET, such as Django's `request.GET`
            or FastAPI's `request.query_params`.
        expected_token: The token you typed into Meta's dashboard. Its own
            forms call this the verify token.

    Returns:
        The challenge. Send it back as the whole body, with a 200 and a
        content type of `text/plain`.

    Raises:
        SignatureError: If this is not a setup check, or the token is wrong.
            Answer 403 and send nothing back.
    """
    challenge = params.get("hub.challenge")
    if params.get("hub.mode") != "subscribe" or challenge is None:
        message = (
            "This is not a setup check: it has no hub.mode of 'subscribe' "
            "and a hub.challenge to answer with."
        )
        raise SignatureError(message)

    token = params.get("hub.verify_token")
    if token is None or not hmac.compare_digest(token, expected_token):
        message = (
            "The token in this setup check is not the one we expected, so it "
            "did not come from the network. Refusing it."
        )
        raise SignatureError(message)

    return challenge

Dispatcher

Dispatcher(*, seen: SeenUpdates | None = None)

Sends each update to the code that cares about it.

Register handlers by kind, or one that hears about everything, then hand updates to deliver. Where they came from - a signed request from Meta, a socket held open to Mastodon, or Poller checking LinkedIn on a timer - makes no difference here.

Example

dispatcher = Dispatcher(seen=InMemorySeenUpdates()) dispatcher.on(UpdateKind.COMMENT_CREATED, reply_to_comment) await dispatcher.deliver(update)

Start with no handlers registered.

Parameters:

Name	Type	Description	Default
seen	SeenUpdates | None	A memory of updates already handled. Given one, an update that arrives twice is only handled once. Leave it out and every update is handled every time it arrives.	None

Source code in src/socialchimp/events.py

def __init__(self, *, seen: SeenUpdates | None = None) -> None:
    """Start with no handlers registered.

    Args:
        seen: A memory of updates already handled. Given one, an update
            that arrives twice is only handled once. Leave it out and
            every update is handled every time it arrives.
    """
    self._by_kind: dict[UpdateKind, list[Handler]] = {}
    self._catch_all: list[Handler] = []
    self._seen = seen

on

on(kind: UpdateKind, handler: Handler) -> None

Call this handler for updates of one kind.

Parameters:

Name	Type	Description	Default
kind	UpdateKind	Which updates it wants.	required
handler	Handler	What to call. Registering several for the same kind is fine; they run in the order they were registered.	required

Source code in src/socialchimp/events.py

def on(self, kind: UpdateKind, handler: Handler) -> None:
    """Call this handler for updates of one kind.

    Args:
        kind: Which updates it wants.
        handler: What to call. Registering several for the same kind is
            fine; they run in the order they were registered.
    """
    self._by_kind.setdefault(kind, []).append(handler)

on_any

on_any(handler: Handler) -> None

Call this handler for every update, whatever kind it is.

Useful for writing everything to a log or a queue. It is also the only way to see updates of a kind we have no name for yet.

Parameters:

Name	Type	Description	Default
handler	Handler	What to call.	required

Source code in src/socialchimp/events.py

def on_any(self, handler: Handler) -> None:
    """Call this handler for every update, whatever kind it is.

    Useful for writing everything to a log or a queue. It is also the
    only way to see updates of a kind we have no name for yet.

    Args:
        handler: What to call.
    """
    self._catch_all.append(handler)

deliver async

deliver(update: Update) -> None

Hand one update to every handler that wants it.

Handlers run one after another rather than all at once, so their order is the order you registered them in. A handler that raises does not stop the rest: they all get the update, and what they raised is kept until the end. One broken handler must not cost you the others.

Then, if anything was raised, two things happen. The update is not remembered as handled, because it was not - so the network's retry arrives to a clean slate instead of being skipped by the seen check. And the failures come back to you as an ExceptionGroup, because a handler that could not do its job is your problem to log, alert on or retry, and only your app knows which.

It is always a group, even when only one handler failed. That way there is one shape to catch, and registering a second handler tomorrow does not change what your code has to catch today. Catch it with except*, or with except ExceptionGroup if you only want to know that something went wrong.

A route that lets the group out answers 500, which is exactly how a network is told to send the update again - see socialchimp.contrib.shared.Routes.webhook.

Parameters:

Name	Type	Description	Default
update	Update	What happened.	required

Raises:

Type	Description
ExceptionGroup	Holding what every failed handler raised, if any did. Nothing is raised when they all succeeded.

Source code in src/socialchimp/events.py

async def deliver(self, update: Update) -> None:
    """Hand one update to every handler that wants it.

    Handlers run one after another rather than all at once, so their
    order is the order you registered them in. A handler that raises does
    not stop the rest: they all get the update, and what they raised is
    kept until the end. One broken handler must not cost you the others.

    Then, if anything was raised, two things happen. The update is **not**
    remembered as handled, because it was not - so the network's retry
    arrives to a clean slate instead of being skipped by the `seen` check.
    And the failures come back to you as an `ExceptionGroup`, because a
    handler that could not do its job is your problem to log, alert on or
    retry, and only your app knows which.

    It is always a group, even when only one handler failed. That way
    there is one shape to catch, and registering a second handler
    tomorrow does not change what your code has to catch today. Catch it
    with `except*`, or with `except ExceptionGroup` if you only want to
    know that something went wrong.

    A route that lets the group out answers 500, which is exactly how a
    network is told to send the update again - see
    `socialchimp.contrib.shared.Routes.webhook`.

    Args:
        update: What happened.

    Raises:
        ExceptionGroup: Holding what every failed handler raised, if any
            did. Nothing is raised when they all succeeded.
    """
    if self._seen is not None and await self._seen.seen(update.id):
        logger.debug(
            "Update %s has been handled already, so skipping it.", update.id
        )
        return

    failures: list[Exception] = []
    for handler in [*self._by_kind.get(update.kind, []), *self._catch_all]:
        try:
            await handler(update)
        except Exception as failure:
            # Kept rather than raised here, so the handlers after this one
            # still get their update. They are all raised together below.
            failures.append(failure)

    if failures:
        # Deliberately before `remember`, and instead of it. An update
        # nobody got through is not handled, and writing it down as
        # handled would make the `seen` check above skip the network's
        # retry - the only second chance there is.
        message = (
            f"{len(failures)} handler(s) for update {update.id} failed, so "
            f"it has not been remembered as handled. If the network sends "
            f"it again, the handlers get another go at it."
        )
        raise ExceptionGroup(message, failures)

    if self._seen is not None:
        await self._seen.remember(update.id)

Polling the networks that cannot push

"On a timer" in Networks means this.

Poller

Poller(
    *,
    fetch: FetchUpdates,
    deliver: DeliverUpdate,
    every_seconds: float = 60.0,
    since: datetime | None = None,
    save_marker: SaveMarker | None = None,
)

Checks a network on a timer, for networks that never tell us anything.

LinkedIn, Pinterest, Reddit and Tumblr have no way to push anything to you, so the only way to know about a new comment is to ask. This asks, works out which items are new since last time, and hands them on as the same Update objects a network that does push would have sent. Your handlers cannot tell the difference.

New means "happened after the marker", where the marker is the time of the newest update handed on so far. Give save_marker somewhere to write it and a restart carries on from where it left off instead of going quiet or repeating a day's worth of comments.

Example

poller = Poller(fetch=recent_comments, deliver=dispatcher.deliver) await poller.run_forever()

Set up the checking, without starting it.

Parameters:

Name	Type	Description	Default
fetch	FetchUpdates	Asks the network what it has right now, as updates. It is fine for this to return the same items every time; working out which are new is this class's job.	required
deliver	DeliverUpdate	Where a new update goes. Dispatcher.deliver fits.	required
every_seconds	float	How long to wait between rounds. Watch the network's rate limit: asking every second uses up an hourly allowance in minutes.	60.0
since	datetime | None	The marker you saved last time. Leave it out and everything the first round finds counts as new; pass datetime.now(UTC) to start from this moment instead.	None
save_marker	SaveMarker | None	Called with the new marker whenever it moves, so you can store it. Leave it out and the marker is lost on restart.	None

Source code in src/socialchimp/events.py

def __init__(
    self,
    *,
    fetch: FetchUpdates,
    deliver: DeliverUpdate,
    every_seconds: float = 60.0,
    since: datetime | None = None,
    save_marker: SaveMarker | None = None,
) -> None:
    """Set up the checking, without starting it.

    Args:
        fetch: Asks the network what it has right now, as updates. It is
            fine for this to return the same items every time; working
            out which are new is this class's job.
        deliver: Where a new update goes. `Dispatcher.deliver` fits.
        every_seconds: How long to wait between rounds. Watch the
            network's rate limit: asking every second uses up an hourly
            allowance in minutes.
        since: The marker you saved last time. Leave it out and
            everything the first round finds counts as new; pass
            `datetime.now(UTC)` to start from this moment instead.
        save_marker: Called with the new marker whenever it moves, so you
            can store it. Leave it out and the marker is lost on restart.
    """
    self._fetch = fetch
    self._deliver = deliver
    self._every_seconds = every_seconds
    self._since = since
    self._save_marker = save_marker

check_once async

check_once() -> list[Update]

Do one round: ask, work out what is new, hand it on.

Updates are handed on oldest first, so handlers see things in the order they happened rather than the order the network listed them.

Returns:

Type	Description
list[Update]	The updates that were new this round, oldest first.

Raises:

Type	Description
Exception	Whatever fetch or deliver raised. run_forever catches these; call this yourself and you handle them.

Source code in src/socialchimp/events.py

async def check_once(self) -> list[Update]:
    """Do one round: ask, work out what is new, hand it on.

    Updates are handed on oldest first, so handlers see things in the
    order they happened rather than the order the network listed them.

    Returns:
        The updates that were new this round, oldest first.

    Raises:
        Exception: Whatever `fetch` or `deliver` raised. `run_forever`
            catches these; call this yourself and you handle them.
    """
    found = await self._fetch()
    new = sorted(
        (update for update in found if self._is_new(update)),
        key=lambda update: update.created_at,
    )

    for update in new:
        await self._deliver(update)

    # Moved only after everything has been handed on, so a failure part
    # way through means the next round tries those items again.
    if new:
        self._since = new[-1].created_at
        if self._save_marker is not None:
            await self._save_marker(self._since)

    return new

run_forever async

run_forever() -> None

Keep checking until this task is cancelled.

A round that fails is logged and the next one still happens. Networks go down, tokens hiccup and rate limits bite, and none of those are a reason to stop checking for good. The marker does not move on a failed round, so nothing is skipped over.

Raises:

Type	Description
CancelledError	When the task is cancelled. Passed on rather than swallowed, so shutting down actually shuts down.

Source code in src/socialchimp/events.py

async def run_forever(self) -> None:
    """Keep checking until this task is cancelled.

    A round that fails is logged and the next one still happens. Networks
    go down, tokens hiccup and rate limits bite, and none of those are a
    reason to stop checking for good. The marker does not move on a
    failed round, so nothing is skipped over.

    Raises:
        asyncio.CancelledError: When the task is cancelled. Passed on
            rather than swallowed, so shutting down actually shuts down.
    """
    try:
        while True:
            try:
                await self.check_once()
            except Exception:
                logger.exception(
                    "A round of checking for updates failed. Trying "
                    "again in %s seconds.",
                    self._every_seconds,
                )
            await asyncio.sleep(self._every_seconds)
    except asyncio.CancelledError:
        logger.info("Stopped checking for updates.")
        raise

poll async

poll(
    *,
    fetch: FetchUpdates,
    deliver: DeliverUpdate,
    every_seconds: float = 60.0,
    since: datetime | None = None,
    save_marker: SaveMarker | None = None,
) -> None

Check a network on a timer until this task is cancelled.

The short way to write Poller(...).run_forever(). Use Poller itself when you want to run a single round by hand, such as from a cron job.

Parameters:

Name	Type	Description	Default
fetch	FetchUpdates	Asks the network what it has right now, as updates.	required
deliver	DeliverUpdate	Where a new update goes.	required
every_seconds	float	How long to wait between rounds.	60.0
since	datetime | None	The marker you saved last time.	None
save_marker	SaveMarker | None	Called with the new marker whenever it moves.	None

Raises:

Type	Description
CancelledError	When the task is cancelled.

Source code in src/socialchimp/events.py

async def poll(
    *,
    fetch: FetchUpdates,
    deliver: DeliverUpdate,
    every_seconds: float = 60.0,
    since: datetime | None = None,
    save_marker: SaveMarker | None = None,
) -> None:
    """Check a network on a timer until this task is cancelled.

    The short way to write `Poller(...).run_forever()`. Use `Poller` itself
    when you want to run a single round by hand, such as from a cron job.

    Args:
        fetch: Asks the network what it has right now, as updates.
        deliver: Where a new update goes.
        every_seconds: How long to wait between rounds.
        since: The marker you saved last time.
        save_marker: Called with the new marker whenever it moves.

    Raises:
        asyncio.CancelledError: When the task is cancelled.
    """
    await Poller(
        fetch=fetch,
        deliver=deliver,
        every_seconds=every_seconds,
        since=since,
        save_marker=save_marker,
    ).run_forever()

SeenUpdates

Bases: Protocol

A memory of which updates have already been handled.

Every network promises to deliver at least once, which is a promise to deliver twice sometimes: a slow reply, a timeout, a retry after your server restarts. Without this, one comment can send two notifications or post two replies.

seen async

seen(update_id: str) -> bool

Say whether this update has already been handled.

Parameters:

Name	Type	Description	Default
update_id	str	The network's identifier for the update.	required

Returns:

Type	Description
bool	True if it has been handled before.

Source code in src/socialchimp/events.py

async def seen(self, update_id: str) -> bool:
    """Say whether this update has already been handled.

    Args:
        update_id: The network's identifier for the update.

    Returns:
        True if it has been handled before.
    """
    ...

remember async

remember(update_id: str) -> None

Note that this update has now been handled.

Parameters:

Name	Type	Description	Default
update_id	str	The network's identifier for the update.	required

Source code in src/socialchimp/events.py

async def remember(self, update_id: str) -> None:
    """Note that this update has now been handled.

    Args:
        update_id: The network's identifier for the update.
    """
    ...

InMemorySeenUpdates

InMemorySeenUpdates(max_size: int = 10000)

A memory that lives in one process and is lost on restart.

Good for tests, examples and a single small server. Not good for production: two workers do not share it, so the same update can be handled once by each, and a restart forgets everything just when a network is most likely to retry.

Back it with your database instead - a table of update ids with a unique index, and remember doing an insert that ignores duplicates. That gets you both workers agreeing and a memory that survives a restart.

The memory here is capped so that a busy account cannot fill up the process. Once it is full the oldest ids are forgotten first, on the basis that a network that is going to retry does so within minutes.

Start with an empty memory.

Parameters:

Name	Type	Description	Default
max_size	int	How many update ids to keep before forgetting the oldest.	10000

Source code in src/socialchimp/events.py

def __init__(self, max_size: int = 10_000) -> None:
    """Start with an empty memory.

    Args:
        max_size: How many update ids to keep before forgetting the
            oldest.
    """
    self._max_size = max_size
    self._ids: OrderedDict[str, None] = OrderedDict()

seen async

seen(update_id: str) -> bool

Say whether this update has already been handled.

Parameters:

Name	Type	Description	Default
update_id	str	The network's identifier for the update.	required

Returns:

Type	Description
bool	True if it has been handled before.

Source code in src/socialchimp/events.py

async def seen(self, update_id: str) -> bool:
    """Say whether this update has already been handled.

    Args:
        update_id: The network's identifier for the update.

    Returns:
        True if it has been handled before.
    """
    return update_id in self._ids

remember async

remember(update_id: str) -> None

Note that this update has now been handled.

Parameters:

Name	Type	Description	Default
update_id	str	The network's identifier for the update.	required

Source code in src/socialchimp/events.py

async def remember(self, update_id: str) -> None:
    """Note that this update has now been handled.

    Args:
        update_id: The network's identifier for the update.
    """
    self._ids[update_id] = None
    while len(self._ids) > self._max_size:
        self._ids.popitem(last=False)

Polling with a resumable marker

What account.fetch_updates_after(...) hands back - added in 0.8.0. See the social inbox use case for a worked example.

UpdateBatch dataclass

UpdateBatch(
    updates: tuple[Update, ...],
    marker: str | None,
    more: bool,
)

One page of updates read with fetch_updates_after.

Kept beside Update here rather than in socialchimp.models, because socialchimp.models imports nothing from anywhere else in socialchimp - see the note near the top of that file - and this shape is only ever built from an Update.

Attributes:

Name	Type	Description
updates	tuple[Update, ...]	What happened, oldest first, only the ones newer than the marker that was asked for.
marker	str | None	Store this and pass it back next time. None only when nothing has ever been seen on this account.
more	bool	True when the network has more new updates waiting beyond this page - call fetch_updates_after again straight away rather than waiting for the next round.