Client

Start here. SocialChimp is the class your app creates once. Account is what you get back for a single connection, and it is what you call .post() and .direct on.

SocialChimp

SocialChimp

SocialChimp(
    storage: Storage,
    *,
    platforms: Mapping[str, Platform] | None = None,
    token_manager: TokenManager | None = None,
    make_lock: MakeLock | None = None,
    http: HttpClient | None = None,
)

The way in. One of these is enough for a whole app.

Give it somewhere to keep connections and it does the rest: finds the platform for each network, keeps tokens working, checks posts before sending them, and closes what it opened.

sc = SocialChimp(storage=MyStorage())
step = await sc.start_login("mastodon", host="mastodon.social",
                            redirect_uri="https://example.com/cb")

Keep one for the life of your process. The locks that stop two workers renewing the same token at once live on it, so a new one per request protects nothing.

Attributes:

Name	Type	Description
storage		Where connections and app credentials are kept.

Set up one app's use of socialchimp.

Parameters:

Name	Type	Description	Default
storage	Storage	Where connections and app credentials are kept. The one thing you have to provide.	required
platforms	Mapping[str, Platform] | None	Ready-made platforms, by name. Anything not named here is found among the installed platforms and created with no arguments, so this is where a platform that needs settings of its own goes - and where a test puts a fake.	None
token_manager	TokenManager | None	Renews tokens. Left out, one is made for each network, which is what you want almost always. Pass your own only if you need to change how renewal works entirely - and note that yours has to look up app credentials itself, which make_lock saves you from.	None
make_lock	MakeLock | None	Makes the lock held while a token is renewed. Pass one that every process shares - built on Redis, say - if you run more than one web or queue worker. The default only holds inside one process, so without this two workers can renew the same connection at once, and on the networks that replace the refresh token each time that disconnects the account.	None
http	HttpClient | None	Sends requests for Account.direct. Left out, one client is made for each network, server and event loop, and closed by aclose. One you pass in is used everywhere and is yours to close - so leave this out if your app runs each call on a new event loop, as Django does under WSGI, and let socialchimp keep one per loop for you.	None

Source code in src/socialchimp/client.py

def __init__(
    self,
    storage: Storage,
    *,
    platforms: Mapping[str, Platform] | None = None,
    token_manager: TokenManager | None = None,
    make_lock: MakeLock | None = None,
    http: HttpClient | None = None,
) -> None:
    """Set up one app's use of socialchimp.

    Args:
        storage: Where connections and app credentials are kept. The one
            thing you have to provide.
        platforms: Ready-made platforms, by name. Anything not named here
            is found among the installed platforms and created with no
            arguments, so this is where a platform that needs settings of
            its own goes - and where a test puts a fake.
        token_manager: Renews tokens. Left out, one is made for each
            network, which is what you want almost always. Pass your own
            only if you need to change how renewal works entirely - and
            note that yours has to look up app credentials itself, which
            `make_lock` saves you from.
        make_lock: Makes the lock held while a token is renewed. Pass one
            that every process shares - built on Redis, say - if you run
            more than one web or queue worker. The default only holds
            inside one process, so without this two workers can renew the
            same connection at once, and on the networks that replace the
            refresh token each time that disconnects the account.
        http: Sends requests for `Account.direct`. Left out, one client
            is made for each network, server and event loop, and closed
            by `aclose`. One you pass in is used everywhere and is yours
            to close - so leave this out if your app runs each call on a
            new event loop, as Django does under WSGI, and let
            socialchimp keep one per loop for you.
    """
    self.storage = storage
    self._platforms: dict[str, Platform] = dict(platforms or {})
    self._one_token_manager = token_manager
    self._make_lock = make_lock
    self._token_managers: dict[str, TokenManager] = {}
    self._http = http
    self._http_made: dict[_ClientKey, HttpClient] = {}

platform_for

platform_for(name: str) -> Platform

Return the platform for one network, making it if need be.

Parameters:

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

Returns:

Type	Description
Platform	The platform. The same one every time, so anything it remembers
Platform	is kept.

Raises:

Type	Description
ConfigError	If nothing is installed or registered under that name. The message lists what is, and how to install the network you asked for when it is one socialchimp covers.

Source code in src/socialchimp/client.py

def platform_for(self, name: str) -> Platform:
    """Return the platform for one network, making it if need be.

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

    Returns:
        The platform. The same one every time, so anything it remembers
        is kept.

    Raises:
        ConfigError: If nothing is installed or registered under that
            name. The message lists what is, and how to install the
            network you asked for when it is one socialchimp covers.
    """
    ready = self._platforms.get(name)
    if ready is None:
        # Nothing is imported until a platform is actually asked for, so
        # an app with ten installed pays for the one it uses.
        ready = get_platform_class(name)()
        self._platforms[name] = ready
    return ready

features

features(platform: str) -> Feature

Ask what a network can do, by name, with no connection needed.

Useful for deciding which button to show before anyone has connected an account. Once an account exists, Account.features is usually the better call: this looks the platform up by name alone, so it cannot tell you anything about that particular account.

Parameters:

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

Returns:

Type	Description
Feature	The features that network supports.

Raises:

Type	Description
ConfigError	If nothing is installed or registered under that name.

Source code in src/socialchimp/client.py

def features(self, platform: str) -> Feature:
    """Ask what a network can do, by name, with no connection needed.

    Useful for deciding which button to show before anyone has connected
    an account. Once an account exists, `Account.features` is usually
    the better call: this looks the platform up by name alone, so it
    cannot tell you anything about that particular account.

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

    Returns:
        The features that network supports.

    Raises:
        ConfigError: If nothing is installed or registered under that
            name.
    """
    return self.platform_for(platform).features

tokens_for

tokens_for(name: str) -> TokenManager

Return the token manager for one network.

Parameters:

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

Returns:

Type	Description
TokenManager	The manager, made on first use unless you passed one in. The
TokenManager	same one every time, because the locks that stop two renewals
TokenManager	colliding live on it.

Source code in src/socialchimp/client.py

def tokens_for(self, name: str) -> TokenManager:
    """Return the token manager for one network.

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

    Returns:
        The manager, made on first use unless you passed one in. The
        same one every time, because the locks that stop two renewals
        colliding live on it.
    """
    if self._one_token_manager is not None:
        return self._one_token_manager

    manager = self._token_managers.get(name)
    if manager is None:
        # make_lock is only passed on when given, so the default
        # stays whatever TokenManager decides rather than being
        # duplicated here and drifting from it.
        manager = (
            TokenManager(self.storage, self._renewal_for(name))
            if self._make_lock is None
            else TokenManager(
                self.storage,
                self._renewal_for(name),
                make_lock=self._make_lock,
            )
        )
        self._token_managers[name] = manager
    return manager

http_for

http_for(connection: Connection) -> HttpClient

Return the HTTP client for one connection's network and address.

Parameters:

Name	Type	Description	Default
connection	Connection	The account whose network we are talking to.	required

Returns:

Type	Description
HttpClient	The client, made on first use unless you passed one in. One per
HttpClient	network, address and event loop, so accounts on the same server
HttpClient	share one.

Raises:

Type	Description
ConfigError	If you call this from outside async code, where there is no loop for a client to belong to.

Source code in src/socialchimp/client.py

def http_for(self, connection: Connection) -> HttpClient:
    """Return the HTTP client for one connection's network and address.

    Args:
        connection: The account whose network we are talking to.

    Returns:
        The client, made on first use unless you passed one in. One per
        network, address and event loop, so accounts on the same server
        share one.

    Raises:
        ConfigError: If you call this from outside async code, where
            there is no loop for a client to belong to.
    """
    if self._http is not None:
        return self._http

    # The platform says where its API is: the account's own server for
    # Mastodon, one address for everybody on Facebook. So the address is
    # what tells two clients apart, rather than anything on the
    # connection.
    address = self.platform_for(connection.platform).api_base(connection)
    # And the loop tells them apart as well, because a client holds open
    # sockets that belong to the loop it was made on. Django under WSGI
    # runs each request on a loop of its own and closes it at the end, so
    # without this the second request reuses a client whose loop, and
    # whose sockets, have gone. One long-lived loop - FastAPI, a script -
    # asks for the same key every time and keeps its pooling.
    key = _ClientKey(self._this_loop(), connection.platform, address)
    self._forget_finished_loops()
    made = self._http_made.get(key)
    if made is None:
        made = HttpClient(address, platform=connection.platform)
        self._http_made[key] = made
    return made

fresh_connection async

fresh_connection(connection_id: str) -> Connection

Read one connection, with a token that works right now.

Every call that acts as an account goes through here first, so a token is always renewed before it is used.

Parameters:

Name	Type	Description	Default
connection_id	str	The id your app gave this connection.	required

Returns:

Type	Description
Connection	The connection, renewed first if its token was running out.

Raises:

Type	Description
ConfigError	If nothing is stored under that id.
TokenExpiredError	If the token needed renewing and could not be.

Source code in src/socialchimp/client.py

