Writing a platform¶

You only need this page if you are adding support for a network socialchimp does not have yet - see Adding a platform for the story. Everyone else can skip straight to Networks.

The protocol¶

Every platform provides these seven methods. A platform never subclasses anything; matching the shape is enough.

Platform ¶

Bases: Protocol

What every platform file provides.

A platform is built with no arguments - MastodonPlatform() - because SocialChimp builds it for you from the name you asked for. Everything it needs for a particular account arrives as an argument: credentials on the LoginRequest, the account on the Connection. Keep nothing per account on the instance, so one platform can serve every account your app holds.

Attributes:

Name	Type	Description
`name`	`str`	How this network is named in code, for example `"mastodon"`.
`features`	`Feature`	What this network can do. Anything not listed here is refused with a clear message instead of being attempted.

api_base ¶

api_base(connection: Connection) -> str

Return where this network's API lives for this account.

Most networks have one address for everyone. Mastodon has a different one per server, which is why the connection is passed in rather than this being a plain attribute.

Parameters:

Name	Type	Description	Default
`connection`	`Connection`	The account we are about to act as.	required

Returns:

Type	Description
`str`	The address to send requests to, with no trailing slash, for
`str`	example `"https://graph.facebook.com/v21.0"`.

Source code in src/socialchimp/platform.py

def api_base(self, connection: Connection) -> str:
    """Return where this network's API lives for this account.

    Most networks have one address for everyone. Mastodon has a
    different one per server, which is why the connection is passed in
    rather than this being a plain attribute.

    Args:
        connection: The account we are about to act as.

    Returns:
        The address to send requests to, with no trailing slash, for
        example `"https://graph.facebook.com/v21.0"`.
    """
    ...

auth_headers ¶

auth_headers(connection: Connection) -> Mapping[str, str]

Return the headers that prove we may act as this account.

Usually one Authorization header. Called on every request, so keep it cheap and do not go to the network here - the token has already been renewed by the time this runs.

Parameters:

Name	Type	Description	Default
`connection`	`Connection`	The account we are acting as.	required

Returns:

Type	Description
`Mapping[str, str]`	Headers to add to the request.

Source code in src/socialchimp/platform.py

def auth_headers(self, connection: Connection) -> Mapping[str, str]:
    """Return the headers that prove we may act as this account.

    Usually one `Authorization` header. Called on every request, so
    keep it cheap and do not go to the network here - the token has
    already been renewed by the time this runs.

    Args:
        connection: The account we are acting as.

    Returns:
        Headers to add to the request.
    """
    ...

limits `async` ¶

limits(connection: Connection) -> Limits

Look up the numbers this network is enforcing right now.

Some of these genuinely change: a Mastodon server's post length is set by whoever runs it, and Instagram counts down how many posts are left today. Results are worth caching for a short while.

Parameters:

Name	Type	Description	Default
`connection`	`Connection`	The account to ask about.	required

Returns:

Type	Description
`Limits`	The current limits.

Source code in src/socialchimp/platform.py

async def limits(self, connection: Connection) -> Limits:
    """Look up the numbers this network is enforcing right now.

    Some of these genuinely change: a Mastodon server's post length is
    set by whoever runs it, and Instagram counts down how many posts are
    left today. Results are worth caching for a short while.

    Args:
        connection: The account to ask about.

    Returns:
        The current limits.
    """
    ...

start_login `async` ¶

start_login(request: LoginRequest) -> LoginStep

Begin signing someone in.

Most networks answer with SendToNetwork: redirect the person there and wait for them to come back. Networks that use an app password or a bot token answer with AskForDetails instead, because there is nowhere to send anyone - your app shows a form and passes the answers to finish_login.

Parameters:

Name	Type	Description	Default
`request`	`LoginRequest`	Where to send them back to, and what to ask for.	required

Returns:

Type	Description
`LoginStep`	What to do next.

Source code in src/socialchimp/platform.py

async def start_login(self, request: LoginRequest) -> LoginStep:
    """Begin signing someone in.

    Most networks answer with `SendToNetwork`: redirect the person there
    and wait for them to come back. Networks that use an app password or
    a bot token answer with `AskForDetails` instead, because there is
    nowhere to send anyone - your app shows a form and passes the answers
    to `finish_login`.

    Args:
        request: Where to send them back to, and what to ask for.

    Returns:
        What to do next.
    """
    ...

finish_login `async` ¶

finish_login(
    request: LoginRequest,
    callback: Mapping[str, str],
    remember: RawData | None = None,
) -> LoginStep

Carry on after the person comes back from the network.

Usually this finishes the job and returns Finished. Networks that need to know which page or channel to use return ChooseAccount first.

Parameters:

Name	Type	Description	Default
`request`	`LoginRequest`	The same request used to start the login.	required
`callback`	`Mapping[str, str]`	The query values the network sent back.	required
`remember`	`RawData \| None`	Whatever `start_login` put in `SendToNetwork.remember`.	`None`

Returns:

Type	Description
`LoginStep`	Either the finished connection or a question to ask.

Source code in src/socialchimp/platform.py

async def finish_login(
    self,
    request: LoginRequest,
    callback: Mapping[str, str],
    remember: RawData | None = None,
) -> LoginStep:
    """Carry on after the person comes back from the network.

    Usually this finishes the job and returns `Finished`. Networks that
    need to know which page or channel to use return `ChooseAccount`
    first.

    Args:
        request: The same request used to start the login.
        callback: The query values the network sent back.
        remember: Whatever `start_login` put in `SendToNetwork.remember`.

    Returns:
        Either the finished connection or a question to ask.
    """
    ...

refresh `async` ¶

refresh(
    connection: Connection,
    app: AppCredentials | None = None,
) -> Token

Get a fresh token for an account.

Called for you before a token runs out. A platform whose tokens do not expire can return the existing one unchanged.

Parameters:

Name	Type	Description	Default
`connection`	`Connection`	The account whose token is running out.	required
`app`	`AppCredentials \| None`	Your app's credentials for this network. Most networks want them to renew a token - Google, Meta and X all do. Networks that do not can ignore this.	`None`

Returns:

Type	Description
`Token`	The new token. Save it - if the network rotates refresh tokens,
`Token`	the old one has already stopped working.

Source code in src/socialchimp/platform.py

async def refresh(
    self,
    connection: Connection,
    app: AppCredentials | None = None,
) -> Token:
    """Get a fresh token for an account.

    Called for you before a token runs out. A platform whose tokens do
    not expire can return the existing one unchanged.

    Args:
        connection: The account whose token is running out.
        app: Your app's credentials for this network. Most networks want
            them to renew a token - Google, Meta and X all do. Networks
            that do not can ignore this.

    Returns:
        The new token. Save it - if the network rotates refresh tokens,
        the old one has already stopped working.
    """
    ...

publish `async` ¶

publish(connection: Connection, post: Post) -> PostResult

Publish a post.

Networks that publish in several steps, such as Instagram, do all of them here and return once the post is live or the network has taken over.

Parameters:

Name	Type	Description	Default
`connection`	`Connection`	The account to publish as.	required
`post`	`Post`	What to publish.	required

Returns:

Type	Description
`PostResult`	What the network said about the new post.

Source code in src/socialchimp/platform.py

async def publish(self, connection: Connection, post: Post) -> PostResult:
    """Publish a post.

    Networks that publish in several steps, such as Instagram, do all of
    them here and return once the post is live or the network has taken
    over.

    Args:
        connection: The account to publish as.
        post: What to publish.

    Returns:
        What the network said about the new post.
    """
    ...

What a platform can opt into¶

Anything a network cannot do is left off rather than stubbed - a platform with no create_app simply has no create_app method, and socialchimp asks before calling it.

CanCreateApp ¶

Bases: Protocol

Extra for networks that let us register an app automatically.

Mastodon is the only one today. Everywhere else you register your app by hand in a developer portal, and several networks review it before it works at all.

create_app `async` ¶

create_app(
    *,
    name: str,
    redirect_uri: str,
    host: str | None = None,
    scopes: tuple[str, ...] = (),
) -> AppCredentials

Register an app with the network and return its credentials.

Parameters:

Name	Type	Description	Default
`name`	`str`	The app name people will see when approving it.	required
`redirect_uri`	`str`	Where the network sends people back to.	required
`host`	`str \| None`	Which server to register on, for networks with many.	`None`
`scopes`	`tuple[str, ...]`	Permissions the app will ask for.	`()`

Returns:

Type	Description
`AppCredentials`	Credentials to save and reuse. Registering again for the same
`AppCredentials`	server wastes a record on that server, so save these.