async def fresh_connection(self, connection_id: str) -> Connection:
    """Read one connection, with a token that works right now.

    Every call that acts as an account goes through here first, so a
    token is always renewed before it is used.

    Args:
        connection_id: The id your app gave this connection.

    Returns:
        The connection, renewed first if its token was running out.

    Raises:
        ConfigError: If nothing is stored under that id.
        TokenExpiredError: If the token needed renewing and could not be.
    """
    # Read once to learn which network this is, because tokens are
    # renewed by the platform that issued them. The read after it is the
    # one that renews, and it is the answer we hand back.
    known = await self.storage.get_connection(connection_id)
    if known is None:
        message = (
            f"No connection is stored with the id {connection_id!r}. "
            f"Check the id, or connect the account again."
        )
        raise ConfigError(message)
    return await self.tokens_for(known.platform).valid_token(connection_id)

account

account(connection_id: str) -> Account

Return a handle for one connected account.

Cheap to make and reads nothing, so a handle for a connection that has not been saved yet is fine to hold. The connection is looked up when you actually do something with it.

Parameters:

Name	Type	Description	Default
connection_id	str	The id your app gave this connection.	required

Returns:

Type	Description
Account	The handle.

Source code in src/socialchimp/client.py

def account(self, connection_id: str) -> Account:
    """Return a handle for one connected account.

    Cheap to make and reads nothing, so a handle for a connection that
    has not been saved yet is fine to hold. The connection is looked up
    when you actually do something with it.

    Args:
        connection_id: The id your app gave this connection.

    Returns:
        The handle.
    """
    return Account(self, connection_id)

create_app async

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

Register your app with a network, and save what it gives back.

Only Mastodon can do this, and it has to be done once per server. Everywhere else you register the app by hand in a developer portal, and several networks review it before it works at all - so asking here says exactly that instead of failing later.

Parameters:

Name	Type	Description	Default
platform	str	Which network, for example "mastodon".	required
name	str	The app name people see when they approve 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	The credentials, already saved through your storage.

Raises:

Type	Description
NotSupportedError	If this network cannot register an app for you. The message says where to register it by hand.
ConfigError	If the platform says it can but has no method for it.

Source code in src/socialchimp/client.py

async def create_app(
    self,
    platform: str,
    *,
    name: str,
    redirect_uri: str,
    host: str | None = None,
    scopes: tuple[str, ...] = (),
) -> AppCredentials:
    """Register your app with a network, and save what it gives back.

    Only Mastodon can do this, and it has to be done once per server.
    Everywhere else you register the app by hand in a developer portal,
    and several networks review it before it works at all - so asking
    here says exactly that instead of failing later.

    Args:
        platform: Which network, for example `"mastodon"`.
        name: The app name people see when they approve 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:
        The credentials, already saved through your storage.

    Raises:
        NotSupportedError: If this network cannot register an app for
            you. The message says where to register it by hand.
        ConfigError: If the platform says it can but has no method for
            it.
    """
    maker = self.platform_for(platform)
    _refuse(
        maker,
        Feature.CREATE_APP,
        _NOTHING_TO_REGISTER
        if Feature.NEEDS_NO_APP in maker.features
        else _REGISTER_BY_HAND,
    )
    if not isinstance(maker, CanCreateApp):
        raise _missing_method(maker, "create_app")

    app = await maker.create_app(
        name=name,
        redirect_uri=redirect_uri,
        host=host,
        scopes=scopes,
    )
    # Saved for you, because registering again on the same server wastes
    # a record on that server and hands you a different id and secret.
    await self.storage.save_app(app)
    return app

start_login async

start_login(
    platform: str,
    *,
    redirect_uri: str,
    scopes: tuple[str, ...] = (),
    host: str | None = None,
    state: str | None = None,
) -> LoginStep

Begin signing someone in to a network.

Parameters:

Name	Type	Description	Default
platform	str	Which network, for example "mastodon".	required
redirect_uri	str	Where the network sends the person back to. It has to match what the network's developer portal has on file.	required
scopes	tuple[str, ...]	Permissions to ask for. Empty uses the platform's sensible defaults.	()
host	str | None	Which server, for networks that have more than one.	None
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.	None

Returns:

Type	Description
LoginStep	What to do next, handed back exactly as the platform gave it.
LoginStep	Usually SendToNetwork: redirect the person's browser to
LoginStep	step.url, and keep step.remember with that person's session,
LoginStep	because finish_login needs it back and only you can carry it
LoginStep	there.
LoginStep	Networks signed in to with an app password or a bot token answer
LoginStep	with AskForDetails instead, because there is nowhere to send
LoginStep	anybody. Show a box for each of step.fields, hide the ones
LoginStep	marked secret, and pass what the person typed to finish_login
LoginStep	as callback, under the names the fields gave.

Raises:

Type	Description
ConfigError	If your app is not registered with this network yet, on a network that needs one registered. Bluesky has no app to register, so nothing has to be saved first.

Source code in src/socialchimp/client.py

async def start_login(
    self,
    platform: str,
    *,
    redirect_uri: str,
    scopes: tuple[str, ...] = (),
    host: str | None = None,
    state: str | None = None,
) -> LoginStep:
    """Begin signing someone in to a network.

    Args:
        platform: Which network, for example `"mastodon"`.
        redirect_uri: Where the network sends the person back to. It has
            to match what the network's developer portal has on file.
        scopes: Permissions to ask for. Empty uses the platform's
            sensible defaults.
        host: Which server, for networks that have more than one.
        state: 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.

    Returns:
        What to do next, handed back exactly as the platform gave it.

        Usually `SendToNetwork`: redirect the person's browser to
        `step.url`, and keep `step.remember` with that person's session,
        because `finish_login` needs it back and only you can carry it
        there.

        Networks signed in to with an app password or a bot token answer
        with `AskForDetails` instead, because there is nowhere to send
        anybody. Show a box for each of `step.fields`, hide the ones
        marked `secret`, and pass what the person typed to `finish_login`
        as `callback`, under the names the fields gave.

    Raises:
        ConfigError: If your app is not registered with this network
            yet, on a network that needs one registered. Bluesky has
            no app to register, so nothing has to be saved first.
    """
    # The platform is found first, so a name nobody has is answered with
    # the registry's message rather than one about app credentials.
    starter = self.platform_for(platform)
    request = await self._login_request(
        starter,
        redirect_uri=redirect_uri,
        scopes=scopes,
        host=host,
        state=state,
    )
    step = await starter.start_login(request)
    # Saved here too, not just in finish_login. A network that needs
    # nothing from the person could answer with Finished right away, and
    # a connection dropped on one path out of three is the kind of bug
    # that only shows up on the one network that does it.
    return await self._save_if_finished(step)

finish_login async

finish_login(
    platform: str,
    *,
    callback: Mapping[str, str],
    redirect_uri: str,
    scopes: tuple[str, ...] = (),
    host: str | None = None,
    state: str | None = None,
    remember: RawData | None = None,
) -> LoginStep

Carry on after the person comes back from the network.

Parameters:

Name	Type	Description	Default
platform	str	Which network, for example "mastodon".	required
callback	Mapping[str, str]	The query values the network sent back, such as Django's request.GET or FastAPI's request.query_params. For a network that asked for details instead of sending the person anywhere, this is what they typed into your form, under the names AskForDetails gave.	required
redirect_uri	str	The same one the login was started with.	required
scopes	tuple[str, ...]	The same ones the login was started with.	()
host	str | None	The same server the login was started on.	None
state	str | None	The value you started with, if you chose one.	None
remember	RawData | None	What start_login handed you in SendToNetwork.remember. Keep it with that person's session and give it back here. socialchimp cannot keep it for you: the person can be sent away by one web worker and come back to another, so anything held in memory works on your laptop and fails in production.	None

Returns:

Type	Description
LoginStep	Finished when the account is connected, and the connection is
LoginStep	saved for you. ChooseAccount when the network needs to know
LoginStep	which page or channel to use - show the options, then call
LoginStep	choose.

Raises:

Type	Description
ConfigError	If your app is not registered with this network yet, on a network that needs one registered. Bluesky has no app to register, so nothing has to be saved first.

Source code in src/socialchimp/client.py

async def finish_login(
    self,
    platform: str,
    *,
    callback: Mapping[str, str],
    redirect_uri: str,
    scopes: tuple[str, ...] = (),
    host: str | None = None,
    state: str | None = None,
    remember: RawData | None = None,
) -> LoginStep:
    """Carry on after the person comes back from the network.

    Args:
        platform: Which network, for example `"mastodon"`.
        callback: The query values the network sent back, such as
            Django's `request.GET` or FastAPI's `request.query_params`.
            For a network that asked for details instead of sending the
            person anywhere, this is what they typed into your form,
            under the names `AskForDetails` gave.
        redirect_uri: The same one the login was started with.
        scopes: The same ones the login was started with.
        host: The same server the login was started on.
        state: The value you started with, if you chose one.
        remember: What `start_login` handed you in
            `SendToNetwork.remember`. Keep it with that person's session
            and give it back here. socialchimp cannot keep it for you:
            the person can be sent away by one web worker and come back
            to another, so anything held in memory works on your laptop
            and fails in production.

    Returns:
        `Finished` when the account is connected, and the connection is
        saved for you. `ChooseAccount` when the network needs to know
        which page or channel to use - show the options, then call
        `choose`.

    Raises:
        ConfigError: If your app is not registered with this network
            yet, on a network that needs one registered. Bluesky has
            no app to register, so nothing has to be saved first.
    """
    finisher = self.platform_for(platform)
    request = await self._login_request(
        finisher,
        redirect_uri=redirect_uri,
        scopes=scopes,
        host=host,
        state=state,
    )
    step = await finisher.finish_login(request, callback, remember)
    return await self._save_if_finished(step)