Source code in src/socialchimp/platform.py

async def create_app(
    self,
    *,
    name: str,
    redirect_uri: str,
    host: str | None = None,
    scopes: tuple[str, ...] = (),
) -> AppCredentials:
    """Register an app with the network and return its credentials.

    Args:
        name: The app name people will see when approving it.
        redirect_uri: Where the network sends people back to.
        host: Which server to register on, for networks with many.
        scopes: Permissions the app will ask for.

    Returns:
        Credentials to save and reuse. Registering again for the same
        server wastes a record on that server, so save these.
    """
    ...

CanResumeLogin ¶

Bases: Protocol

Extra for networks that pause to ask which account to use.

Facebook asks which page, YouTube which channel. Those platforms answer finish_login with ChooseAccount, and finish the job here once the person has picked one.

resume_login `async` ¶

resume_login(
    request: LoginRequest,
    *,
    resume_token: str,
    account_id: str,
    remember: RawData | None = None,
) -> LoginStep

Carry on with the login, now that an account has been picked.

Parameters:

Name	Type	Description	Default
`request`	`LoginRequest`	The same request the login was started with.	required
`resume_token`	`str`	The value from `ChooseAccount`, handed straight back. Only this platform understands it.	required
`account_id`	`str`	Which of the offered accounts the person picked.	required
`remember`	`RawData \| None`	Whatever `start_login` put in `SendToNetwork.remember`, the same as `finish_login` was given.	`None`

Returns:

Type	Description
`LoginStep`	Usually the finished connection. A network that asks twice can
`LoginStep`	answer with another question instead.

Source code in src/socialchimp/platform.py

async def resume_login(
    self,
    request: LoginRequest,
    *,
    resume_token: str,
    account_id: str,
    remember: RawData | None = None,
) -> LoginStep:
    """Carry on with the login, now that an account has been picked.

    Args:
        request: The same request the login was started with.
        resume_token: The value from `ChooseAccount`, handed straight
            back. Only this platform understands it.
        account_id: Which of the offered accounts the person picked.
        remember: Whatever `start_login` put in `SendToNetwork.remember`,
            the same as `finish_login` was given.

    Returns:
        Usually the finished connection. A network that asks twice can
        answer with another question instead.
    """
    ...

CanDeletePosts ¶

Bases: Protocol

Extra for networks that let us remove a post we published.

delete_post `async` ¶

delete_post(connection: Connection, post_id: str) -> None

Remove a post.

Parameters:

Name	Type	Description	Default
`connection`	`Connection`	The account that published it.	required
`post_id`	`str`	The network's identifier for the post.	required

Source code in src/socialchimp/platform.py

async def delete_post(self, connection: Connection, post_id: str) -> None:
    """Remove a post.

    Args:
        connection: The account that published it.
        post_id: The network's identifier for the post.
    """
    ...

CanReadStats ¶

Bases: Protocol

Extra for networks that say how a published post is doing.

Mastodon counts replies, favourites and boosts, and hands all three back on the post itself. Plenty of networks keep nothing an app can read, and a few keep numbers behind a permission most apps never ask for - so this is an extra rather than something every platform has.

A platform with this also lists Feature.READ_STATS, because that flag is what socialchimp reads before calling. Account.read_stats is what your app calls; it checks the flag first and refuses plainly where a network keeps no numbers.

read_stats `async` ¶

read_stats(
    connection: Connection, post_id: str
) -> PostStats

Ask the network how a post is doing.

Called with both arguments by position, so the order matters and the names do not.

Parameters:

Name	Type	Description	Default
`connection`	`Connection`	The account the post belongs to, with a token that works right now.	required
`post_id`	`str`	The network's identifier for the post, which is what `publish` handed back.	required

Returns:

Type	Description
`PostStats`	The numbers that network keeps. Anything it does not count comes
`PostStats`	back as `None` rather than as a zero.

Source code in src/socialchimp/platform.py

async def read_stats(self, connection: Connection, post_id: str) -> PostStats:
    """Ask the network how a post is doing.

    Called with both arguments by position, so the order matters and
    the names do not.

    Args:
        connection: The account the post belongs to, with a token that
            works right now.
        post_id: The network's identifier for the post, which is what
            `publish` handed back.

    Returns:
        The numbers that network keeps. Anything it does not count comes
        back as `None` rather than as a zero.
    """
    ...

CanReadUpdates ¶

Bases: Protocol

Extra for networks we can ask "what has happened since?".

Used for networks that cannot tell us themselves. LinkedIn, Pinterest, Reddit and Tumblr all work this way: socialchimp calls this on a timer, works out what is new, and hands your app the same Update objects a pushing network would have produced. Your handlers never learn which kind of network they are dealing with.

A network that pushes updates should say so with Feature.PUSH_UPDATES and provide CanCheckSignature instead. Providing both is fine, and lets an app fall back to checking on a timer if it cannot receive incoming requests.

fetch_updates `async` ¶

fetch_updates(
    connection: Connection, since: datetime | None
) -> Sequence[Update]

Return what has happened on this account since a moment in time.

Parameters:

Name	Type	Description	Default
`connection`	`Connection`	The account to ask about.	required
`since`	`datetime \| None`	Only return things newer than this. `None` on the first call, when there is no marker saved yet - return a recent page rather than the whole history.	required

Returns:

Type	Description
`Sequence[Update]`	The updates, oldest first.

Source code in src/socialchimp/platform.py

async def fetch_updates(
    self,
    connection: Connection,
    since: datetime | None,
) -> Sequence[Update]:
    """Return what has happened on this account since a moment in time.

    Args:
        connection: The account to ask about.
        since: Only return things newer than this. `None` on the first
            call, when there is no marker saved yet - return a recent
            page rather than the whole history.

    Returns:
        The updates, oldest first.
    """
    ...

CanCheckSignature ¶

Bases: Protocol

Extra for networks that send us requests when something happens.

Every network signs these differently: Meta uses HMAC-SHA256 in a header, Telegram echoes a shared secret, Discord signs with Ed25519. A platform file knows which, and this is where it says so.

The check must work on the raw bytes of the request, exactly as they arrived. Any framework that parses the JSON and builds it again first will change the bytes and break the signature, so never accept a parsed body here.

check_signature ¶

check_signature(
    body: bytes, headers: Mapping[str, str], *, secret: str
) -> None

Check an incoming request really came from the network.

Parameters:

Name	Type	Description	Default
`body`	`bytes`	The request body, untouched.	required
`headers`	`Mapping[str, str]`	The request headers.	required
`secret`	`str`	The shared secret for this network, from your settings.	required

Raises:

Type	Description
`SignatureError`	If the request cannot be trusted. Answer 401 and do nothing else with it.

Source code in src/socialchimp/platform.py

def check_signature(
    self,
    body: bytes,
    headers: Mapping[str, str],
    *,
    secret: str,
) -> None:
    """Check an incoming request really came from the network.

    Args:
        body: The request body, untouched.
        headers: The request headers.
        secret: The shared secret for this network, from your settings.

    Raises:
        SignatureError: If the request cannot be trusted. Answer 401 and
            do nothing else with it.
    """
    ...

read_update ¶

read_update(
    body: bytes, headers: Mapping[str, str]
) -> Update

Turn a checked request into an update your app understands.

Only call this after check_signature has passed.

Parameters:

Name	Type	Description	Default
`body`	`bytes`	The request body, untouched.	required
`headers`	`Mapping[str, str]`	The request headers.	required

Returns:

Type	Description
`Update`	What happened, in socialchimp's own words.

Source code in src/socialchimp/platform.py

def read_update(
    self,
    body: bytes,
    headers: Mapping[str, str],
) -> Update:
    """Turn a checked request into an update your app understands.

    Only call this after `check_signature` has passed.

    Args:
        body: The request body, untouched.
        headers: The request headers.

    Returns:
        What happened, in socialchimp's own words.
    """
    ...

CanCheckState ¶

Bases: Protocol

Extra for networks that keep working after they accept a post.

YouTube encodes a video for minutes, sometimes hours. TikTok can put one in somebody's drafts instead of publishing it. Both answer publish before they have finished, so a PostResult that comes back PROCESSING is not the end of the story - this is how an app finds out the rest of it.

Account.check_state is what your app calls. It looks for this, renews the token, and hands the connection down.

check_state `async` ¶

check_state(
    connection: Connection, post_id: str
) -> PostResult

Ask the network how far it has got with a post.

Called with both arguments by position, so the order matters and the names do not.

Parameters:

Name	Type	Description	Default
`connection`	`Connection`	The account the post belongs to, with a token that works right now.	required
`post_id`	`str`	The network's identifier for the post, which is what `publish` handed back.	required

Returns:

Type	Description
`PostResult`	Where the post has got to now. The same shape `publish` gave,
`PostResult`	so an app can treat the two the same way.

Source code in src/socialchimp/platform.py

async def check_state(self, connection: Connection, post_id: str) -> PostResult:
    """Ask the network how far it has got with a post.

    Called with both arguments by position, so the order matters and
    the names do not.

    Args:
        connection: The account the post belongs to, with a token that
            works right now.
        post_id: The network's identifier for the post, which is what
            `publish` handed back.

    Returns:
        Where the post has got to now. The same shape `publish` gave,
        so an app can treat the two the same way.
    """
    ...

CanAnswerSetupCheck ¶

Bases: Protocol

Extra for networks that ask a question before they will push anything.

Facebook, Instagram and Threads all do this. Point Meta at a URL of yours and it does a GET to it first, carrying a token you chose and a challenge to echo back. Get it wrong and Meta says the URL could not be verified, without saying why.

This happens before anybody has connected an account, so there is no connection to hang it on. SocialChimp.answer_setup_check is what your app calls.

answer_setup_check ¶

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

Answer the one-off check and hand back what to reply with.

Parameters:

Name	Type	Description	Default
`params`	`Mapping[str, str]`	The query values from that GET, such as Django's `request.GET` or FastAPI's `request.query_params`.	required
`verify_token`	`str`	The token you typed into the network's own form.	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/platform.py

def answer_setup_check(
    self,
    params: Mapping[str, str],
    *,
    verify_token: str,
) -> str:
    """Answer the one-off check and hand back what to reply with.

    Args:
        params: The query values from that GET, such as Django's
            `request.GET` or FastAPI's `request.query_params`.
        verify_token: The token you typed into the network's own form.

    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.
    """
    ...

CanReadPushedUpdates ¶

Bases: Protocol

Extra for reading everything one pushed request carries.

Not the same as CanReadUpdates, which asks a network what has happened since. This is for a request the network sent us, and it hands back the whole of what that request held.

Which matters because Meta batches. One message from Facebook can carry changes for several pages, and several changes for each of them, and it does that when it is busy - exactly when you least want to drop the rest. read_update on CanCheckSignature gives you the first one only.

A network that never batches still has this, handing back a list of one, so that an app written against one network works against all of them.

SocialChimp.read_updates is what your app calls, after SocialChimp.check_signature has passed.

read_updates ¶

read_updates(body: bytes) -> list[Update]

Turn a checked request into every update it carries.

Parameters:

Name	Type	Description	Default
`body`	`bytes`	The request body, exactly as it arrived. Check its signature first.	required

Returns:

Type	Description
`list[Update]`	What happened, in the order the network listed it. Empty when
`list[Update]`	the message carried nothing we can act on, which is not an
`list[Update]`	error - networks send shapes we have no interest in.

Source code in src/socialchimp/platform.py

def read_updates(self, body: bytes) -> list[Update]:
    """Turn a checked request into every update it carries.

    Args:
        body: The request body, exactly as it arrived. Check its
            signature first.

    Returns:
        What happened, in the order the network listed it. Empty when
        the message carried nothing we can act on, which is not an
        error - networks send shapes we have no interest in.
    """
    ...

Added in 0.8.0. See the use case for how an app calls these through Account, and docs/social-inbox-contract.md in the repository for the design behind them.

CanReadPost ¶

Bases: Protocol

Extra for reading one post back in full.

Not only what publishing it returned - everything socialchimp models about a post. Account.read_post is what your app calls.

read_post `async` ¶

read_post(
    connection: Connection, post_id: str
) -> PostDetails

Read one post, with everything socialchimp models about it.

Parameters:

Name	Type	Description	Default
`connection`	`Connection`	The account to read it as.	required
`post_id`	`str`	The network's identifier for the post or comment.	required

Returns:

Type	Description
`PostDetails`	The post, in full.

Raises:

Type	Description
`SocialChimpError`	If the network refuses, or the post is gone.

Source code in src/socialchimp/platform.py

async def read_post(self, connection: Connection, post_id: str) -> PostDetails:
    """Read one post, with everything socialchimp models about it.

    Args:
        connection: The account to read it as.
        post_id: The network's identifier for the post or comment.

    Returns:
        The post, in full.

    Raises:
        SocialChimpError: If the network refuses, or the post is gone.
    """
    ...

CanReadThread ¶

Bases: Protocol

Extra for reading a post together with its replies.

Account.read_thread is what your app calls.

read_thread `async` ¶

read_thread(
    connection: Connection,
    post_id: str,
    *,
    depth: int | None = None,
    limit: int | None = None,
) -> Thread

Read a post and the replies underneath it.

Parameters:

Name	Type	Description	Default
`connection`	`Connection`	The account to read it as.	required
`post_id`	`str`	The network's identifier for the post to read.	required
`depth`	`int \| None`	How many reply levels to fetch. `None` uses the network's own default.	`None`
`limit`	`int \| None`	A cap on how many replies come back.	`None`

Returns:

Type	Description
`Thread`	The post and its replies. `Thread.complete` is `False` if
`Thread`	`depth`, `limit` or one of the network's own caps cut the
`Thread`	replies off before the end.

Source code in src/socialchimp/platform.py

async def read_thread(
    self,
    connection: Connection,
    post_id: str,
    *,
    depth: int | None = None,
    limit: int | None = None,
) -> Thread:
    """Read a post and the replies underneath it.

    Args:
        connection: The account to read it as.
        post_id: The network's identifier for the post to read.
        depth: How many reply levels to fetch. `None` uses the
            network's own default.
        limit: A cap on how many replies come back.

    Returns:
        The post and its replies. `Thread.complete` is `False` if
        `depth`, `limit` or one of the network's own caps cut the
        replies off before the end.
    """
    ...

CanReply ¶

Bases: Protocol

Extra for replying to any post or comment, at any depth.

publish(Post(reply_to=...)) keeps working without this. This is the recommended way to reply once a network provides it, because it can do things publish cannot know to - keeping a Mastodon reply's visibility no wider than its parent's, or mentioning the people already in the thread.

Account.reply is what your app calls.

reply `async` ¶

reply(
    connection: Connection,
    post_id: str,
    text: str,
    *,
    media: tuple[Media, ...] = (),
    options: RawData | None = None,
) -> PostResult

Reply to a post or comment.

Parameters:

Name	Type	Description	Default
`connection`	`Connection`	The account to reply as.	required
`post_id`	`str`	The post or comment being replied to, at any depth.	required
`text`	`str`	The reply's words.	required
`media`	`tuple[Media, ...]`	Pictures or videos to attach to the reply.	`()`
`options`	`RawData \| None`	Settings for one network only.	`None`

Returns:

Type	Description
`PostResult`	What the network said about the new reply.

Raises:

Type	Description
`SocialChimpError`	If the network refuses, or the target is gone.

Source code in src/socialchimp/platform.py

async def reply(
    self,
    connection: Connection,
    post_id: str,
    text: str,
    *,
    media: tuple[Media, ...] = (),
    options: RawData | None = None,
) -> PostResult:
    """Reply to a post or comment.

    Args:
        connection: The account to reply as.
        post_id: The post or comment being replied to, at any depth.
        text: The reply's words.
        media: Pictures or videos to attach to the reply.
        options: Settings for one network only.

    Returns:
        What the network said about the new reply.

    Raises:
        SocialChimpError: If the network refuses, or the target is gone.
    """
    ...

CanLike ¶

Bases: Protocol

Extra for liking and unliking a post or a comment.

Both calls are idempotent: liking something twice, or unliking something not liked, succeeds and does nothing.

Account.like and Account.unlike are what your app calls.

like `async` ¶

like(connection: Connection, post_id: str) -> LikeResult

Like a post or a comment.

Parameters:

Name	Type	Description	Default
`connection`	`Connection`	The account doing the liking.	required
`post_id`	`str`	The post or comment to like.	required

Returns:

Type	Description
`LikeResult`	What the network said about the like. Liking something already
`LikeResult`	liked returns the existing like rather than making a new one.

Source code in src/socialchimp/platform.py

async def like(self, connection: Connection, post_id: str) -> LikeResult:
    """Like a post or a comment.

    Args:
        connection: The account doing the liking.
        post_id: The post or comment to like.

    Returns:
        What the network said about the like. Liking something already
        liked returns the existing like rather than making a new one.
    """
    ...