choose async

choose(
    platform: str,
    *,
    account_id: str,
    resume_token: str,
    redirect_uri: str,
    scopes: tuple[str, ...] = (),
    host: str | None = None,
    state: str | None = None,
    remember: RawData | None = None,
) -> LoginStep

Carry on a login after the person picked which account to use.

Parameters:

Name	Type	Description	Default
platform	str	Which network, for example "facebook".	required
account_id	str	The id of the option they picked, from ChooseAccount.options.	required
resume_token	str	The value from ChooseAccount, handed straight back.	required
redirect_uri	str	The same one the login was started with.	required
scopes	tuple[str, ...]	The same ones the login was started with.	()
host	str | None	The same server the login was started on.	None
state	str | None	The value you started with, if you chose one.	None
remember	RawData | None	The same value finish_login was given, still kept with that person's session.	None

Returns:

Type	Description
LoginStep	Finished when the account is connected, and the connection is
LoginStep	saved for you. A network that asks twice can answer with another
LoginStep	ChooseAccount.

Raises:

Type	Description
NotSupportedError	If this network never pauses to ask, so there is nothing to carry on from.
ConfigError	If your app is not registered with this network yet, on a network that needs one registered. Bluesky has no app to register, so nothing has to be saved first.

Source code in src/socialchimp/client.py

async def choose(
    self,
    platform: str,
    *,
    account_id: str,
    resume_token: str,
    redirect_uri: str,
    scopes: tuple[str, ...] = (),
    host: str | None = None,
    state: str | None = None,
    remember: RawData | None = None,
) -> LoginStep:
    """Carry on a login after the person picked which account to use.

    Args:
        platform: Which network, for example `"facebook"`.
        account_id: The id of the option they picked, from
            `ChooseAccount.options`.
        resume_token: The value from `ChooseAccount`, handed straight
            back.
        redirect_uri: The same one the login was started with.
        scopes: The same ones the login was started with.
        host: The same server the login was started on.
        state: The value you started with, if you chose one.
        remember: The same value `finish_login` was given, still kept
            with that person's session.

    Returns:
        `Finished` when the account is connected, and the connection is
        saved for you. A network that asks twice can answer with another
        `ChooseAccount`.

    Raises:
        NotSupportedError: If this network never pauses to ask, so there
            is nothing to carry on from.
        ConfigError: If your app is not registered with this network
            yet, on a network that needs one registered. Bluesky has
            no app to register, so nothing has to be saved first.
    """
    chooser = self.platform_for(platform)
    if not isinstance(chooser, CanResumeLogin):
        raise NotSupportedError(
            platform=chooser.name,
            what="choosing an account part way through a login",
            suggestion=(
                "It signs someone in in one step, so finish_login is the "
                "whole of it."
            ),
        )

    request = await self._login_request(
        chooser,
        redirect_uri=redirect_uri,
        scopes=scopes,
        host=host,
        state=state,
    )
    step = await chooser.resume_login(
        request,
        resume_token=resume_token,
        account_id=account_id,
        remember=remember,
    )
    return await self._save_if_finished(step)

answer_setup_check

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

Answer the one-off question a network asks before it pushes.

Meta does this on Facebook, Instagram and Threads: point it 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.

Nobody has connected an account by this point, so this takes the network's name rather than going through Account - the same as start_login does, and for the same reason. It is a plain function rather than async because nothing is sent anywhere, so a synchronous view can call it without a bridge.

Parameters:

Name	Type	Description	Default
platform	str	Which network, for example "facebook".	required
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 that network's own form. Not your app secret - that one is for check_signature.	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
NotSupportedError	If this network asks nothing before it starts sending, or never sends anything at all. The message says which, because what to do about them is different.
SignatureError	If this is not a setup check, or the token is wrong. Answer 403 and send nothing back.

Source code in src/socialchimp/client.py

def answer_setup_check(
    self,
    platform: str,
    params: Mapping[str, str],
    *,
    verify_token: str,
) -> str:
    """Answer the one-off question a network asks before it pushes.

    Meta does this on Facebook, Instagram and Threads: point it 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.

    Nobody has connected an account by this point, so this takes the
    network's name rather than going through `Account` - the same as
    `start_login` does, and for the same reason. It is a plain function
    rather than `async` because nothing is sent anywhere, so a
    synchronous view can call it without a bridge.

    Args:
        platform: Which network, for example `"facebook"`.
        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 that network's own form.
            Not your app secret - that one is for `check_signature`.

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

    Raises:
        NotSupportedError: If this network asks nothing before it starts
            sending, or never sends anything at all. The message says
            which, because what to do about them is different.
        SignatureError: If this is not a setup check, or the token is
            wrong. Answer 403 and send nothing back.
    """
    answerer = self.platform_for(platform)
    if not isinstance(answerer, CanAnswerSetupCheck):
        raise NotSupportedError(
            platform=answerer.name,
            what="a setup check before it will push anything",
            suggestion=_no_setup_check(answerer),
        )
    return answerer.answer_setup_check(params, verify_token=verify_token)

check_signature

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

Check a request a network pushed to us really came from it.

The body must be the raw bytes of the request, exactly as they arrived. A signature is over those exact bytes, so a framework that parsed the JSON and built it again has already broken it - the spacing and the key order will not match. Read the body, check it here, and let read_updates parse it afterwards. This is the single most common reason a correct signature appears to fail.

This takes the network's name because the request arrives before we know whose account it concerns; read_updates is what tells you that.

Parameters:

Name	Type	Description	Default
platform	str	Which network, for example "facebook".	required
body	bytes	The request body, untouched.	required
headers	Mapping[str, str]	The request headers. Case does not matter.	required
secret	str	The secret you share with that network. Meta calls this the app secret, and it is not the verify token.	required

Raises:

Type	Description
NotSupportedError	If this network never sends us anything.
SignatureError	If the request cannot be trusted. Answer 401 and do nothing else with it - and say nothing about which check failed, because that only helps whoever is guessing.

Source code in src/socialchimp/client.py

def check_signature(
    self,
    platform: str,
    body: bytes,
    headers: Mapping[str, str],
    *,
    secret: str,
) -> None:
    """Check a request a network pushed to us really came from it.

    The body must be the **raw bytes** of the request, exactly as they
    arrived. A signature is over those exact bytes, so a framework that
    parsed the JSON and built it again has already broken it - the
    spacing and the key order will not match. Read the body, check it
    here, and let `read_updates` parse it afterwards. This is the single
    most common reason a correct signature appears to fail.

    This takes the network's name because the request arrives before we
    know whose account it concerns; `read_updates` is what tells you
    that.

    Args:
        platform: Which network, for example `"facebook"`.
        body: The request body, untouched.
        headers: The request headers. Case does not matter.
        secret: The secret you share with that network. Meta calls this
            the app secret, and it is not the verify token.

    Raises:
        NotSupportedError: If this network never sends us anything.
        SignatureError: If the request cannot be trusted. Answer 401 and
            do nothing else with it - and say nothing about which check
            failed, because that only helps whoever is guessing.
    """
    pusher = self.platform_for(platform)
    if not isinstance(pusher, CanCheckSignature):
        raise NotSupportedError(
            platform=pusher.name,
            what="pushing requests to a URL of yours",
            suggestion=(
                "Ask it on a timer instead, with Account.fetch_updates "
                "and socialchimp.events.Poller."
            ),
        )
    pusher.check_signature(body, headers, secret=secret)

read_updates

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

Turn a checked request into every update it carries.

Call this after check_signature has passed, never before.

All of them, not the first: Meta batches changes into one message when it is busy, which is exactly when you least want to drop the rest. Each update carries its own change on raw, so a handler reads that straight rather than hunting through the message for the change it is about.

Parameters:

Name	Type	Description	Default
platform	str	Which network, for example "facebook".	required
body	bytes	The request body, untouched.	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. Each
list[Update]	update names the connection it concerns.

Raises:

Type	Description
NotSupportedError	If this network never sends us anything, or its platform file was written before read_updates existed.
PlatformError	If the body is not one of that network's messages.