unlike `async` ¶

unlike(
    connection: Connection,
    post_id: str,
    *,
    like_id: str | None = None,
) -> None

Take back a like.

Parameters:

Name	Type	Description	Default
`connection`	`Connection`	The account taking the like back.	required
`post_id`	`str`	The post or comment to unlike.	required
`like_id`	`str \| None`	The like's own identifier, from `LikeResult.like_id`, where passing it saves a lookup. Left out, the network is asked which like to remove.	`None`

Raises:

Type	Description
`SocialChimpError`	If the network refuses.

Source code in src/socialchimp/platform.py

async def unlike(
    self,
    connection: Connection,
    post_id: str,
    *,
    like_id: str | None = None,
) -> None:
    """Take back a like.

    Args:
        connection: The account taking the like back.
        post_id: The post or comment to unlike.
        like_id: The like's own identifier, from `LikeResult.like_id`,
            where passing it saves a lookup. Left out, the network is
            asked which like to remove.

    Raises:
        SocialChimpError: If the network refuses.
    """
    ...

CanReadLikes ¶

Bases: Protocol

Extra for listing who liked a post.

Some networks that can like something cannot list who did - a Facebook Page can like a comment but only ever sees the count - so this is a separate extra from CanLike rather than part of it.

Account.read_likes is what your app calls.

read_likes `async` ¶

read_likes(
    connection: Connection,
    post_id: str,
    *,
    after: str | None = None,
    limit: int | None = None,
) -> Page[Like]

List who liked a post.

Parameters:

Name	Type	Description	Default
`connection`	`Connection`	The account to ask as.	required
`post_id`	`str`	The post or comment to list likes for.	required
`after`	`str \| None`	A `Page.next` from a previous call, to read further in.	`None`
`limit`	`int \| None`	A cap on how many come back. `None` uses the network's own default.	`None`

Returns:

Type	Description
`Page[Like]`	One page of likes.

Source code in src/socialchimp/platform.py

async def read_likes(
    self,
    connection: Connection,
    post_id: str,
    *,
    after: str | None = None,
    limit: int | None = None,
) -> Page[Like]:
    """List who liked a post.

    Args:
        connection: The account to ask as.
        post_id: The post or comment to list likes for.
        after: A `Page.next` from a previous call, to read further in.
        limit: A cap on how many come back. `None` uses the network's
            own default.

    Returns:
        One page of likes.
    """
    ...

CanReadUpdatesAfter ¶

Bases: Protocol

Extra for polling with a marker that can be resumed after a restart.

Unlike CanReadUpdates.fetch_updates, which takes a moment in time, this takes an opaque marker your app stores and passes back - see UpdateBatch. That is what makes it resumable: a moment in time can miss or repeat updates around the edges, where a marker cannot.

Account.fetch_updates_after and Account.mark_seen are what your app calls.

fetch_updates_after `async` ¶

fetch_updates_after(
    connection: Connection,
    marker: str | None,
    *,
    limit: int | None = None,
) -> UpdateBatch

Read what is new since a marker.

Parameters:

Name	Type	Description	Default
`connection`	`Connection`	The account to ask about.	required
`marker`	`str \| None`	The marker from the last call's `UpdateBatch.marker`. `None` on the first call, when there is nothing saved yet - the network answers with its latest page instead, which sets a starting point.	required
`limit`	`int \| None`	A cap on how many updates come back in this page.	`None`

Returns:

Type	Description
`UpdateBatch`	The new updates, and a marker to store for next time.

Source code in src/socialchimp/platform.py

async def fetch_updates_after(
    self,
    connection: Connection,
    marker: str | None,
    *,
    limit: int | None = None,
) -> UpdateBatch:
    """Read what is new since a marker.

    Args:
        connection: The account to ask about.
        marker: The marker from the last call's `UpdateBatch.marker`.
            `None` on the first call, when there is nothing saved yet -
            the network answers with its latest page instead, which
            sets a starting point.
        limit: A cap on how many updates come back in this page.

    Returns:
        The new updates, and a marker to store for next time.
    """
    ...

mark_seen `async` ¶

mark_seen(connection: Connection, marker: str) -> None

Tell the network a marker has been seen.

Parameters:

Name	Type	Description	Default
`connection`	`Connection`	The account to mark it for.	required
`marker`	`str`	The marker that has been handled.	required

Source code in src/socialchimp/platform.py

async def mark_seen(self, connection: Connection, marker: str) -> None:
    """Tell the network a marker has been seen.

    Args:
        connection: The account to mark it for.
        marker: The marker that has been handled.
    """
    ...

CanMessage ¶

Bases: Protocol

Extra for reading and sending direct messages.

Account.read_conversations, Account.read_messages, Account.send_message and Account.mark_read are what your app calls.

read_conversations `async` ¶

read_conversations(
    connection: Connection,
    *,
    after: str | None = None,
    limit: int | None = None,
) -> Page[Conversation]

List this account's conversations.

Parameters:

Name	Type	Description	Default
`connection`	`Connection`	The account to ask as.	required
`after`	`str \| None`	A `Page.next` from a previous call.	`None`
`limit`	`int \| None`	A cap on how many come back.	`None`

Returns:

Type	Description
`Page[Conversation]`	One page of conversations.

Source code in src/socialchimp/platform.py

async def read_conversations(
    self,
    connection: Connection,
    *,
    after: str | None = None,
    limit: int | None = None,
) -> Page[Conversation]:
    """List this account's conversations.

    Args:
        connection: The account to ask as.
        after: A `Page.next` from a previous call.
        limit: A cap on how many come back.

    Returns:
        One page of conversations.
    """
    ...

read_messages `async` ¶

read_messages(
    connection: Connection,
    conversation_id: str,
    *,
    after: str | None = None,
    limit: int | None = None,
) -> Page[Message]

Read the messages in one conversation, newest first.

Parameters:

Name	Type	Description	Default
`connection`	`Connection`	The account to ask as.	required
`conversation_id`	`str`	Which conversation to read.	required
`after`	`str \| None`	A `Page.next` from a previous call. Passing it goes further back in time.	`None`
`limit`	`int \| None`	A cap on how many come back.	`None`

Returns:

Type	Description
`Page[Message]`	One page of messages, newest first.

Source code in src/socialchimp/platform.py

async def read_messages(
    self,
    connection: Connection,
    conversation_id: str,
    *,
    after: str | None = None,
    limit: int | None = None,
) -> Page[Message]:
    """Read the messages in one conversation, newest first.

    Args:
        connection: The account to ask as.
        conversation_id: Which conversation to read.
        after: A `Page.next` from a previous call. Passing it goes
            further back in time.
        limit: A cap on how many come back.

    Returns:
        One page of messages, newest first.
    """
    ...

send_message `async` ¶

send_message(
    connection: Connection,
    conversation_id: str,
    text: str,
    *,
    options: RawData | None = None,
) -> Message

Send a message into an existing conversation.

Parameters:

Name	Type	Description	Default
`connection`	`Connection`	The account to send as.	required
`conversation_id`	`str`	Which conversation to send into.	required
`text`	`str`	The message's words.	required
`options`	`RawData \| None`	Settings for one network only, such as a Meta message tag.	`None`

Returns:

Type	Description
`Message`	The message that was sent.

Raises:

Type	Description
`ReplyWindowClosedError`	If a 24-hour reply window has closed.

Source code in src/socialchimp/platform.py

async def send_message(
    self,
    connection: Connection,
    conversation_id: str,
    text: str,
    *,
    options: RawData | None = None,
) -> Message:
    """Send a message into an existing conversation.

    Args:
        connection: The account to send as.
        conversation_id: Which conversation to send into.
        text: The message's words.
        options: Settings for one network only, such as a Meta message
            tag.

    Returns:
        The message that was sent.

    Raises:
        ReplyWindowClosedError: If a 24-hour reply window has closed.
    """
    ...

mark_read `async` ¶

mark_read(
    connection: Connection, conversation_id: str
) -> None

Mark a conversation as read.

Parameters:

Name	Type	Description	Default
`connection`	`Connection`	The account to mark it for.	required
`conversation_id`	`str`	Which conversation to mark.	required

Source code in src/socialchimp/platform.py

async def mark_read(self, connection: Connection, conversation_id: str) -> None:
    """Mark a conversation as read.

    Args:
        connection: The account to mark it for.
        conversation_id: Which conversation to mark.
    """
    ...

CanStartConversations ¶

Bases: Protocol

Extra for starting a new conversation, rather than only answering one.