Source code in src/socialchimp/client.py

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

    Call this after `check_signature` has passed, never before.

    All of them, not the first: Meta batches changes into one message
    when it is busy, which is exactly when you least want to drop the
    rest. Each update carries its own change on `raw`, so a handler
    reads that straight rather than hunting through the message for the
    change it is about.

    Args:
        platform: Which network, for example `"facebook"`.
        body: The request body, untouched.

    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. Each
        update names the connection it concerns.

    Raises:
        NotSupportedError: If this network never sends us anything, or
            its platform file was written before `read_updates` existed.
        PlatformError: If the body is not one of that network's
            messages.
    """
    reader = self.platform_for(platform)
    if not isinstance(reader, CanReadPushedUpdates):
        raise NotSupportedError(
            platform=reader.name,
            what="reading every update out of one pushed request",
            suggestion=(
                "Either it never sends us anything, or its platform file "
                "has no read_updates(body) method yet. A platform written "
                "against socialchimp 0.1 may only have read_update, which "
                "hands back the first change and drops the rest."
            ),
        )
    return reader.read_updates(body)

aclose async

aclose() -> None

Close the HTTP clients this made.

A client you passed in yourself is left alone - it is yours, and you may still be using it.

A client whose loop has finished is let go of rather than closed, for the same reason _forget_finished_loops gives: there is nobody left to ask to close its sockets, and trying would raise here and leave the rest of the clients open behind it.

Source code in src/socialchimp/client.py

async def aclose(self) -> None:
    """Close the HTTP clients this made.

    A client you passed in yourself is left alone - it is yours, and you
    may still be using it.

    A client whose loop has finished is let go of rather than closed,
    for the same reason `_forget_finished_loops` gives: there is nobody
    left to ask to close its sockets, and trying would raise here and
    leave the rest of the clients open behind it.
    """
    made = list(self._http_made.items())
    self._http_made.clear()
    for key, http in made:
        if not key.loop.is_closed():
            await http.aclose()

Account

Account

Account(client: SocialChimp, connection_id: str)

One connected account, and the things you can do as it.

Made by SocialChimp.account. Making one reads nothing: the connection is looked up when you actually do something, so a handle for an account that does not exist yet is fine to hold.

account = sc.account(connection_id)
result = await account.post(Post(text="hello"))

Every call here renews the token first, so a post never fails just because a token aged out while it sat in a queue.

Attributes:

Name	Type	Description
id		The id your app gave this connection.
direct		Your own requests to the same network as the same account.

Point a handle at one connection, without reading anything.

Parameters:

Name	Type	Description	Default
client	SocialChimp	The client this account belongs to.	required
connection_id	str	The id your app gave this connection.	required

Source code in src/socialchimp/client.py

def __init__(self, client: SocialChimp, connection_id: str) -> None:
    """Point a handle at one connection, without reading anything.

    Args:
        client: The client this account belongs to.
        connection_id: The id your app gave this connection.
    """
    self.id = connection_id
    self.direct = Direct(client, connection_id)
    self._client = client

connection async

connection() -> Connection

Read this connection, with a token that works right now.

Returns:

Type	Description
Connection	The connection, renewed first if its token was running out.

Raises:

Type	Description
ConfigError	If nothing is stored under this id.
TokenExpiredError	If the token needed renewing and could not be.

Source code in src/socialchimp/client.py

async def connection(self) -> Connection:
    """Read this connection, with a token that works right now.

    Returns:
        The connection, renewed first if its token was running out.

    Raises:
        ConfigError: If nothing is stored under this id.
        TokenExpiredError: If the token needed renewing and could not be.
    """
    return await self._client.fresh_connection(self.id)

profile async

profile() -> AccountProfile

Ask the network for this account's current name and picture.

Nothing is saved to storage here - this is for showing a fresh name and picture, or for renewing one that has gone stale, not for keeping a copy yourself.

Returns:

Type	Description
AccountProfile	The name and picture the network has on file right now.

Raises:

Type	Description
NotSupportedError	If this network cannot be asked for its own name and picture this way.

Source code in src/socialchimp/client.py

async def profile(self) -> AccountProfile:
    """Ask the network for this account's current name and picture.

    Nothing is saved to storage here - this is for showing a fresh name
    and picture, or for renewing one that has gone stale, not for
    keeping a copy yourself.

    Returns:
        The name and picture the network has on file right now.

    Raises:
        NotSupportedError: If this network cannot be asked for its own
            name and picture this way.
    """
    connection = await self.connection()
    platform = self._client.platform_for(connection.platform)
    if not isinstance(platform, CanReadProfile):
        raise NotSupportedError(
            platform=platform.name,
            what="reading the account's name and picture",
        )
    return await platform.read_profile(connection)

limits async

limits() -> Limits

Look up what this network is allowing this account right now.

Worth reading before a burst of posts: a Mastodon server's post length is set by whoever runs it, and Instagram counts down how many posts are left today.

Returns:

Type	Description
Limits	The current limits.

Source code in src/socialchimp/client.py

async def limits(self) -> Limits:
    """Look up what this network is allowing this account right now.

    Worth reading before a burst of posts: a Mastodon server's post
    length is set by whoever runs it, and Instagram counts down how many
    posts are left today.

    Returns:
        The current limits.
    """
    connection = await self.connection()
    return await self._client.platform_for(connection.platform).limits(connection)

post async

post(post: Post) -> PostResult

Publish a post as this account.

The post is checked against the network's features and limits first, so an over-long post or a schedule the network cannot keep fails before a request is spent on it.

Parameters:

Name	Type	Description	Default
post	Post	What to publish.	required

Returns:

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

Raises:

Type	Description
InvalidPostError	If the post breaks one of the network's limits.
NotSupportedError	If the post needs something the network cannot do, such as scheduling.

Source code in src/socialchimp/client.py

async def post(self, post: Post) -> PostResult:
    """Publish a post as this account.

    The post is checked against the network's features and limits first,
    so an over-long post or a schedule the network cannot keep fails
    before a request is spent on it.

    Args:
        post: What to publish.

    Returns:
        What the network said about the new post.

    Raises:
        InvalidPostError: If the post breaks one of the network's limits.
        NotSupportedError: If the post needs something the network cannot
            do, such as scheduling.
    """
    connection = await self.connection()
    return await _publish(
        self._client.platform_for(connection.platform), connection, post
    )

check_state async

check_state(post_id: str) -> PostResult

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

Some networks keep working after they accept an upload. YouTube encodes a video for minutes, sometimes hours; TikTok can put one in somebody's drafts instead of publishing it. Both answer post() before they are finished, so a result that came back PROCESSING is not the end of the story.

The token is renewed first, the same as every other call here, so this is safe to put on a timer.

Parameters:

Name	Type	Description	Default
post_id	str	The network's identifier for the post, which is what post() handed back.	required

Returns:

Type	Description
PostResult	Where the post has got to now, in the same shape post() gave.

Raises:

Type	Description
NotSupportedError	If this network finishes before it answers, so there is nothing to ask about.

Source code in src/socialchimp/client.py

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

    Some networks keep working after they accept an upload. YouTube
    encodes a video for minutes, sometimes hours; TikTok can put one in
    somebody's drafts instead of publishing it. Both answer `post()`
    before they are finished, so a result that came back `PROCESSING`
    is not the end of the story.

    The token is renewed first, the same as every other call here, so
    this is safe to put on a timer.

    Args:
        post_id: The network's identifier for the post, which is what
            `post()` handed back.

    Returns:
        Where the post has got to now, in the same shape `post()` gave.

    Raises:
        NotSupportedError: If this network finishes before it answers,
            so there is nothing to ask about.
    """
    connection = await self.connection()
    platform = self._client.platform_for(connection.platform)
    if not isinstance(platform, CanCheckState):
        raise NotSupportedError(
            platform=platform.name,
            what="being asked how a post is getting on",
            suggestion=(
                "It finishes while we wait, so what publish gave you is "
                "the final answer. YouTube and TikTok are the two that "
                "keep working afterwards."
            ),
        )
    return await platform.check_state(connection, post_id)

fetch_updates async

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

Ask the network what has happened on this account since a moment.

For networks that never tell us anything themselves. Hand this to socialchimp.events.Poller and it runs on a timer, works out what is new, and delivers the same Update objects a pushing network would have produced.

The token is renewed first, so a poller left running for weeks does not quietly stop.

Parameters:

Name	Type	Description	Default
since	datetime | None	Only return things newer than this. None on the first call, when there is no marker saved yet - the network answers with a recent page rather than the whole history.	None

Returns:

Type	Description
Sequence[Update]	What has happened, oldest first.

Raises:

Type	Description
NotSupportedError	If this network cannot be asked. Where it pushes instead, receive its requests with SocialChimp.check_signature and SocialChimp.read_updates.

Source code in src/socialchimp/client.py

async def fetch_updates(
    self,
    since: datetime | None = None,
) -> Sequence[Update]:
    """Ask the network what has happened on this account since a moment.

    For networks that never tell us anything themselves. Hand this to
    `socialchimp.events.Poller` and it runs on a timer, works out what
    is new, and delivers the same `Update` objects a pushing network
    would have produced.

    The token is renewed first, so a poller left running for weeks does
    not quietly stop.

    Args:
        since: Only return things newer than this. `None` on the first
            call, when there is no marker saved yet - the network
            answers with a recent page rather than the whole history.

    Returns:
        What has happened, oldest first.

    Raises:
        NotSupportedError: If this network cannot be asked. Where it
            pushes instead, receive its requests with
            `SocialChimp.check_signature` and `SocialChimp.read_updates`.
    """
    connection = await self.connection()
    platform = self._client.platform_for(connection.platform)
    if not isinstance(platform, CanReadUpdates):
        raise NotSupportedError(
            platform=platform.name,
            what="being asked what has happened since",
            suggestion=(
                "Where a network pushes instead, take its requests with "
                "SocialChimp.check_signature and SocialChimp.read_updates."
            ),
        )
    return await platform.fetch_updates(connection, since)

read_replies async

read_replies(
    post_id: str, *, whole_conversation: bool = False
) -> Sequence[Update]

Read the replies to one of this account's posts.

Different from fetch_updates, which asks the whole account what is new. This reads one post, and can read further back than a poll would bother to.

Parameters:

Name	Type	Description	Default
post_id	str	The network's identifier for the post, which is what post() handed back.	required
whole_conversation	bool	True to read every depth of the thread - a reply to a reply included - rather than only the ones sent straight to the post.	False

Returns:

Type	Description
Sequence[Update]	The replies, oldest first.

Raises:

Type	Description
NotSupportedError	If this network keeps nothing an app can read this way.

Source code in src/socialchimp/client.py

async def read_replies(
    self,
    post_id: str,
    *,
    whole_conversation: bool = False,
) -> Sequence[Update]:
    """Read the replies to one of this account's posts.

    Different from `fetch_updates`, which asks the whole account what is
    new. This reads one post, and can read further back than a poll
    would bother to.

    Args:
        post_id: The network's identifier for the post, which is what
            `post()` handed back.
        whole_conversation: `True` to read every depth of the thread - a
            reply to a reply included - rather than only the ones sent
            straight to the post.

    Returns:
        The replies, oldest first.

    Raises:
        NotSupportedError: If this network keeps nothing an app can read
            this way.
    """
    connection = await self.connection()
    platform = self._client.platform_for(connection.platform)
    if not isinstance(platform, CanReadReplies):
        raise NotSupportedError(
            platform=platform.name,
            what="reading the replies to one post",
            suggestion=(
                "It keeps nothing worth reading this way, or its "
                "platform file has no read_replies yet."
            ),
        )
    return await platform.read_replies(
        connection, post_id, whole_conversation=whole_conversation
    )

read_post async

read_post(post_id: str) -> PostDetails

Read one post back in full, not only what publishing it returned.

Parameters:

Name	Type	Description	Default
post_id	str	The network's identifier for the post or comment.	required

Returns:

Type	Description
PostDetails	The post, in full.

Raises:

Type	Description
NotSupportedError	If this network cannot be asked for a post this way.
ConfigError	If the platform says it can but has no method for it.

Source code in src/socialchimp/client.py

async def read_post(self, post_id: str) -> PostDetails:
    """Read one post back in full, not only what publishing it returned.

    Args:
        post_id: The network's identifier for the post or comment.

    Returns:
        The post, in full.

    Raises:
        NotSupportedError: If this network cannot be asked for a post
            this way.
        ConfigError: If the platform says it can but has no method for
            it.
    """
    connection = await self.connection()
    platform = self._client.platform_for(connection.platform)
    _refuse(platform, Feature.READ_POST, "reading one post back in full")
    if not isinstance(platform, CanReadPost):
        raise _missing_method(platform, "read_post")
    return await platform.read_post(connection, post_id)

read_thread async

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

Read a post together with the replies underneath it.

Parameters:

Name	Type	Description	Default
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.

Raises:

Type	Description
NotSupportedError	If this network cannot be asked for a whole thread this way.
ConfigError	If the platform says it can but has no method for it.

Source code in src/socialchimp/client.py

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

    Args:
        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.

    Raises:
        NotSupportedError: If this network cannot be asked for a whole
            thread this way.
        ConfigError: If the platform says it can but has no method for
            it.
    """
    connection = await self.connection()
    platform = self._client.platform_for(connection.platform)
    _refuse(
        platform, Feature.READ_THREAD, "reading a post together with its replies"
    )
    if not isinstance(platform, CanReadThread):
        raise _missing_method(platform, "read_thread")
    return await platform.read_thread(connection, post_id, depth=depth, limit=limit)

reply async

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

Reply to any post or comment, at any depth.

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

Parameters:

Name	Type	Description	Default
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
NotSupportedError	If this network has no way to reply to a comment this way.
ConfigError	If the platform says it can but has no method for it.

Source code in src/socialchimp/client.py

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

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

    Args:
        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:
        NotSupportedError: If this network has no way to reply to a
            comment this way.
        ConfigError: If the platform says it can but has no method for
            it.
    """
    connection = await self.connection()
    platform = self._client.platform_for(connection.platform)
    _refuse(platform, Feature.REPLY_TO_COMMENTS, "replying to a comment")
    if not isinstance(platform, CanReply):
        raise _missing_method(platform, "reply")
    return await platform.reply(
        connection, post_id, text, media=media, options=options
    )

like async

like(post_id: str) -> LikeResult

Like a post or a comment.

Liking something already liked succeeds and does nothing.

Parameters:

Name	Type	Description	Default
post_id	str	The post or comment to like.	required

Returns:

Type	Description
LikeResult	What the network said about the like.

Raises:

Type	Description
NotSupportedError	If this network cannot like a post.
ConfigError	If the platform says it can but has no method for it.

Source code in src/socialchimp/client.py

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

    Liking something already liked succeeds and does nothing.

    Args:
        post_id: The post or comment to like.

    Returns:
        What the network said about the like.

    Raises:
        NotSupportedError: If this network cannot like a post.
        ConfigError: If the platform says it can but has no method for
            it.
    """
    connection = await self.connection()
    platform = self._client.platform_for(connection.platform)
    _refuse(platform, Feature.LIKE, "liking a post")
    if not isinstance(platform, CanLike):
        raise _missing_method(platform, "like")
    return await platform.like(connection, post_id)

unlike async

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

Take back a like on a post or a comment.

Unliking something not liked succeeds and does nothing.

Parameters:

Name	Type	Description	Default
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.	None

Raises:

Type	Description
NotSupportedError	If this network cannot unlike a post.
ConfigError	If the platform says it can but has no method for it.

Source code in src/socialchimp/client.py

async def unlike(self, post_id: str, *, like_id: str | None = None) -> None:
    """Take back a like on a post or a comment.

    Unliking something not liked succeeds and does nothing.

    Args:
        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.

    Raises:
        NotSupportedError: If this network cannot unlike a post.
        ConfigError: If the platform says it can but has no method for
            it.
    """
    connection = await self.connection()
    platform = self._client.platform_for(connection.platform)
    _refuse(platform, Feature.LIKE, "unliking a post")
    if not isinstance(platform, CanLike):
        raise _missing_method(platform, "unlike")
    await platform.unlike(connection, post_id, like_id=like_id)

read_likes async

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

List who liked a post.

Parameters:

Name	Type	Description	Default
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.

Raises:

Type	Description
NotSupportedError	If this network cannot list who liked a post. Some networks that can like something cannot list who did.
ConfigError	If the platform says it can but has no method for it.

Source code in src/socialchimp/client.py

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

    Args:
        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.

    Raises:
        NotSupportedError: If this network cannot list who liked a post.
            Some networks that can like something cannot list who did.
        ConfigError: If the platform says it can but has no method for
            it.
    """
    connection = await self.connection()
    platform = self._client.platform_for(connection.platform)
    _refuse(platform, Feature.READ_LIKES, "listing who liked a post")
    if not isinstance(platform, CanReadLikes):
        raise _missing_method(platform, "read_likes")
    return await platform.read_likes(connection, post_id, after=after, limit=limit)

fetch_updates_after async

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

Read what has happened since a marker, resumable across a restart.

Unlike fetch_updates, which takes a moment in time, this takes an opaque marker your app stores and passes back - see socialchimp.events.UpdateBatch. A moment in time can miss or repeat updates around the edges; a marker cannot.

Parameters:

Name	Type	Description	Default
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.

Raises:

Type	Description
NotSupportedError	If this network cannot be polled with a marker this way.
ConfigError	If the platform says it can but has no method for it.

Source code in src/socialchimp/client.py

async def fetch_updates_after(
    self,
    marker: str | None,
    *,
    limit: int | None = None,
) -> UpdateBatch:
    """Read what has happened since a marker, resumable across a restart.

    Unlike `fetch_updates`, which takes a moment in time, this takes an
    opaque marker your app stores and passes back - see
    `socialchimp.events.UpdateBatch`. A moment in time can miss or
    repeat updates around the edges; a marker cannot.

    Args:
        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.

    Raises:
        NotSupportedError: If this network cannot be polled with a
            marker this way.
        ConfigError: If the platform says it can but has no method for
            it.
    """
    connection = await self.connection()
    platform = self._client.platform_for(connection.platform)
    _refuse(
        platform,
        Feature.READ_UPDATES_AFTER,
        "being asked what has happened since a marker",
    )
    if not isinstance(platform, CanReadUpdatesAfter):
        raise _missing_method(platform, "fetch_updates_after")
    return await platform.fetch_updates_after(connection, marker, limit=limit)

mark_seen async

mark_seen(marker: str) -> None

Tell the network a marker from fetch_updates_after has been seen.

Parameters:

Name	Type	Description	Default
marker	str	The marker that has been handled.	required

Raises:

Type	Description
NotSupportedError	If this network has no marker to mark seen.
ConfigError	If the platform says it can but has no method for it.

Source code in src/socialchimp/client.py

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

    Args:
        marker: The marker that has been handled.

    Raises:
        NotSupportedError: If this network has no marker to mark seen.
        ConfigError: If the platform says it can but has no method for
            it.
    """
    connection = await self.connection()
    platform = self._client.platform_for(connection.platform)
    _refuse(platform, Feature.READ_UPDATES_AFTER, "marking a marker as seen")
    if not isinstance(platform, CanReadUpdatesAfter):
        raise _missing_method(platform, "mark_seen")
    await platform.mark_seen(connection, marker)

read_conversations async

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

List this account's direct message conversations.

Parameters:

Name	Type	Description	Default
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.

Raises:

Type	Description
NotSupportedError	If this network has no direct messages.
ConfigError	If the platform says it can but has no method for it.

Source code in src/socialchimp/client.py

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

    Args:
        after: A `Page.next` from a previous call.
        limit: A cap on how many come back.

    Returns:
        One page of conversations.

    Raises:
        NotSupportedError: If this network has no direct messages.
        ConfigError: If the platform says it can but has no method for
            it.
    """
    connection = await self.connection()
    platform = self._client.platform_for(connection.platform)
    _refuse(platform, Feature.MESSAGES, "reading direct message conversations")
    if not isinstance(platform, CanMessage):
        raise _missing_method(platform, "read_conversations")
    return await platform.read_conversations(connection, after=after, limit=limit)