Meta cannot do this: the customer has to write first. A platform with CanMessage but not this one can still be replied to - it just cannot open the first message.

Account.start_conversation is what your app calls.

start_conversation `async` ¶

start_conversation(
    connection: Connection,
    person_ids: Sequence[str],
    text: str,
) -> Message

Start a conversation with one or more people.

Parameters:

Name	Type	Description	Default
`connection`	`Connection`	The account to send as.	required
`person_ids`	`Sequence[str]`	Who to start it with.	required
`text`	`str`	The first message's words.	required

Returns:

Type	Description
`Message`	The message that was sent.

Source code in src/socialchimp/platform.py

async def start_conversation(
    self,
    connection: Connection,
    person_ids: Sequence[str],
    text: str,
) -> Message:
    """Start a conversation with one or more people.

    Args:
        connection: The account to send as.
        person_ids: Who to start it with.
        text: The first message's words.

    Returns:
        The message that was sent.
    """
    ...

Signing someone in¶

start_login and finish_login return one of these four. socialchimp's type for "one of these four" is LoginStep = SendToNetwork | AskForDetails | ChooseAccount | Finished.

LoginRequest `dataclass` ¶

LoginRequest(
    redirect_uri: str,
    scopes: tuple[str, ...] = (),
    host: str | None = None,
    state: str | None = None,
    app: AppCredentials | None = None,
)

What we need in order to start signing someone in.

Attributes:

Name	Type	Description
`redirect_uri`	`str`	Where the network sends the person back to. It must match what the network's developer portal has on file.
`scopes`	`tuple[str, ...]`	Permissions to ask for. Each platform's page lists sensible defaults; leaving this empty uses them.
`host`	`str \| None`	Which server, for networks that have more than one. Required for Mastodon, ignored elsewhere.
`state`	`str \| None`	A value handed back to you at the end, so you can tell which of your users came back. One is made for you if you leave it out.
`app`	`AppCredentials \| None`	Your app's credentials for this network. `SocialChimp` fills this in from storage; you only set it yourself if you are calling a platform directly.

SendToNetwork `dataclass` ¶

SendToNetwork(
    url: str, state: str, remember: RawData = dict()
)

Step one: send the person to the network to approve your app.

Attributes:

Name	Type	Description
`url`	`str`	Where to send them. Redirect their browser here.
`state`	`str`	The value that will come back, for matching up the reply.
`remember`	`RawData`	Something the platform needs again when the person comes back, such as the secret half of a PKCE pair. Keep it with the rest of that person's session and hand it back to `finish_login`. It has to travel through your app because the two halves of a sign-in can happen in different processes - the person may be sent away by one web worker and come back to another. Holding it in memory would work on your laptop and fail in production.

LoginField `dataclass` ¶

LoginField(
    name: str,
    label: str,
    secret: bool = False,
    help_text: str | None = None,
)

One thing to ask a person for.

Attributes:

Name	Type	Description
`name`	`str`	What to call this value when handing it back. Put it in the `callback` mapping given to `finish_login` under this name.
`label`	`str`	What to show next to the box, in words a person understands.
`secret`	`bool`	True for anything that should not be shown as it is typed, or written to a log.
`help_text`	`str \| None`	A sentence under the box, usually saying where on the network to find the value.

AskForDetails `dataclass` ¶

AskForDetails(
    fields: tuple[LoginField, ...],
    help_url: str | None = None,
)

Step one, for networks that have no sign-in page to send people to.

Bluesky uses an app password, and Discord and Telegram use a bot token that someone pastes in. There is nowhere to redirect to, so instead the platform says what to ask for, your app shows a form, and the answers go back through finish_login as the callback mapping.

Show the fields in the order given, and never log anything marked secret.

Nothing leaves your app on this route, so LoginRequest.state is not used - there is no trip through a browser to match up afterwards. If your sign-in code expects every step to carry state back, this is the one that will not.

Attributes:

Name	Type	Description
`fields`	`tuple[LoginField, ...]`	What to ask for.
`help_url`	`str \| None`	A page explaining where to get these, worth linking to beside the form.

AccountChoice `dataclass` ¶

AccountChoice(id: str, name: str, kind: str | None = None)

One of several accounts a person could connect.

Attributes:

Name	Type	Description
`id`	`str`	The network's identifier, passed back to carry on.
`name`	`str`	Something a person would recognise, shown in your UI.
`kind`	`str \| None`	What sort of thing it is, such as `"page"` or `"channel"`.

ChooseAccount `dataclass` ¶

ChooseAccount(
    options: tuple[AccountChoice, ...], resume_token: str
)

A pause: the person has approved, but we need to know which account.

Facebook asks which page, YouTube which channel. Show options, then carry on with the one they picked.

Attributes:

Name	Type	Description
`options`	`tuple[AccountChoice, ...]`	What they can choose from.
`resume_token`	`str`	Hand this back to carry on. Treat it as meaningless text; only the platform file understands it. Treat it as a secret. On some networks it has to carry the tokens themselves, because the sign-in code can only be swapped once and that happens before the person picks. Keep it with their session, the way you keep `SendToNetwork.remember`. Do not put it in a URL, a hidden form field, or a log.

Finished `dataclass` ¶

Finished(connection: Connection)

The last step: the account is connected.

Attributes:

Name	Type	Description
`connection`	`Connection`	Save this. It is everything needed to act as the account.

Finding installed platforms¶

How SocialChimp turns a name like "facebook" into a platform instance. See socialchimp.registry for the full story of how packages register themselves.

register_platform ¶

register_platform(
    name: str, platform_class: type[Platform]
) -> None

Tell socialchimp about a platform, without installing a package.

A platform registered this way is used in place of an installed one with the same name, which is how a test swaps in a fake.

Parameters:

Name	Type	Description	Default
`name`	`str`	How the platform will be asked for, for example `"mastodon"`.	required
`platform_class`	`type[Platform]`	The class to use. It is not created here; that happens when the platform is used.	required

Raises:

Type	Description
`ConfigError`	If the class does not provide what a platform must.

Source code in src/socialchimp/registry.py

def register_platform(name: str, platform_class: type[Platform]) -> None:
    """Tell socialchimp about a platform, without installing a package.

    A platform registered this way is used in place of an installed one with
    the same name, which is how a test swaps in a fake.

    Args:
        name: How the platform will be asked for, for example `"mastodon"`.
        platform_class: The class to use. It is not created here; that
            happens when the platform is used.

    Raises:
        ConfigError: If the class does not provide what a platform must.
    """
    _check_it_is_a_platform(name, platform_class)
    _registered[name] = platform_class

unregister_platform ¶

unregister_platform(name: str) -> None

Forget a platform that was registered in code.

Quiet if there was nothing registered under that name. Installed packages are left alone - this only undoes register_platform.

Parameters:

Name	Type	Description	Default
`name`	`str`	The name it was registered under.	required

Source code in src/socialchimp/registry.py

def unregister_platform(name: str) -> None:
    """Forget a platform that was registered in code.

    Quiet if there was nothing registered under that name. Installed
    packages are left alone - this only undoes `register_platform`.

    Args:
        name: The name it was registered under.
    """
    _registered.pop(name, None)

available_platforms ¶

available_platforms() -> list[str]

List every platform socialchimp can use right now.

Returns:

Type	Description
`list[str]`	The names, in alphabetical order, from both places we look. A name
`list[str]`	appearing here does not promise its package imports cleanly - a
`list[str]`	broken package says so when you ask for it.

Source code in src/socialchimp/registry.py

def available_platforms() -> list[str]:
    """List every platform socialchimp can use right now.

    Returns:
        The names, in alphabetical order, from both places we look. A name
        appearing here does not promise its package imports cleanly - a
        broken package says so when you ask for it.
    """
    return sorted(set(_registered) | set(_find_installed()))

get_platform_class ¶

get_platform_class(name: str) -> type[Platform]

Find the class for one platform.

The class is imported the first time it is asked for and remembered after that.

Parameters:

Name	Type	Description	Default
`name`	`str`	Which network, for example `"mastodon"`.	required

Returns:

Type	Description
`type[Platform]`	The platform class. Create it to use it.

Raises:

Type	Description
`ConfigError`	If nothing is installed under that name, if its package could not be imported, or if what came back is not a platform.

Source code in src/socialchimp/registry.py

def get_platform_class(name: str) -> type[Platform]:
    """Find the class for one platform.

    The class is imported the first time it is asked for and remembered
    after that.

    Args:
        name: Which network, for example `"mastodon"`.

    Returns:
        The platform class. Create it to use it.

    Raises:
        ConfigError: If nothing is installed under that name, if its package
            could not be imported, or if what came back is not a platform.
    """
    if name in _registered:
        return _registered[name]
    if name in _imported:
        return _imported[name]

    installed = _find_installed()
    if name not in installed:
        raise ConfigError(_no_such_platform(name))

    platform_class = _import_platform(name, installed[name])
    _imported[name] = platform_class
    return platform_class

clear_platform_cache ¶

clear_platform_cache() -> None

Look for installed platforms again next time one is asked for.

socialchimp reads the installed packages once and remembers what it found. Call this after installing a package while the program is running, and between tests. Platforms registered in code are kept.

Source code in src/socialchimp/registry.py

def clear_platform_cache() -> None:
    """Look for installed platforms again next time one is asked for.

    socialchimp reads the installed packages once and remembers what it
    found. Call this after installing a package while the program is
    running, and between tests. Platforms registered in code are kept.
    """
    global _installed
    _installed = None
    _imported.clear()

Making requests¶

Every platform file sends its requests through HttpClient: retrying after a hiccup, waiting as long as a network asks, and turning an unhappy reply into a socialchimp error, written once instead of nine times.

HttpClient ¶

HttpClient(
    base_url: str = "",
    *,
    platform: str = _UNNAMED,
    headers: Mapping[str, str] | None = None,
    timeout: float | Timeout = _DEFAULT_TIMEOUT,
    transport: AsyncBaseTransport | None = None,
    retries: Retries | None = None,
    errors: Callable[[Response], SocialChimpError]
    | None = None,
)

The shared way to send requests to one network.

Wraps an httpx.AsyncClient and adds the parts every platform needs: trying again after a hiccup, waiting as long as the network asks, remembering how much of the allowance is left, and raising a socialchimp error instead of handing back a reply nobody checked.

A failed request is sent again with exactly the arguments you gave, so pass bytes rather than an open file - a file read once cannot be read again.

Example

async with HttpClient( "https://mastodon.social", platform="mastodon", ) as http: me = await http.json("GET", "/api/v1/accounts/verify_credentials")

Set up a client for one network.

Parameters:

Name	Type	Description	Default
`base_url`	`str`	What every path is joined onto, such as `"https://mastodon.social"`.	`''`
`platform`	`str`	Which network this talks to, used in error messages.	`_UNNAMED`
`headers`	`Mapping[str, str] \| None`	Sent with every request. A token usually goes here.	`None`
`timeout`	`float \| Timeout`	Seconds to wait for a reply before giving up on it.	`_DEFAULT_TIMEOUT`
`transport`	`AsyncBaseTransport \| None`	Where requests actually go. Leave it out for ordinary network calls; pass your own to send them through something else, which is also how tests answer without a network.	`None`
`retries`	`Retries \| None`	How many times to try again, and how long to wait in between. Left out, four tries with growing waits.	`None`
`errors`	`Callable[[Response], SocialChimpError] \| None`	Your own function turning an unhappy reply into an error, for a network with quirks worth naming. Left out, `error_from_response` is used.	`None`

Source code in src/socialchimp/http.py

def __init__(
    self,
    base_url: str = "",
    *,
    platform: str = _UNNAMED,
    headers: Mapping[str, str] | None = None,
    timeout: float | httpx.Timeout = _DEFAULT_TIMEOUT,
    transport: httpx.AsyncBaseTransport | None = None,
    retries: Retries | None = None,
    errors: Callable[[httpx.Response], SocialChimpError] | None = None,
) -> None:
    """Set up a client for one network.

    Args:
        base_url: What every path is joined onto, such as
            `"https://mastodon.social"`.
        platform: Which network this talks to, used in error messages.
        headers: Sent with every request. A token usually goes here.
        timeout: Seconds to wait for a reply before giving up on it.
        transport: Where requests actually go. Leave it out for ordinary
            network calls; pass your own to send them through something
            else, which is also how tests answer without a network.
        retries: How many times to try again, and how long to wait in
            between. Left out, four tries with growing waits.
        errors: Your own function turning an unhappy reply into an
            error, for a network with quirks worth naming. Left out,
            `error_from_response` is used.
    """
    self.platform = platform
    self.retries = retries if retries is not None else Retries()
    self._errors = errors
    self._rate_limit: RateLimit | None = None
    self._client = httpx.AsyncClient(
        base_url=base_url,
        headers=dict(headers) if headers is not None else None,
        timeout=timeout,
        transport=transport,
    )

rate_limit `property` ¶

rate_limit: RateLimit | None

What the network last said about how much allowance is left.

None until a reply mentions it. A later reply that says nothing leaves the last figures alone, because "no headers" means "no news", not "nothing left".

is_closed `property` ¶

is_closed: bool

Whether this client has been closed and cannot send any more.

request `async` ¶

request(
    method: str, path: str, **kwargs: object
) -> Response

Send a request, trying again if it is worth it.

Parameters:

Name	Type	Description	Default
`method`	`str`	`"GET"`, `"POST"` and so on.	required
`path`	`str`	Joined onto the base url.	required
`**kwargs`	`object`	Anything `httpx.AsyncClient.request` takes, such as `params`, `json`, `content`, `files` or extra `headers`.	`{}`

Returns:

Type	Description
`Response`	The reply, which is always one the network was happy with.

Raises:

Type	Description
`SocialChimpError`	If the network refused, or could not be reached at all. Which error depends on what it said; see `error_from_response`.

Source code in src/socialchimp/http.py

async def request(
    self,
    method: str,
    path: str,
    **kwargs: object,
) -> httpx.Response:
    """Send a request, trying again if it is worth it.

    Args:
        method: `"GET"`, `"POST"` and so on.
        path: Joined onto the base url.
        **kwargs: Anything `httpx.AsyncClient.request` takes, such as
            `params`, `json`, `content`, `files` or extra `headers`.

    Returns:
        The reply, which is always one the network was happy with.

    Raises:
        SocialChimpError: If the network refused, or could not be
            reached at all. Which error depends on what it said; see
            `error_from_response`.
    """
    # httpx already names every option it takes - params, json, files
    # and the rest - so we hand them straight on rather than writing that
    # list out a second time here.
    options = cast("dict[str, Any]", kwargs)

    failures = 0
    while True:
        try:
            response = await self._client.request(method, path, **options)
        except httpx.TransportError as problem:
            failures += 1
            if failures >= self.retries.attempts:
                message = (
                    f"Could not reach {self.platform} after "
                    f"{failures} tries: {problem}"
                )
                raise PlatformError(
                    message,
                    platform=self.platform,
                ) from problem
            await _wait(self.retries.wait_after(failures))
            continue

        seen = rate_limit_from_headers(response.headers)
        if seen is not None:
            self._rate_limit = seen

        failures += 1
        if failures < self.retries.attempts and _worth_another_try(response):
            await _wait(
                self.retries.wait_after(failures, retry_after_seconds(response))
            )
            continue

        if response.is_error:
            raise self._error_for(response)
        return response

get `async` ¶

get(path: str, **kwargs: object) -> Response

Send a GET request.

Parameters:

Name	Type	Description	Default
`path`	`str`	Joined onto the base url.	required
`**kwargs`	`object`	Anything `request` takes.	`{}`

Returns:

Type	Description
`Response`	The reply.

Source code in src/socialchimp/http.py

async def get(self, path: str, **kwargs: object) -> httpx.Response:
    """Send a GET request.

    Args:
        path: Joined onto the base url.
        **kwargs: Anything `request` takes.

    Returns:
        The reply.
    """
    return await self.request("GET", path, **kwargs)

post `async` ¶

post(path: str, **kwargs: object) -> Response

Send a POST request.

Parameters:

Name	Type	Description	Default
`path`	`str`	Joined onto the base url.	required
`**kwargs`	`object`	Anything `request` takes.	`{}`

Returns:

Type	Description
`Response`	The reply.

Source code in src/socialchimp/http.py

async def post(self, path: str, **kwargs: object) -> httpx.Response:
    """Send a POST request.

    Args:
        path: Joined onto the base url.
        **kwargs: Anything `request` takes.

    Returns:
        The reply.
    """
    return await self.request("POST", path, **kwargs)

put `async` ¶

put(path: str, **kwargs: object) -> Response

Send a PUT request.

Parameters:

Name	Type	Description	Default
`path`	`str`	Joined onto the base url.	required
`**kwargs`	`object`	Anything `request` takes.	`{}`

Returns:

Type	Description
`Response`	The reply.