read_messages async

read_messages(
    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
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.

Raises:

Type	Description
NotSupportedError	If this network has no direct messages.
ConfigError	If the platform says it can but has no method for it.

Source code in src/socialchimp/client.py

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

    Args:
        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.

    Raises:
        NotSupportedError: If this network has no direct messages.
        ConfigError: If the platform says it can but has no method for
            it.
    """
    connection = await self.connection()
    platform = self._client.platform_for(connection.platform)
    _refuse(platform, Feature.MESSAGES, "reading direct messages")
    if not isinstance(platform, CanMessage):
        raise _missing_method(platform, "read_messages")
    return await platform.read_messages(
        connection, conversation_id, after=after, limit=limit
    )

send_message async

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

Send a message into an existing conversation.

Parameters:

Name	Type	Description	Default
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
NotSupportedError	If this network has no direct messages.
ConfigError	If the platform says it can but has no method for it.

Source code in src/socialchimp/client.py

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

    Args:
        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:
        NotSupportedError: If this network has no direct messages.
        ConfigError: If the platform says it can but has no method for
            it.
    """
    connection = await self.connection()
    platform = self._client.platform_for(connection.platform)
    _refuse(platform, Feature.MESSAGES, "sending a direct message")
    if not isinstance(platform, CanMessage):
        raise _missing_method(platform, "send_message")
    return await platform.send_message(
        connection, conversation_id, text, options=options
    )

mark_read async

mark_read(conversation_id: str) -> None

Mark a conversation as read.

Parameters:

Name	Type	Description	Default
conversation_id	str	Which conversation to mark.	required

Raises:

Type	Description
NotSupportedError	If this network has no direct messages.
ConfigError	If the platform says it can but has no method for it.

Source code in src/socialchimp/client.py

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

    Args:
        conversation_id: Which conversation to mark.

    Raises:
        NotSupportedError: If this network has no direct messages.
        ConfigError: If the platform says it can but has no method for
            it.
    """
    connection = await self.connection()
    platform = self._client.platform_for(connection.platform)
    _refuse(platform, Feature.MESSAGES, "marking a conversation as read")
    if not isinstance(platform, CanMessage):
        raise _missing_method(platform, "mark_read")
    await platform.mark_read(connection, conversation_id)

start_conversation async

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

Start a new conversation with one or more people.

Parameters:

Name	Type	Description	Default
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.

Raises:

Type	Description
NotSupportedError	If this network cannot start a conversation. Meta cannot: the customer has to write first.
ConfigError	If the platform says it can but has no method for it.

Source code in src/socialchimp/client.py

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

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

    Returns:
        The message that was sent.

    Raises:
        NotSupportedError: If this network cannot start a conversation.
            Meta cannot: the customer has to write first.
        ConfigError: If the platform says it can but has no method for
            it.
    """
    connection = await self.connection()
    platform = self._client.platform_for(connection.platform)
    _refuse(platform, Feature.START_CONVERSATIONS, "starting a new conversation")
    if not isinstance(platform, CanStartConversations):
        raise _missing_method(platform, "start_conversation")
    return await platform.start_conversation(connection, person_ids, text)

features async

features() -> Feature

Ask what this account's network can do.

Looks the connection up lazily, the same as every other call here, so this is safe to call before deciding which of the calls above to make.

Returns:

Type	Description
Feature	The features this network supports.

Source code in src/socialchimp/client.py

async def features(self) -> Feature:
    """Ask what this account's network can do.

    Looks the connection up lazily, the same as every other call here,
    so this is safe to call before deciding which of the calls above to
    make.

    Returns:
        The features this network supports.
    """
    connection = await self.connection()
    return self._client.platform_for(connection.platform).features

delete_post async

delete_post(post_id: str) -> None

Take a post back down again.

Parameters:

Name	Type	Description	Default
post_id	str	The network's identifier for the post, which is what post() handed back.	required

Raises:

Type	Description
NotSupportedError	If this network cannot remove posts.
ConfigError	If the platform says it can but has no method for it.

Source code in src/socialchimp/client.py

async def delete_post(self, post_id: str) -> None:
    """Take a post back down again.

    Args:
        post_id: The network's identifier for the post, which is what
            `post()` handed back.

    Raises:
        NotSupportedError: If this network cannot remove posts.
        ConfigError: If the platform says it can but has no method for
            it.
    """
    connection = await self.connection()
    platform = self._client.platform_for(connection.platform)
    _refuse(platform, Feature.DELETE_POST, "removing a post once it is published")
    if not isinstance(platform, CanDeletePosts):
        raise _missing_method(platform, "delete_post")
    await platform.delete_post(connection, post_id)

read_stats async

read_stats(post_id: str) -> PostStats

Read how one of this account's posts is doing.

The token is renewed first, the same as every other call here, so this is safe to put on a timer.

Parameters:

Name	Type	Description	Default
post_id	str	The network's identifier for the post, which is what post() handed back.	required

Returns:

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

Raises:

Type	Description
NotSupportedError	If this network keeps no numbers an app can read.
ConfigError	If the platform says it can but has no method for it.

Source code in src/socialchimp/client.py

async def read_stats(self, post_id: str) -> PostStats:
    """Read how one of this account's posts is doing.

    The token is renewed first, the same as every other call here, so
    this is safe to put on a timer.

    Args:
        post_id: The network's identifier for the post, which is what
            `post()` handed back.

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

    Raises:
        NotSupportedError: If this network keeps no numbers an app can
            read.
        ConfigError: If the platform says it can but has no method for
            it.
    """
    connection = await self.connection()
    platform = self._client.platform_for(connection.platform)
    _refuse(platform, Feature.READ_STATS, "reading a post's numbers back")
    if not isinstance(platform, CanReadStats):
        raise _missing_method(platform, "read_stats")
    return await platform.read_stats(connection, post_id)

reply_to_update async

reply_to_update(update: Update, text: str) -> None

Answer an update - a review, a question - in place.

A review or a question is not a post, so there is nothing on post() for it. The token is renewed first, the same as every other call here.

Parameters:

Name	Type	Description	Default
update	Update	The update to answer, exactly as fetch_updates or SocialChimp.read_updates handed it back.	required
text	str	The reply.	required

Raises:

Type	Description
NotSupportedError	If this network has nothing to answer, or cannot answer this kind of update.

Source code in src/socialchimp/client.py

async def reply_to_update(self, update: Update, text: str) -> None:
    """Answer an update - a review, a question - in place.

    A review or a question is not a post, so there is nothing on `post()`
    for it. The token is renewed first, the same as every other call
    here.

    Args:
        update: The update to answer, exactly as `fetch_updates` or
            `SocialChimp.read_updates` handed it back.
        text: The reply.

    Raises:
        NotSupportedError: If this network has nothing to answer, or
            cannot answer this kind of update.
    """
    connection = await self.connection()
    platform = self._client.platform_for(connection.platform)
    if not isinstance(platform, CanReplyToUpdates):
        raise NotSupportedError(
            platform=platform.name,
            what="answering a review or a question",
            suggestion=(
                "It keeps nothing worth replying to, or its platform "
                "file has no reply_to_update yet."
            ),
        )
    await platform.reply_to_update(connection, update, text)

delete_comment async

delete_comment(update: Update) -> None

Remove a comment outright.

The token is renewed first, the same as every other call here.

Parameters:

Name	Type	Description	Default
update	Update	The comment to remove, exactly as fetch_updates or SocialChimp.read_updates handed it back.	required

Raises:

Type	Description
NotSupportedError	If this network cannot remove this kind of update.

Source code in src/socialchimp/client.py

async def delete_comment(self, update: Update) -> None:
    """Remove a comment outright.

    The token is renewed first, the same as every other call here.

    Args:
        update: The comment to remove, exactly as `fetch_updates` or
            `SocialChimp.read_updates` handed it back.

    Raises:
        NotSupportedError: If this network cannot remove this kind of
            update.
    """
    connection = await self.connection()
    platform = self._client.platform_for(connection.platform)
    if not isinstance(platform, CanModerateComments):
        raise NotSupportedError(
            platform=platform.name,
            what="removing a comment",
            suggestion=(
                "It keeps nothing that can be removed this way, or its "
                "platform file has no delete_comment yet."
            ),
        )
    await platform.delete_comment(connection, update)

set_comment_visibility async

set_comment_visibility(
    update: Update, *, hidden: bool
) -> None

Hide a comment from public view, or show one again.

The token is renewed first, the same as every other call here.

Parameters:

Name	Type	Description	Default
update	Update	The comment to hide or show, exactly as fetch_updates or SocialChimp.read_updates handed it back.	required
hidden	bool	True to hide it, False to show it again.	required

Raises:

Type	Description
NotSupportedError	If this network has no visibility to change on this kind of update.

Source code in src/socialchimp/client.py

async def set_comment_visibility(self, update: Update, *, hidden: bool) -> None:
    """Hide a comment from public view, or show one again.

    The token is renewed first, the same as every other call here.

    Args:
        update: The comment to hide or show, exactly as `fetch_updates`
            or `SocialChimp.read_updates` handed it back.
        hidden: `True` to hide it, `False` to show it again.

    Raises:
        NotSupportedError: If this network has no visibility to change
            on this kind of update.
    """
    connection = await self.connection()
    platform = self._client.platform_for(connection.platform)
    if not isinstance(platform, CanModerateComments):
        raise NotSupportedError(
            platform=platform.name,
            what="changing whether a comment is hidden",
            suggestion=(
                "It keeps nothing with visibility to change this way, "
                "or its platform file has no set_comment_visibility yet."
            ),
        )
    await platform.set_comment_visibility(connection, update, hidden=hidden)

get_location async

get_location() -> BusinessLocation

Read the business information this account holds.

Returns:

Type	Description
BusinessLocation	What the network currently has on file.

Raises:

Type	Description
NotSupportedError	If this network keeps nothing beyond posts.

Source code in src/socialchimp/client.py

async def get_location(self) -> BusinessLocation:
    """Read the business information this account holds.

    Returns:
        What the network currently has on file.

    Raises:
        NotSupportedError: If this network keeps nothing beyond posts.
    """
    connection = await self.connection()
    platform = self._client.platform_for(connection.platform)
    if not isinstance(platform, CanEditBusinessInfo):
        raise NotSupportedError(
            platform=platform.name,
            what="reading business information back",
            suggestion="It has nothing beyond posts to read.",
        )
    return await platform.get_location(connection)

update_location async

update_location(fields: RawData) -> BusinessLocation

Change some of this account's business information.

Parameters:

Name	Type	Description	Default
fields	RawData	The fields to change, named the way the network's own API names them. Fields left out are left alone.	required

Returns:

Type	Description
BusinessLocation	The location as it stands after the change.

Raises:

Type	Description
NotSupportedError	If this network keeps nothing beyond posts.

Source code in src/socialchimp/client.py

async def update_location(self, fields: RawData) -> BusinessLocation:
    """Change some of this account's business information.

    Args:
        fields: The fields to change, named the way the network's own
            API names them. Fields left out are left alone.

    Returns:
        The location as it stands after the change.

    Raises:
        NotSupportedError: If this network keeps nothing beyond posts.
    """
    connection = await self.connection()
    platform = self._client.platform_for(connection.platform)
    if not isinstance(platform, CanEditBusinessInfo):
        raise NotSupportedError(
            platform=platform.name,
            what="changing business information",
            suggestion="It has nothing beyond posts to change.",
        )
    return await platform.update_location(connection, fields)

verification_options async

verification_options() -> Sequence[VerificationOption]

List the ways this account's location could be verified right now.

Returns:

Type	Description
Sequence[VerificationOption]	What the network will offer.

Raises:

Type	Description
NotSupportedError	If this network has no verification process.

Source code in src/socialchimp/client.py

async def verification_options(self) -> Sequence[VerificationOption]:
    """List the ways this account's location could be verified right now.

    Returns:
        What the network will offer.

    Raises:
        NotSupportedError: If this network has no verification process.
    """
    connection = await self.connection()
    platform = self._client.platform_for(connection.platform)
    if not isinstance(platform, CanManageVerification):
        raise NotSupportedError(
            platform=platform.name,
            what="verifying a location",
            suggestion="It has no verification process of its own.",
        )
    return await platform.verification_options(connection)

start_verification async

start_verification(method: str) -> Verification

Ask the network to verify this account's location.

This is what makes the network act - mail a postcard, place a call, send a text or an email. Nothing about the proof passes through socialchimp.

Parameters:

Name	Type	Description	Default
method	str	One of the methods verification_options offered.	required

Returns:

Type	Description
Verification	The verification now in progress.

Raises:

Type	Description
NotSupportedError	If this network has no verification process.

Source code in src/socialchimp/client.py

async def start_verification(self, method: str) -> Verification:
    """Ask the network to verify this account's location.

    This is what makes the network act - mail a postcard, place a call,
    send a text or an email. Nothing about the proof passes through
    socialchimp.

    Args:
        method: One of the methods `verification_options` offered.

    Returns:
        The verification now in progress.

    Raises:
        NotSupportedError: If this network has no verification process.
    """
    connection = await self.connection()
    platform = self._client.platform_for(connection.platform)
    if not isinstance(platform, CanManageVerification):
        raise NotSupportedError(
            platform=platform.name,
            what="verifying a location",
            suggestion="It has no verification process of its own.",
        )
    return await platform.start_verification(connection, method)

complete_verification async

complete_verification(
    verification_id: str, pin: str
) -> Verification

Finish a verification with the code the business owner was sent.

Parameters:

Name	Type	Description	Default
verification_id	str	The id start_verification returned.	required
pin	str	The code the business owner received.	required

Returns:

Type	Description
Verification	The verification's new state.

Raises:

Type	Description
NotSupportedError	If this network has no verification process.

Source code in src/socialchimp/client.py

async def complete_verification(
    self,
    verification_id: str,
    pin: str,
) -> Verification:
    """Finish a verification with the code the business owner was sent.

    Args:
        verification_id: The id `start_verification` returned.
        pin: The code the business owner received.

    Returns:
        The verification's new state.

    Raises:
        NotSupportedError: If this network has no verification process.
    """
    connection = await self.connection()
    platform = self._client.platform_for(connection.platform)
    if not isinstance(platform, CanManageVerification):
        raise NotSupportedError(
            platform=platform.name,
            what="verifying a location",
            suggestion="It has no verification process of its own.",
        )
    return await platform.complete_verification(connection, verification_id, pin)

verification_state async

verification_state() -> str

Ask where this account's location's verification stands.

Returns:

Type	Description
str	The network's own word for the state.

Raises:

Type	Description
NotSupportedError	If this network has no verification process.

Source code in src/socialchimp/client.py

async def verification_state(self) -> str:
    """Ask where this account's location's verification stands.

    Returns:
        The network's own word for the state.

    Raises:
        NotSupportedError: If this network has no verification process.
    """
    connection = await self.connection()
    platform = self._client.platform_for(connection.platform)
    if not isinstance(platform, CanManageVerification):
        raise NotSupportedError(
            platform=platform.name,
            what="verifying a location",
            suggestion="It has no verification process of its own.",
        )
    return await platform.verification_state(connection)

Sending your own request

account.direct sends a request of your own to the same network, through the same token, the same retries and the same rate-limit handling. Only the request itself is yours - see the tutorial for why this exists.

Direct

Direct(client: SocialChimp, connection_id: str)

Your own requests to a network, sent as one connected account.

Reached through Account.direct. The token is renewed before every request, and retries and rate limits are handled exactly as they are for post(). Only the request itself is yours.

reply = await account.direct.post(
    "/api/v1/statuses",
    json={"status": "hello", "visibility": "unlisted"},
)

Paths are joined onto the address the platform gives for this account - the account's own server for Mastodon, the one address everybody uses for Facebook. Pass a whole address instead and it is used as it is.

Point direct access at one connected account.

Parameters:

Name	Type	Description	Default
client	SocialChimp	The client this account belongs to.	required
connection_id	str	The id your app gave this connection.	required

Source code in src/socialchimp/client.py

def __init__(self, client: SocialChimp, connection_id: str) -> None:
    """Point direct access at one connected account.

    Args:
        client: The client this account belongs to.
        connection_id: The id your app gave this connection.
    """
    self._client = client
    self._connection_id = connection_id

request async

request(
    method: str,
    path: str,
    *,
    headers: Mapping[str, str] | None = None,
    **kwargs: object,
) -> Response

Send a request as this account.

Parameters:

Name	Type	Description	Default
method	str	"GET", "POST" and so on.	required
path	str	Joined onto the address the platform gives for this account.	required
headers	Mapping[str, str] | None	Sent along with the ones the platform set. A header you set here wins, so a request that has to be signed some other way is still yours to send.	None
**kwargs	object	Anything httpx.AsyncClient.request takes, such as params, json, content or files.	{}

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. See socialchimp.http.error_from_response.

Source code in src/socialchimp/client.py

async def request(
    self,
    method: str,
    path: str,
    *,
    headers: Mapping[str, str] | None = None,
    **kwargs: object,
) -> httpx.Response:
    """Send a request as this account.

    Args:
        method: `"GET"`, `"POST"` and so on.
        path: Joined onto the address the platform gives for this
            account.
        headers: Sent along with the ones the platform set. A header
            you set here wins, so a request that has to be signed some
            other way is still yours to send.
        **kwargs: Anything `httpx.AsyncClient.request` takes, such as
            `params`, `json`, `content` or `files`.

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

    Raises:
        SocialChimpError: If the network refused, or could not be
            reached. See `socialchimp.http.error_from_response`.
    """
    http, sending = await self._ready(headers)
    return await http.request(method, path, headers=sending, **kwargs)

get async

get(
    path: str,
    *,
    headers: Mapping[str, str] | None = None,
    **kwargs: object,
) -> Response

Send a GET request as this account.

Parameters:

Name	Type	Description	Default
path	str	Joined onto the address the platform gives for this account.	required
headers	Mapping[str, str] | None	Sent along with the ones the platform set.	None
**kwargs	object	Anything request takes.	{}

Returns:

Type	Description
Response	The reply.

Source code in src/socialchimp/client.py

async def get(
    self,
    path: str,
    *,
    headers: Mapping[str, str] | None = None,
    **kwargs: object,
) -> httpx.Response:
    """Send a GET request as this account.

    Args:
        path: Joined onto the address the platform gives for this
            account.
        headers: Sent along with the ones the platform set.
        **kwargs: Anything `request` takes.

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

post async

post(
    path: str,
    *,
    headers: Mapping[str, str] | None = None,
    **kwargs: object,
) -> Response

Send a POST request as this account.

Parameters:

Name	Type	Description	Default
path	str	Joined onto the address the platform gives for this account.	required
headers	Mapping[str, str] | None	Sent along with the ones the platform set.	None
**kwargs	object	Anything request takes.	{}

Returns:

Type	Description
Response	The reply.

Source code in src/socialchimp/client.py

async def post(
    self,
    path: str,
    *,
    headers: Mapping[str, str] | None = None,
    **kwargs: object,
) -> httpx.Response:
    """Send a POST request as this account.

    Args:
        path: Joined onto the address the platform gives for this
            account.
        headers: Sent along with the ones the platform set.
        **kwargs: Anything `request` takes.

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

put async

put(
    path: str,
    *,
    headers: Mapping[str, str] | None = None,
    **kwargs: object,
) -> Response

Send a PUT request as this account.

Parameters:

Name	Type	Description	Default
path	str	Joined onto the address the platform gives for this account.	required
headers	Mapping[str, str] | None	Sent along with the ones the platform set.	None
**kwargs	object	Anything request takes.	{}

Returns:

Type	Description
Response	The reply.

Source code in src/socialchimp/client.py

async def put(
    self,
    path: str,
    *,
    headers: Mapping[str, str] | None = None,
    **kwargs: object,
) -> httpx.Response:
    """Send a PUT request as this account.

    Args:
        path: Joined onto the address the platform gives for this
            account.
        headers: Sent along with the ones the platform set.
        **kwargs: Anything `request` takes.

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

delete async

delete(
    path: str,
    *,
    headers: Mapping[str, str] | None = None,
    **kwargs: object,
) -> Response

Send a DELETE request as this account.

Parameters:

Name	Type	Description	Default
path	str	Joined onto the address the platform gives for this account.	required
headers	Mapping[str, str] | None	Sent along with the ones the platform set.	None
**kwargs	object	Anything request takes.	{}

Returns:

Type	Description
Response	The reply.

Source code in src/socialchimp/client.py

async def delete(
    self,
    path: str,
    *,
    headers: Mapping[str, str] | None = None,
    **kwargs: object,
) -> httpx.Response:
    """Send a DELETE request as this account.

    Args:
        path: Joined onto the address the platform gives for this
            account.
        headers: Sent along with the ones the platform set.
        **kwargs: Anything `request` takes.

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

json async

json(
    method: str,
    path: str,
    *,
    headers: Mapping[str, str] | None = None,
    **kwargs: object,
) -> RawData

Send a request as this account and read the reply as JSON.

Parameters:

Name	Type	Description	Default
method	str	"GET", "POST" and so on.	required
path	str	Joined onto the address the platform gives for this account.	required
headers	Mapping[str, str] | None	Sent along with the ones the platform set.	None
**kwargs	object	Anything request takes.	{}

Returns:

Type	Description
RawData	The reply, parsed.

Raises:

Type	Description
PlatformError	If the reply was not a JSON object.
SocialChimpError	If the network refused the request.

Source code in src/socialchimp/client.py

async def json(
    self,
    method: str,
    path: str,
    *,
    headers: Mapping[str, str] | None = None,
    **kwargs: object,
) -> RawData:
    """Send a request as this account and read the reply as JSON.

    Args:
        method: `"GET"`, `"POST"` and so on.
        path: Joined onto the address the platform gives for this
            account.
        headers: Sent along with the ones the platform set.
        **kwargs: Anything `request` takes.

    Returns:
        The reply, parsed.

    Raises:
        PlatformError: If the reply was not a JSON object.
        SocialChimpError: If the network refused the request.
    """
    http, sending = await self._ready(headers)
    return await http.json(method, path, headers=sending, **kwargs)

Posting to more than one account

There is nothing here for it, on purpose. Account.post posts as one account and raises if that account fails; looping over your accounts, and deciding what one failure means for the rest, is your app's job. See the tutorial.

Keeping tokens working

SocialChimp uses this to renew a token a little before it runs out, taking a lock first so two workers renewing the same connection at once cannot disconnect an account. You will not normally construct this yourself.

TokenManager

TokenManager(
    storage: Storage,
    get_new_token: GetNewToken,
    *,
    refresh_before_seconds: float = DEFAULT_REFRESH_BEFORE_SECONDS,
    make_lock: MakeLock = _lock_within_this_process,
)

Hands out connections whose token is usable right now.

Ask it for a connection and it either gives you the one you have, or renews the token first, saves it, and gives you that.

tokens = TokenManager(storage, renew)
connection = await tokens.valid_token("conn-1")

One of these can be shared by everything in your process, and should be: the locks that stop two renewals colliding live on the instance, so a new TokenManager per request protects nothing.

Set up token renewal for one app.

Parameters:

Name	Type	Description	Default
storage	Storage	Where connections are read from and written back to.	required
get_new_token	GetNewToken	Asks a network for a new token. Wrap Platform.refresh in something that looks your app's credentials up first - most networks will not renew without them, and SocialChimp does exactly that when it makes one of these itself.	required
refresh_before_seconds	float	How long before a token runs out to renew it. The default of 60 seconds leaves room for a slow request.	DEFAULT_REFRESH_BEFORE_SECONDS
make_lock	MakeLock	Makes the lock used while renewing one connection. The default only holds inside this process; pass your own, backed by something like Redis, if you run more than one.	_lock_within_this_process

Source code in src/socialchimp/tokens.py

def __init__(
    self,
    storage: Storage,
    get_new_token: GetNewToken,
    *,
    refresh_before_seconds: float = DEFAULT_REFRESH_BEFORE_SECONDS,
    make_lock: MakeLock = _lock_within_this_process,
) -> None:
    """Set up token renewal for one app.

    Args:
        storage: Where connections are read from and written back to.
        get_new_token: Asks a network for a new token. Wrap
            `Platform.refresh` in something that looks your app's
            credentials up first - most networks will not renew without
            them, and `SocialChimp` does exactly that when it makes one
            of these itself.
        refresh_before_seconds: How long before a token runs out to renew
            it. The default of 60 seconds leaves room for a slow request.
        make_lock: Makes the lock used while renewing one connection. The
            default only holds inside this process; pass your own, backed
            by something like Redis, if you run more than one.
    """
    self._storage = storage
    self._get_new_token = get_new_token
    self._refresh_before_seconds = refresh_before_seconds
    self._make_lock = make_lock
    self._locks: dict[str, Lock] = {}
    self._listeners: list[TokenRenewed] = []

on_token_renewed

on_token_renewed(listener: TokenRenewed) -> None

Ask to be told whenever a token was renewed.

Handy for logging, or for warming a cache of your own. socialchimp has already saved the connection by the time you hear about it, so there is nothing you must do.

Anything your listener raises is logged and dropped. A listener watches; it never gets to fail a renewal.

Parameters:

Name	Type	Description	Default
listener	TokenRenewed	Called with the connection carrying its new token.	required

Source code in src/socialchimp/tokens.py

def on_token_renewed(self, listener: TokenRenewed) -> None:
    """Ask to be told whenever a token was renewed.

    Handy for logging, or for warming a cache of your own. socialchimp
    has already saved the connection by the time you hear about it, so
    there is nothing you must do.

    Anything your listener raises is logged and dropped. A listener
    watches; it never gets to fail a renewal.

    Args:
        listener: Called with the connection carrying its new token.
    """
    self._listeners.append(listener)

valid_token async

valid_token(connection_id: str) -> Connection

Return a connection whose token works right now.

Renews the token first if it is close to running out. Safe to call from anywhere, as often as you like - a connection that is fine costs one read.

Parameters:

Name	Type	Description	Default
connection_id	str	The id your app gave this connection.	required

Returns:

Type	Description
Connection	The connection, with a token that is good for a while yet.

Raises:

Type	Description
ConfigError	If no connection is stored under that id.
TokenExpiredError	If the token needed renewing and could not be, because there is no refresh token, because the refresh token has itself run out, or because the network refused the one we have. The person has to sign in again.

Source code in src/socialchimp/tokens.py

async def valid_token(self, connection_id: str) -> Connection:
    """Return a connection whose token works right now.

    Renews the token first if it is close to running out. Safe to call
    from anywhere, as often as you like - a connection that is fine costs
    one read.

    Args:
        connection_id: The id your app gave this connection.

    Returns:
        The connection, with a token that is good for a while yet.

    Raises:
        ConfigError: If no connection is stored under that id.
        TokenExpiredError: If the token needed renewing and could not be,
            because there is no refresh token, because the refresh token
            has itself run out, or because the network refused the one
            we have. The person has to sign in again.
    """
    connection = await self._load(connection_id)
    if not self._running_out(connection):
        return connection

    async with self._lock_for(connection_id):
        # Whoever else was renewing this connection has finished by now,
        # so read it again rather than trusting what we saw outside the
        # lock. Their new token is already saved, and renewing a second
        # time would throw it away.
        connection = await self._load(connection_id)
        if not self._running_out(connection):
            return connection
        return await self._renew(connection)