Source code in src/socialchimp/http.py

async def put(self, path: str, **kwargs: object) -> httpx.Response:
    """Send a PUT request.

    Args:
        path: Joined onto the base url.
        **kwargs: Anything `request` takes.

    Returns:
        The reply.
    """
    return await self.request("PUT", path, **kwargs)

delete `async` ¶

delete(path: str, **kwargs: object) -> Response

Send a DELETE request.

Parameters:

Name	Type	Description	Default
`path`	`str`	Joined onto the base url.	required
`**kwargs`	`object`	Anything `request` takes.	`{}`

Returns:

Type	Description
`Response`	The reply.

Source code in src/socialchimp/http.py

async def delete(self, path: str, **kwargs: object) -> httpx.Response:
    """Send a DELETE request.

    Args:
        path: Joined onto the base url.
        **kwargs: Anything `request` takes.

    Returns:
        The reply.
    """
    return await self.request("DELETE", path, **kwargs)

json `async` ¶

json(method: str, path: str, **kwargs: object) -> RawData

Send a request and read the reply as a JSON object.

Parameters:

Name	Type	Description	Default
`method`	`str`	`"GET"`, `"POST"` and so on.	required
`path`	`str`	Joined onto the base url.	required
`**kwargs`	`object`	Anything `request` takes.	`{}`

Returns:

Type	Description
`RawData`	The reply, parsed.

Raises:

Type	Description
`PlatformError`	If the reply was not JSON, or was JSON but not an object. The reply itself is kept on `raw`.
`SocialChimpError`	If the network refused the request.

Source code in src/socialchimp/http.py

async def json(self, method: str, path: str, **kwargs: object) -> RawData:
    """Send a request and read the reply as a JSON object.

    Args:
        method: `"GET"`, `"POST"` and so on.
        path: Joined onto the base url.
        **kwargs: Anything `request` takes.

    Returns:
        The reply, parsed.

    Raises:
        PlatformError: If the reply was not JSON, or was JSON but not an
            object. The reply itself is kept on `raw`.
        SocialChimpError: If the network refused the request.
    """
    response = await self.request(method, path, **kwargs)
    try:
        parsed = response.json()
    except ValueError as problem:
        message = (
            f"{self.platform} was expected to answer with JSON but sent "
            f"something else. It starts: {response.text[:200]!r}"
        )
        raise PlatformError(
            message,
            platform=self.platform,
            status_code=response.status_code,
            raw={"body": response.text},
        ) from problem

    if not isinstance(parsed, dict):
        message = (
            f"{self.platform} answered with JSON, but with a "
            f"{type(parsed).__name__} where an object was expected."
        )
        raise PlatformError(
            message,
            platform=self.platform,
            status_code=response.status_code,
            raw={"body": parsed},
        )
    return parsed

aclose `async` ¶

aclose() -> None

Close the connections this client is holding open.

Source code in src/socialchimp/http.py

async def aclose(self) -> None:
    """Close the connections this client is holding open."""
    await self._client.aclose()

Retries `dataclass` ¶

Retries(
    attempts: int = 4,
    first_wait: float = 0.5,
    biggest_wait: float = 30.0,
    spread: float = 0.5,
)

How many times to try again, and how long to wait in between.

Waits double after each failure, so a network that is struggling is not hammered. Part of each wait is random, so that every client which failed at the same moment does not come back at the same moment.

Attributes:

Name	Type	Description
`attempts`	`int`	How many tries in total, counting the first one.
`first_wait`	`float`	Seconds to wait after the first failure.
`biggest_wait`	`float`	The longest we will ever wait between two tries.
`spread`	`float`	How much of a wait is random, from 0 (never) to 1 (up to double the wait).

wait_after ¶

wait_after(
    failures: int, asked_for: float | None = None
) -> float

Work out how long to wait before the next try.

Parameters:

Name	Type	Description	Default
`failures`	`int`	How many tries have failed so far.	required
`asked_for`	`float \| None`	Seconds the network itself asked us to wait, when it said. This is the least we will wait; if the wait we worked out is longer, we take the longer one.	`None`

Returns:

Type	Description
`float`	Seconds to wait.

Source code in src/socialchimp/http.py

def wait_after(self, failures: int, asked_for: float | None = None) -> float:
    """Work out how long to wait before the next try.

    Args:
        failures: How many tries have failed so far.
        asked_for: Seconds the network itself asked us to wait, when
            it said. This is the least we will wait; if the wait we
            worked out is longer, we take the longer one.

    Returns:
        Seconds to wait.
    """
    wait = min(self.first_wait * 2.0 ** (failures - 1), self.biggest_wait)
    wait += wait * self.spread * _random_fraction()
    if asked_for is not None:
        return max(wait, asked_for)
    return wait

RateLimit `dataclass` ¶

RateLimit(
    limit: int | None = None,
    remaining: int | None = None,
    resets_at: datetime | None = None,
)

How much of a network's allowance is left, as it last told us.

Every field may be None, which means "the network did not say" - never "zero". Read this before sending a burst of requests and you can slow down before being told to.

Attributes:

Name	Type	Description
`limit`	`int \| None`	How many requests are allowed in the current stretch of time.
`remaining`	`int \| None`	How many of those are left.
`resets_at`	`datetime \| None`	When the count starts again.

is_used_up `property` ¶

is_used_up: bool

Whether there are no requests left before the count starts again.

rate_limit_from_headers ¶

rate_limit_from_headers(
    headers: Headers, *, now: datetime | None = None
) -> RateLimit | None

Read a reply's rate-limit headers.

Both spellings are read: x-ratelimit-limit, which nearly everybody uses, and X's x-rate-limit-limit. So is Pinterest's habit of listing every window in one header - "100, 100;w=1, 1000;w=60" - where the bare number in front is the one that applies right now.

Parameters:

Name	Type	Description	Default
`headers`	`Headers`	The reply's headers.	required
`now`	`datetime \| None`	What to treat as the current moment, for networks that count down in seconds. Only useful in tests.	`None`

Returns:

Type	Description
`RateLimit \| None`	What the network said, or `None` if it said nothing we recognise.

Source code in src/socialchimp/http.py

def rate_limit_from_headers(
    headers: httpx.Headers,
    *,
    now: datetime | None = None,
) -> RateLimit | None:
    """Read a reply's rate-limit headers.

    Both spellings are read: `x-ratelimit-limit`, which nearly everybody
    uses, and X's `x-rate-limit-limit`. So is Pinterest's habit of listing
    every window in one header - `"100, 100;w=1, 1000;w=60"` - where the
    bare number in front is the one that applies right now.

    Args:
        headers: The reply's headers.
        now: What to treat as the current moment, for networks that count
            down in seconds. Only useful in tests.

    Returns:
        What the network said, or `None` if it said nothing we recognise.
    """
    limit = _whole_number(_first_header(headers, _LIMIT_HEADERS))
    remaining = _whole_number(_first_header(headers, _REMAINING_HEADERS))
    resets_at = _reset_time(_first_header(headers, _RESET_HEADERS), now)

    if limit is None and remaining is None and resets_at is None:
        return None
    return RateLimit(limit=limit, remaining=remaining, resets_at=resets_at)

retry_after_seconds ¶

retry_after_seconds(
    response: Response, *, now: datetime | None = None
) -> float | None

Read how long a network has asked us to wait, in seconds.

Networks write Retry-After two ways: a number of seconds ("30") or a date ("Wed, 21 Oct 2026 07:28:00 GMT"). Both come back here as seconds from now. A date that has already gone by, or a negative number, comes back as zero rather than as a wait that runs backwards.

Where a reply carries no Retry-After at all, a rate-limit reset header is read instead: Mastodon's X-RateLimit-Reset, written as an ISO-8601 timestamp, or Bluesky's RateLimit-Reset, written as a unix time in seconds. Both come back the same way - seconds from now, never negative.

Parameters:

Name	Type	Description	Default
`response`	`Response`	The reply to read.	required
`now`	`datetime \| None`	What to treat as the current moment. Only useful in tests.	`None`

Returns:

Type	Description
`float \| None`	Seconds to wait, or `None` when the network did not say or wrote
`float \| None`	something we cannot read.

Source code in src/socialchimp/http.py

def retry_after_seconds(
    response: httpx.Response,
    *,
    now: datetime | None = None,
) -> float | None:
    """Read how long a network has asked us to wait, in seconds.

    Networks write `Retry-After` two ways: a number of seconds (`"30"`) or a
    date (`"Wed, 21 Oct 2026 07:28:00 GMT"`). Both come back here as seconds
    from now. A date that has already gone by, or a negative number, comes
    back as zero rather than as a wait that runs backwards.

    Where a reply carries no `Retry-After` at all, a rate-limit reset header
    is read instead: Mastodon's `X-RateLimit-Reset`, written as an ISO-8601
    timestamp, or Bluesky's `RateLimit-Reset`, written as a unix time in
    seconds. Both come back the same way - seconds from now, never negative.

    Args:
        response: The reply to read.
        now: What to treat as the current moment. Only useful in tests.

    Returns:
        Seconds to wait, or `None` when the network did not say or wrote
        something we cannot read.
    """
    header = response.headers.get("retry-after")
    if header is None:
        return _retry_after_from_reset_header(response.headers, now=now)

    text = header.strip()
    try:
        seconds = float(text)
    except ValueError:
        pass
    else:
        # "inf" reads as a float and would park us forever, so it is turned
        # away with everything else we cannot use.
        if not math.isfinite(seconds):
            return None
        return max(seconds, 0.0)

    try:
        when = parsedate_to_datetime(text)
    except ValueError:
        return None

    if when.tzinfo is None:
        when = when.replace(tzinfo=UTC)
    moment = now if now is not None else datetime.now(UTC)
    return max((when - moment).total_seconds(), 0.0)

error_from_response ¶

error_from_response(
    response: Response, *, platform: str = _UNNAMED
) -> SocialChimpError

Turn an unhappy reply into the socialchimp error that describes it.

This is the shared mapping every network starts from. A platform that wants to name its own quirks writes its own function, handles the replies it recognises, and calls this one for the rest:

def bluesky_errors(response: httpx.Response) -> SocialChimpError:
    body = read_body(response)
    if body.get("error") == "TextTooLong":
        return InvalidPostError("This post is too long for bluesky.")
    return error_from_response(response, platform="bluesky")

Parameters:

Name	Type	Description	Default
`response`	`Response`	The reply to turn into an error.	required
`platform`	`str`	Which network sent it, used in the message.	`_UNNAMED`

Returns:

Type	Description
`SocialChimpError`	The error to raise. Always an error, never `None`, so a caller
`SocialChimpError`	cannot forget a case.

Source code in src/socialchimp/http.py

def error_from_response(
    response: httpx.Response,
    *,
    platform: str = _UNNAMED,
) -> SocialChimpError:
    """Turn an unhappy reply into the socialchimp error that describes it.

    This is the shared mapping every network starts from. A platform that
    wants to name its own quirks writes its own function, handles the replies
    it recognises, and calls this one for the rest:

        def bluesky_errors(response: httpx.Response) -> SocialChimpError:
            body = read_body(response)
            if body.get("error") == "TextTooLong":
                return InvalidPostError("This post is too long for bluesky.")
            return error_from_response(response, platform="bluesky")

    Args:
        response: The reply to turn into an error.
        platform: Which network sent it, used in the message.

    Returns:
        The error to raise. Always an error, never `None`, so a caller
        cannot forget a case.
    """
    body = read_body(response)
    said = _what_it_said(body)
    status = response.status_code

    if status == httpx.codes.UNAUTHORIZED:
        return AuthError(
            f"{platform} would not accept our sign-in (401). The person may "
            f"need to connect their account again.{said}"
        )
    if status == httpx.codes.FORBIDDEN:
        return NotAllowedError(
            f"{platform} will not let this account do that (403). It is "
            f"usually a permission that was never asked for.{said}"
        )
    if status == httpx.codes.NOT_FOUND:
        return NotFoundError(
            f"{platform} has no such post, account or page (404).{said}"
        )
    if status == httpx.codes.TOO_MANY_REQUESTS:
        return RateLimitError(
            f"{platform} is asking us to slow down (429).{said}",
            retry_after=retry_after_seconds(response),
        )

    if status >= httpx.codes.INTERNAL_SERVER_ERROR:
        message = (
            f"{platform} had trouble of its own ({status}). Trying again in a "
            f"little while usually works.{said}"
        )
    elif status >= httpx.codes.BAD_REQUEST:
        message = f"{platform} refused this request ({status}).{said}"
    else:
        # Nothing should map a good reply, but a function that can return
        # nothing is a trap for whoever does it by accident.
        message = f"{platform} sent a reply we did not expect ({status}).{said}"

    return PlatformError(
        message,
        platform=platform,
        status_code=status,
        raw=body,
    )

read_body ¶

read_body(response: Response) -> RawData

Return a reply's body as a dictionary, whatever shape it arrived in.

Most networks answer with a JSON object and that is handed straight back. Anything else - a JSON list, a plain string, an HTML error page - is put under a body key, so callers never have to guess what they were given.

Parameters:

Name	Type	Description	Default
`response`	`Response`	The reply to read.	required

Returns:

Type	Description
`RawData`	The body as a dictionary.

Source code in src/socialchimp/http.py

def read_body(response: httpx.Response) -> RawData:
    """Return a reply's body as a dictionary, whatever shape it arrived in.

    Most networks answer with a JSON object and that is handed straight back.
    Anything else - a JSON list, a plain string, an HTML error page - is put
    under a `body` key, so callers never have to guess what they were given.

    Args:
        response: The reply to read.

    Returns:
        The body as a dictionary.
    """
    try:
        parsed = response.json()
    except ValueError:
        return {"body": response.text}

    if isinstance(parsed, dict):
        return parsed
    return {"body": parsed}

Writing a platform¶

The protocol¶

Platform ¶

api_base ¶

auth_headers ¶

limits async ¶

start_login async ¶

finish_login async ¶

refresh async ¶

publish async ¶

What a platform can opt into¶

CanCreateApp ¶

create_app async ¶

CanResumeLogin ¶

resume_login async ¶

CanDeletePosts ¶

delete_post async ¶

CanReadStats ¶

read_stats async ¶

CanReadUpdates ¶

fetch_updates async ¶

CanCheckSignature ¶

check_signature ¶

read_update ¶

CanCheckState ¶

check_state async ¶

CanAnswerSetupCheck ¶

answer_setup_check ¶

CanReadPushedUpdates ¶

read_updates ¶

The social inbox¶

CanReadPost ¶

read_post async ¶

CanReadThread ¶

read_thread async ¶

CanReply ¶

reply async ¶

CanLike ¶

like async ¶

unlike async ¶

CanReadLikes ¶

read_likes async ¶

CanReadUpdatesAfter ¶

fetch_updates_after async ¶

mark_seen async ¶

CanMessage ¶

read_conversations async ¶

read_messages async ¶

send_message async ¶

mark_read async ¶

CanStartConversations ¶

start_conversation async ¶

Signing someone in¶

LoginRequest dataclass ¶

SendToNetwork dataclass ¶

LoginField dataclass ¶

AskForDetails dataclass ¶

AccountChoice dataclass ¶

ChooseAccount dataclass ¶

Finished dataclass ¶

Finding installed platforms¶

register_platform ¶

unregister_platform ¶

available_platforms ¶

get_platform_class ¶

clear_platform_cache ¶

Making requests¶

HttpClient ¶

rate_limit property ¶

is_closed property ¶

request async ¶

get async ¶

post async ¶

put async ¶

delete async ¶

json async ¶

aclose async ¶

Retries dataclass ¶

wait_after ¶

RateLimit dataclass ¶

limits `async` ¶

start_login `async` ¶

finish_login `async` ¶

refresh `async` ¶

publish `async` ¶

create_app `async` ¶

resume_login `async` ¶

delete_post `async` ¶

read_stats `async` ¶

fetch_updates `async` ¶

check_state `async` ¶

read_post `async` ¶

read_thread `async` ¶

reply `async` ¶

like `async` ¶

unlike `async` ¶

read_likes `async` ¶

fetch_updates_after `async` ¶

mark_seen `async` ¶

read_conversations `async` ¶

read_messages `async` ¶

send_message `async` ¶

mark_read `async` ¶

start_conversation `async` ¶

LoginRequest `dataclass` ¶

SendToNetwork `dataclass` ¶

LoginField `dataclass` ¶

AskForDetails `dataclass` ¶

AccountChoice `dataclass` ¶

ChooseAccount `dataclass` ¶

Finished `dataclass` ¶

rate_limit `property` ¶

is_closed `property` ¶

request `async` ¶

get `async` ¶

post `async` ¶

put `async` ¶

delete `async` ¶

json `async` ¶

aclose `async` ¶

Retries `dataclass` ¶

RateLimit `dataclass` ¶

is_used_up `property` ¶