Framework helpers¶

The ready-made routes described in Frameworks, by signature. Importing socialchimp never imports any of these - you only pay for the one you use.

Django¶

urls ¶

urls(
    sc: SocialChimp,
    *,
    redirect_uri: str,
    memory: LoginMemory | None = None,
    scopes: Mapping[str, Sequence[str]] | None = None,
    secrets: Mapping[str, str] | None = None,
    setup_tokens: Mapping[str, str] | None = None,
    deliver: DeliverUpdate | None = None,
) -> list[UrlPattern]

Build the routes for signing in and receiving updates.

Parameters:

Name	Type	Description	Default
`sc`	`SocialChimp`	The client to work through. `get_client()` builds one from your settings and keeps it.	required
`redirect_uri`	`str`	Where networks send people back to. `{platform}` in it is replaced by the network's name.	required
`memory`	`LoginMemory \| None`	Where a half-finished sign-in waits. Left out, one that lives in this process is used - fine to try things out with, wrong in production, because two workers do not share it. See `shared.LoginMemory`.	`None`
`scopes`	`Mapping[str, Sequence[str]] \| None`	Permissions to ask each network for, by network name.	`None`
`secrets`	`Mapping[str, str] \| None`	The secret each network signs its webhooks with, by network name.	`None`
`setup_tokens`	`Mapping[str, str] \| None`	The token each network's setup check quotes back, by network name.	`None`
`deliver`	`DeliverUpdate \| None`	Where a webhook's update goes. `Dispatcher.deliver` fits. Giving `secrets` without it is refused, because it would mean checking a real update and then dropping it.	`None`

Returns:

Type	Description
`list[UrlPattern]`	Patterns to give `include()`, each with a name beginning
`list[UrlPattern]`	`socialchimp-`.

Source code in src/socialchimp/contrib/django.py

def urls(
    sc: SocialChimp,
    *,
    redirect_uri: str,
    memory: LoginMemory | None = None,
    scopes: Mapping[str, Sequence[str]] | None = None,
    secrets: Mapping[str, str] | None = None,
    setup_tokens: Mapping[str, str] | None = None,
    deliver: DeliverUpdate | None = None,
) -> list[UrlPattern]:
    """Build the routes for signing in and receiving updates.

    Args:
        sc: The client to work through. `get_client()` builds one from your
            settings and keeps it.
        redirect_uri: Where networks send people back to. `{platform}` in it
            is replaced by the network's name.
        memory: Where a half-finished sign-in waits. Left out, one that
            lives in this process is used - fine to try things out with,
            wrong in production, because two workers do not share it. See
            `shared.LoginMemory`.
        scopes: Permissions to ask each network for, by network name.
        secrets: The secret each network signs its webhooks with, by network
            name.
        setup_tokens: The token each network's setup check quotes back, by
            network name.
        deliver: Where a webhook's update goes. `Dispatcher.deliver` fits.
            Giving `secrets` without it is refused, because it would mean
            checking a real update and then dropping it.

    Returns:
        Patterns to give `include()`, each with a name beginning
        `socialchimp-`.
    """
    routes = Routes(
        sc,
        redirect_uri=redirect_uri,
        memory=memory,
        scopes=scopes,
        secrets=secrets,
        setup_tokens=setup_tokens,
        deliver=deliver,
    )
    pieces = _django()

    def connect(request: Request, *, platform: str) -> Response:
        """Begin signing someone in to one network."""
        return _answer(pieces.response, _run(routes.start(platform, _query(request))))

    def callback(request: Request, *, platform: str) -> Response:
        """Carry on after the person comes back from the network."""
        return _answer(pieces.response, _run(routes.finish(platform, _values(request))))

    def choose(request: Request, *, platform: str) -> Response:
        """Carry on after the person picked which account to use."""
        return _answer(pieces.response, _run(routes.choose(platform, _values(request))))

    def webhook(request: Request, *, platform: str) -> Response:
        """Answer a network's setup check, or receive an update from it."""
        if request.method == "GET":
            decided = _run(routes.setup_check(platform, _query(request)))
        else:
            # `request.body` is the bytes exactly as they arrived. Never
            # `json.loads` first: a signature is over those exact bytes, and
            # parsing the JSON and building it again changes the spacing and
            # the key order, so the signature no longer matches. That is the
            # single most common reason a correct signature appears to fail.
            decided = _run(
                routes.webhook(platform, request.body, dict(request.headers.items()))
            )
        return _answer(pieces.response, decided)

    return [
        pieces.path("connect/<str:platform>", connect, name="socialchimp-connect"),
        pieces.path("callback/<str:platform>", callback, name="socialchimp-callback"),
        pieces.path("choose/<str:platform>", choose, name="socialchimp-choose"),
        # Only the webhook is exempted. A social network has no way to send
        # one of Django's CSRF tokens, so a protected webhook answers 403 to
        # everything and the network eventually stops trying. The other three
        # are posted to by your own pages, so they keep Django's protection -
        # put {% csrf_token %} in those forms as usual.
        pieces.path(
            "webhooks/<str:platform>",
            pieces.csrf_exempt(webhook),
            name="socialchimp-webhook",
        ),
    ]

get_client `cached` ¶

get_client() -> SocialChimp

Return the one SocialChimp for this process, built from settings.

Reads settings.SOCIALCHIMP, which names your storage class and says which sort it is:

SOCIALCHIMP = {"SYNC_STORAGE": "myapp.social.MyStorage"}

Use SYNC_STORAGE for a class written as ordinary Django ORM code - which is what you want unless you have gone out of your way - and STORAGE for one whose five methods are already async. Exactly one of them, because guessing which you meant is the sort of thing that works until it does not.

The client is built once and kept, because the locks that stop two workers renewing the same token at once live on it. Call get_client.cache_clear() if you really need a new one.

Returns:

Type	Description
`SocialChimp`	The client.

Raises:

Type	Description
`ConfigError`	If the setting is missing, the wrong shape, or names a class that is not there.

Source code in src/socialchimp/contrib/django.py

@cache
def get_client() -> SocialChimp:
    """Return the one `SocialChimp` for this process, built from settings.

    Reads `settings.SOCIALCHIMP`, which names your storage class and says
    which sort it is:

        SOCIALCHIMP = {"SYNC_STORAGE": "myapp.social.MyStorage"}

    Use `SYNC_STORAGE` for a class written as ordinary Django ORM code -
    which is what you want unless you have gone out of your way - and
    `STORAGE` for one whose five methods are already async. Exactly one of
    them, because guessing which you meant is the sort of thing that works
    until it does not.

    The client is built once and kept, because the locks that stop two
    workers renewing the same token at once live on it. Call
    `get_client.cache_clear()` if you really need a new one.

    Returns:
        The client.

    Raises:
        ConfigError: If the setting is missing, the wrong shape, or names a
            class that is not there.
    """
    return SocialChimp(storage=_storage_from_settings(_django().settings))

orm_storage ¶

orm_storage(inner: SyncStorage) -> Storage

Let socialchimp use storage you wrote as ordinary Django ORM code.

Write the five methods with Model.objects.get(...) and .save(), the way you write everything else, and hand the class here.

Example

class MyStorage: def get_connection(self, connection_id): row = SocialAccount.objects.filter(pk=connection_id).first() return row.to_connection() if row else None ...

sc = SocialChimp(storage=orm_storage(MyStorage()))

Parameters:

Name	Type	Description	Default
`inner`	`SyncStorage`	Your storage class. Five methods, none of them async.	required

Returns:

Type	Description
`Storage`	A `Storage` to hand to `SocialChimp`.

Source code in src/socialchimp/contrib/django.py

def orm_storage(inner: SyncStorage) -> Storage:
    """Let socialchimp use storage you wrote as ordinary Django ORM code.

    Write the five methods with `Model.objects.get(...)` and `.save()`, the
    way you write everything else, and hand the class here.

    Example:
        class MyStorage:
            def get_connection(self, connection_id):
                row = SocialAccount.objects.filter(pk=connection_id).first()
                return row.to_connection() if row else None
            ...

        sc = SocialChimp(storage=orm_storage(MyStorage()))

    Args:
        inner: Your storage class. Five methods, none of them async.

    Returns:
        A `Storage` to hand to `SocialChimp`.
    """
    return sync_storage(inner, run=_on_the_request_thread)

Request ¶

Bases: Protocol

The little of Django's request these routes read.

Written down here because Django ships no type information, and because it is a short and useful list: the method, the raw body, the query values and the headers. A real HttpRequest has all four.

Attributes:

Name	Type	Description
`method`	`str`	`"GET"`, `"POST"` and so on.
`body`	`bytes`	The request body, exactly as it arrived. This is the one that matters for webhooks.
`GET`	`Mapping[str, str]`	The query values.
`headers`	`Mapping[str, str]`	The request headers.

View ¶

Bases: Protocol

One of the views below, as Django will call it.

MakeResponse ¶

Bases: Protocol

Django's HttpResponse, as much of it as we use.

MakePath ¶

Bases: Protocol

Django's path, as much of it as we use.

Exempt ¶

Bases: Protocol

Django's csrf_exempt.

Settings ¶

Bases: Protocol

Django's settings, which we only ever read one name out of.

FastAPI¶

router ¶

router(
    sc: SocialChimp,
    *,
    redirect_uri: str,
    memory: LoginMemory | None = None,
    scopes: Mapping[str, Sequence[str]] | None = None,
    secrets: Mapping[str, str] | None = None,
    setup_tokens: Mapping[str, str] | None = None,
    deliver: DeliverUpdate | None = None,
) -> APIRouter

Build the routes for signing in and receiving updates.

Parameters:

Name	Type	Description	Default
`sc`	`SocialChimp`	The client to work through. Keep one for the life of your process, and hand the same one to your own code.	required
`redirect_uri`	`str`	Where networks send people back to. `{platform}` in it is replaced by the network's name.	required
`memory`	`LoginMemory \| None`	Where a half-finished sign-in waits. Left out, one that lives in this process is used - fine to try things out with, wrong in production. See `shared.LoginMemory`.	`None`
`scopes`	`Mapping[str, Sequence[str]] \| None`	Permissions to ask each network for, by network name.	`None`
`secrets`	`Mapping[str, str] \| None`	The secret each network signs its webhooks with, by network name.	`None`
`setup_tokens`	`Mapping[str, str] \| None`	The token each network's setup check quotes back, by network name.	`None`
`deliver`	`DeliverUpdate \| None`	Where a webhook's update goes. `Dispatcher.deliver` fits. Giving `secrets` without it is refused, because it would mean checking a real update and then dropping it.	`None`

Returns:

Type	Description
`APIRouter`	A router to give `app.include_router`.

Source code in src/socialchimp/contrib/fastapi.py

def router(
    sc: SocialChimp,
    *,
    redirect_uri: str,
    memory: LoginMemory | None = None,
    scopes: Mapping[str, Sequence[str]] | None = None,
    secrets: Mapping[str, str] | None = None,
    setup_tokens: Mapping[str, str] | None = None,
    deliver: DeliverUpdate | None = None,
) -> APIRouter:
    """Build the routes for signing in and receiving updates.

    Args:
        sc: The client to work through. Keep one for the life of your
            process, and hand the same one to your own code.
        redirect_uri: Where networks send people back to. `{platform}` in it
            is replaced by the network's name.
        memory: Where a half-finished sign-in waits. Left out, one that
            lives in this process is used - fine to try things out with,
            wrong in production. See `shared.LoginMemory`.
        scopes: Permissions to ask each network for, by network name.
        secrets: The secret each network signs its webhooks with, by network
            name.
        setup_tokens: The token each network's setup check quotes back, by
            network name.
        deliver: Where a webhook's update goes. `Dispatcher.deliver` fits.
            Giving `secrets` without it is refused, because it would mean
            checking a real update and then dropping it.

    Returns:
        A router to give `app.include_router`.
    """
    routes = Routes(
        sc,
        redirect_uri=redirect_uri,
        memory=memory,
        scopes=scopes,
        secrets=secrets,
        setup_tokens=setup_tokens,
        deliver=deliver,
    )
    api = APIRouter()

    @api.get("/connect/{platform}")
    async def connect(platform: str, request: Request) -> Response:
        """Begin signing someone in to one network."""
        return _answer(await routes.start(platform, dict(request.query_params)))

    @api.api_route("/callback/{platform}", methods=["GET", "POST"])
    async def callback(platform: str, request: Request) -> Response:
        """Carry on after the person comes back from the network."""
        return _answer(await routes.finish(platform, await _values(request)))

    @api.post("/choose/{platform}")
    async def choose(platform: str, request: Request) -> Response:
        """Carry on after the person picked which account to use."""
        return _answer(await routes.choose(platform, await _values(request)))

    @api.get("/webhooks/{platform}")
    async def setup_check(platform: str, request: Request) -> Response:
        """Answer the check a network makes before it will send anything."""
        return _answer(await routes.setup_check(platform, dict(request.query_params)))

    @api.post("/webhooks/{platform}")
    async def webhook(platform: str, request: Request) -> Response:
        """Receive one update a network pushed to us."""
        # `request.body()` is the bytes exactly as they arrived. Never
        # `request.json()` here: a signature is over those exact bytes, and
        # parsing the JSON and building it again changes the spacing and the
        # key order, so the signature no longer matches. That is the single
        # most common reason a correct signature appears to fail.
        body = await request.body()
        return _answer(await routes.webhook(platform, body, dict(request.headers)))

    return api

Flask¶

blueprint ¶

blueprint(
    sc: SocialChimp,
    *,
    redirect_uri: str,
    memory: LoginMemory | None = None,
    scopes: Mapping[str, Sequence[str]] | None = None,
    secrets: Mapping[str, str] | None = None,
    setup_tokens: Mapping[str, str] | None = None,
    deliver: DeliverUpdate | None = None,
    name: str = "socialchimp",
) -> Blueprint

Build the routes for signing in and receiving updates.

Parameters:

Name	Type	Description	Default
`sc`	`SocialChimp`	The client to work through. Keep one for the life of your process, and hand the same one to your own code.	required
`redirect_uri`	`str`	Where networks send people back to. `{platform}` in it is replaced by the network's name.	required
`memory`	`LoginMemory \| None`	Where a half-finished sign-in waits. Left out, one that lives in this process is used - fine to try things out with, wrong in production. See `shared.LoginMemory`.	`None`
`scopes`	`Mapping[str, Sequence[str]] \| None`	Permissions to ask each network for, by network name.	`None`
`secrets`	`Mapping[str, str] \| None`	The secret each network signs its webhooks with, by network name.	`None`
`setup_tokens`	`Mapping[str, str] \| None`	The token each network's setup check quotes back, by network name.	`None`
`deliver`	`DeliverUpdate \| None`	Where a webhook's update goes. `Dispatcher.deliver` fits. Giving `secrets` without it is refused, because it would mean checking a real update and then dropping it.	`None`
`name`	`str`	What to call the blueprint. Change it if you register two.	`'socialchimp'`

Returns:

Type	Description
`Blueprint`	A blueprint to give `app.register_blueprint`.

Source code in src/socialchimp/contrib/flask.py

def blueprint(
    sc: SocialChimp,
    *,
    redirect_uri: str,
    memory: LoginMemory | None = None,
    scopes: Mapping[str, Sequence[str]] | None = None,
    secrets: Mapping[str, str] | None = None,
    setup_tokens: Mapping[str, str] | None = None,
    deliver: DeliverUpdate | None = None,
    name: str = "socialchimp",
) -> Blueprint:
    """Build the routes for signing in and receiving updates.

    Args:
        sc: The client to work through. Keep one for the life of your
            process, and hand the same one to your own code.
        redirect_uri: Where networks send people back to. `{platform}` in it
            is replaced by the network's name.
        memory: Where a half-finished sign-in waits. Left out, one that
            lives in this process is used - fine to try things out with,
            wrong in production. See `shared.LoginMemory`.
        scopes: Permissions to ask each network for, by network name.
        secrets: The secret each network signs its webhooks with, by network
            name.
        setup_tokens: The token each network's setup check quotes back, by
            network name.
        deliver: Where a webhook's update goes. `Dispatcher.deliver` fits.
            Giving `secrets` without it is refused, because it would mean
            checking a real update and then dropping it.
        name: What to call the blueprint. Change it if you register two.

    Returns:
        A blueprint to give `app.register_blueprint`.
    """
    routes = Routes(
        sc,
        redirect_uri=redirect_uri,
        memory=memory,
        scopes=scopes,
        secrets=secrets,
        setup_tokens=setup_tokens,
        deliver=deliver,
    )
    pages = Blueprint(name, __name__)

    @pages.get("/connect/<platform>")
    def connect(platform: str) -> Response:
        """Begin signing someone in to one network."""
        return _answer(run(routes.start(platform, request.args.to_dict())))

    @pages.route("/callback/<platform>", methods=["GET", "POST"])
    def callback(platform: str) -> Response:
        """Carry on after the person comes back from the network."""
        return _answer(run(routes.finish(platform, _values())))

    @pages.post("/choose/<platform>")
    def choose(platform: str) -> Response:
        """Carry on after the person picked which account to use."""
        return _answer(run(routes.choose(platform, _values())))

    @pages.get("/webhooks/<platform>")
    def setup_check(platform: str) -> Response:
        """Answer the check a network makes before it will send anything."""
        return _answer(run(routes.setup_check(platform, request.args.to_dict())))

    @pages.post("/webhooks/<platform>")
    def webhook(platform: str) -> Response:
        """Receive one update a network pushed to us."""
        # `request.get_data()` is the bytes exactly as they arrived. Never
        # `request.get_json()` here: a signature is over those exact bytes,
        # and parsing the JSON and building it again changes the spacing and
        # the key order, so the signature no longer matches. That is the
        # single most common reason a correct signature appears to fail.
        body = request.get_data()
        headers = dict(request.headers.items())
        return _answer(run(routes.webhook(platform, body, headers)))

    return pages

run ¶

run(work: Coroutine[Any, Any, T]) -> T

Run one async call from Flask's thread and wait for the answer.

The routes below use this, and so should your own views - it is the same bridge, using the same loop, so the connections socialchimp pools are shared with the routes rather than thrown away after every call.

@app.post("/posts")
def write():
    account = sc.account(request.form["connection_id"])
    result = run(account.post(Post(text=request.form["text"])))
    return {"id": result.id, "url": result.url}

Parameters:

Name	Type	Description	Default
`work`	`Coroutine[Any, Any, T]`	The call to run.	required

Returns:

Type	Description
`T`	What it answered.

Raises:

Type	Description
`Exception`	Whatever the call raised, raised again here.

Source code in src/socialchimp/contrib/flask.py

def run(work: Coroutine[Any, Any, T]) -> T:
    """Run one async call from Flask's thread and wait for the answer.

    The routes below use this, and so should your own views - it is the same
    bridge, using the same loop, so the connections socialchimp pools are
    shared with the routes rather than thrown away after every call.

        @app.post("/posts")
        def write():
            account = sc.account(request.form["connection_id"])
            result = run(account.post(Post(text=request.form["text"])))
            return {"id": result.id, "url": result.url}

    Args:
        work: The call to run.

    Returns:
        What it answered.

    Raises:
        Exception: Whatever the call raised, raised again here.
    """
    return asyncio.run_coroutine_threadsafe(work, _the_loop()).result()

Shared by all three¶

Nothing here knows what a request object looks like or imports any framework. Each framework's file takes a request apart into plain values, calls something here, and turns the result back into that framework's own response.

Routes ¶

Routes(
    sc: SocialChimp,
    *,
    redirect_uri: str,
    memory: LoginMemory | None = None,
    scopes: Mapping[str, Sequence[str]] | None = None,
    secrets: Mapping[str, str] | None = None,
    setup_tokens: Mapping[str, str] | None = None,
    deliver: DeliverUpdate | None = None,
)

Signing in and receiving a webhook, with no framework in sight.

Each method takes plain values - a network's name, a mapping of query values, the raw bytes of a body - and hands back a Reply. A framework file does the taking apart and the putting back together, and nothing else.

Every method is a wrapper around a SocialChimp method you could call yourself. Anything the caller did wrong, and anything a network said no to, comes back as a Reply with a sensible status, so a route never has to catch those. Two things are raised instead, because both are yours to deal with and neither is the caller's fault:

ConfigError. Something is set up wrong - a secret that was never stored, an app that was never registered. It would be the same mistake on every request, so answering a tidy 500 only buries it in a log. Raised, it stops you in development and shows up as an error in production, which is what a mistake in your own set-up deserves.
Whatever deliver raised - an ExceptionGroup of the handlers that failed, if it is Dispatcher.deliver. See webhook.

Example

routes = Routes(sc, redirect_uri="https://app.example/cb/{platform}") reply = await routes.start("mastodon", {"host": "mastodon.social"})

Say how these routes should behave.

Parameters:

Name	Type	Description	Default
`sc`	`SocialChimp`	The client to work through. Keep one for the life of your process - see `SocialChimp`.	required
`redirect_uri`	`str`	Where networks send people back to. `{platform}` in it is replaced by the network's name, so one address covers all of them. It has to match what each network's developer portal has on file.	required
`memory`	`LoginMemory \| None`	Where a half-finished sign-in waits. Left out, one that lives in this process is used, which is fine to try things out with and wrong in production - see `LoginMemory`.	`None`
`scopes`	`Mapping[str, Sequence[str]] \| None`	Permissions to ask each network for, by network name. Anything not named here uses that platform's own defaults.	`None`
`secrets`	`Mapping[str, str] \| None`	The secret each network signs its webhooks with, by network name. Meta calls this the app secret.	`None`
`setup_tokens`	`Mapping[str, str] \| None`	The token each network's setup check quotes back, by network name. Meta's forms call this the verify token.	`None`
`deliver`	`DeliverUpdate \| None`	Where a webhook's update goes. `Dispatcher.deliver` fits exactly. Leave it out only if these routes sign people in and nothing else: giving `secrets` without it is refused here, because it would mean checking a real update and then dropping it.	`None`

Raises:

Type	Description
`ConfigError`	If there are webhook secrets but no `deliver`.

Source code in src/socialchimp/contrib/shared.py

def __init__(
    self,
    sc: SocialChimp,
    *,
    redirect_uri: str,
    memory: LoginMemory | None = None,
    scopes: Mapping[str, Sequence[str]] | None = None,
    secrets: Mapping[str, str] | None = None,
    setup_tokens: Mapping[str, str] | None = None,
    deliver: DeliverUpdate | None = None,
) -> None:
    """Say how these routes should behave.

    Args:
        sc: The client to work through. Keep one for the life of your
            process - see `SocialChimp`.
        redirect_uri: Where networks send people back to. `{platform}`
            in it is replaced by the network's name, so one address
            covers all of them. It has to match what each network's
            developer portal has on file.
        memory: Where a half-finished sign-in waits. Left out, one that
            lives in this process is used, which is fine to try things
            out with and wrong in production - see `LoginMemory`.
        scopes: Permissions to ask each network for, by network name.
            Anything not named here uses that platform's own defaults.
        secrets: The secret each network signs its webhooks with, by
            network name. Meta calls this the app secret.
        setup_tokens: The token each network's setup check quotes back,
            by network name. Meta's forms call this the verify token.
        deliver: Where a webhook's update goes. `Dispatcher.deliver`
            fits exactly. Leave it out only if these routes sign people
            in and nothing else: giving `secrets` without it is refused
            here, because it would mean checking a real update and then
            dropping it.

    Raises:
        ConfigError: If there are webhook secrets but no `deliver`.
    """
    self._sc = sc
    self._redirect_uri = redirect_uri
    self._memory = memory if memory is not None else InMemoryLoginMemory()
    self._scopes = scopes if scopes is not None else {}
    self._setup_tokens = setup_tokens if setup_tokens is not None else {}
    self._webhooks = _webhooks_from(secrets, deliver)

start `async` ¶

start(platform: str, params: Mapping[str, str]) -> Reply

Begin signing someone in.

Parameters:

Name	Type	Description	Default
`platform`	`str`	Which network, for example `"mastodon"`.	required
`params`	`Mapping[str, str]`	The query values. `state` is yours to choose and comes back to you at the end; one is made up if you leave it out. `host` names the server, for networks that have more than one.	required

Returns:

Type	Description
`Reply`	A redirect to the network for most networks. For a network
`Reply`	signed in to with an app password or a bot token, the fields to
`Reply`	show a person, as JSON.

Source code in src/socialchimp/contrib/shared.py

async def start(self, platform: str, params: Mapping[str, str]) -> Reply:
    """Begin signing someone in.

    Args:
        platform: Which network, for example `"mastodon"`.
        params: The query values. `state` is yours to choose and comes
            back to you at the end; one is made up if you leave it out.
            `host` names the server, for networks that have more than
            one.

    Returns:
        A redirect to the network for most networks. For a network
        signed in to with an app password or a bot token, the fields to
        show a person, as JSON.
    """
    try:
        state = params.get("state") or token_urlsafe(_STATE_BYTES)
        host = params.get("host")
        step = await self._sc.start_login(
            platform,
            redirect_uri=self._redirect_for(platform),
            scopes=self._scopes_for(platform),
            host=host,
            state=state,
        )
        return await self._next(state, {"host": host}, step)
    except ConfigError:
        # Your set-up, not this request. See the class docstring.
        raise
    except SocialChimpError as error:
        return Reply.for_error(error)

finish `async` ¶

finish(platform: str, params: Mapping[str, str]) -> Reply

Carry on after the person comes back from the network.

Parameters:

Name	Type	Description	Default
`platform`	`str`	Which network.	required
`params`	`Mapping[str, str]`	The query values the network sent back, or - for a network that asked for details instead - what the person typed. Either way it has to carry the same `state` the sign-in started with.	required

Returns:

Type	Description
`Reply`	The connected account as JSON, or the accounts to choose
`Reply`	between when the network needs to know which page or channel to
`Reply`	use.

Source code in src/socialchimp/contrib/shared.py

async def finish(self, platform: str, params: Mapping[str, str]) -> Reply:
    """Carry on after the person comes back from the network.

    Args:
        platform: Which network.
        params: The query values the network sent back, or - for a
            network that asked for details instead - what the person
            typed. Either way it has to carry the same `state` the
            sign-in started with.

    Returns:
        The connected account as JSON, or the accounts to choose
        between when the network needs to know which page or channel to
        use.
    """
    state = params.get("state")
    if not state:
        return _needs("state", "The network should have sent it back.")

    kept = await self._memory.look_up(state)
    if kept is None:
        return _unknown_state()

    try:
        step = await self._sc.finish_login(
            platform,
            callback=params,
            redirect_uri=self._redirect_for(platform),
            scopes=self._scopes_for(platform),
            host=kept.get("host"),
            state=state,
            remember=kept.get("remember"),
        )
        return await self._next(state, kept, step)
    except ConfigError:
        # Your set-up, not this request. See the class docstring.
        raise
    except SocialChimpError as error:
        return Reply.for_error(error)

choose `async` ¶

choose(platform: str, params: Mapping[str, str]) -> Reply

Carry on a sign-in after the person picked which account to use.

Parameters:

Name	Type	Description	Default
`platform`	`str`	Which network.	required
`params`	`Mapping[str, str]`	`state` from the sign-in, and `account_id` naming which of the offered accounts they picked.	required

Returns:

Type	Description
`Reply`	The connected account as JSON.

Source code in src/socialchimp/contrib/shared.py

async def choose(self, platform: str, params: Mapping[str, str]) -> Reply:
    """Carry on a sign-in after the person picked which account to use.

    Args:
        platform: Which network.
        params: `state` from the sign-in, and `account_id` naming which
            of the offered accounts they picked.

    Returns:
        The connected account as JSON.
    """
    state = params.get("state")
    if not state:
        return _needs("state", "It is the one from the sign-in.")

    account_id = params.get("account_id")
    if not account_id:
        return _needs("account_id", "It is the id of the account they picked.")

    kept = await self._memory.look_up(state)
    if kept is None:
        return _unknown_state()

    resume_token = kept.get("resume_token")
    if not isinstance(resume_token, str):
        message = (
            "This sign-in did not stop to ask which account to use, so "
            "there is nothing to carry on from. Only call this after a "
            "callback answered with choose_account."
        )
        return Reply.json({"error": message}, status=400)

    try:
        step = await self._sc.choose(
            platform,
            account_id=account_id,
            resume_token=resume_token,
            redirect_uri=self._redirect_for(platform),
            scopes=self._scopes_for(platform),
            host=kept.get("host"),
            state=state,
            remember=kept.get("remember"),
        )
        return await self._next(state, kept, step)
    except ConfigError:
        # Your set-up, not this request. See the class docstring.
        raise
    except SocialChimpError as error:
        return Reply.for_error(error)

webhook `async` ¶

webhook(
    platform: str, body: bytes, headers: Mapping[str, str]
) -> Reply

Receive one request a network pushed to us.

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 parses the JSON and builds it again has already broken it - the spacing and the key order will not match. Read the body, pass it here, and let read_update do the parsing afterwards. This is the single most common reason a correct signature appears to fail.

Parameters:

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

Returns:

Type	Description
`Reply`	200 when the request was signed properly and every update in it
`Reply`	was handed on. 401 when it was not signed properly, with nothing
`Reply`	said about which check failed.

Raises:

Type	Description
`ConfigError`	If these routes are not set up to receive this network's webhooks, or the platform file is wrong about itself. Both are mistakes to fix rather than answers to send.
`Exception`	Whatever `deliver` raised, which for `Dispatcher.deliver` is an `ExceptionGroup` of the handlers that failed. Nothing is answered, so the framework's own 500 goes back - and a 500 is how a network is told to send the update again. Answering 200 for an update nothing handled would tell it never to bother.

Source code in src/socialchimp/contrib/shared.py

async def webhook(
    self,
    platform: str,
    body: bytes,
    headers: Mapping[str, str],
) -> Reply:
    """Receive one request a network pushed to us.

    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
    parses the JSON and builds it again has already broken it - the
    spacing and the key order will not match. Read the body, pass it
    here, and let `read_update` do the parsing afterwards. This is the
    single most common reason a correct signature appears to fail.

    Args:
        platform: Which network.
        body: The request body, untouched.
        headers: The request headers.

    Returns:
        200 when the request was signed properly and every update in it
        was handed on. 401 when it was not signed properly, with nothing
        said about which check failed.

    Raises:
        ConfigError: If these routes are not set up to receive this
            network's webhooks, or the platform file is wrong about
            itself. Both are mistakes to fix rather than answers to send.
        Exception: Whatever `deliver` raised, which for
            `Dispatcher.deliver` is an `ExceptionGroup` of the handlers
            that failed. Nothing is answered, so the framework's own 500
            goes back - and a 500 is how a network is told to send the
            update again. Answering 200 for an update nothing handled
            would tell it never to bother.
    """
    try:
        pusher = self._sc.platform_for(platform)

        if Feature.PUSH_UPDATES not in pusher.features:
            raise NotSupportedError(
                platform=pusher.name,
                what="pushing updates to a URL of yours",
                suggestion=(
                    "Ask it on a timer instead, with socialchimp.events.Poller."
                ),
            )

        if not isinstance(pusher, CanCheckSignature):
            message = (
                f"The {pusher.name} platform says it pushes updates, but "
                f"its class has no check_signature method. That is a "
                f"mistake in the platform file: either add it or take "
                f"PUSH_UPDATES off its list."
            )
            raise ConfigError(message)

        webhooks = self._webhooks
        if webhooks is None:
            # No deliver was given, so there are no secrets either -
            # `Routes` refuses that pairing when it is built. These
            # routes receive no webhooks at all.
            raise ConfigError(_no_webhook_secret(platform))

        secret = webhooks.secrets.get(platform)
        if secret is None:
            raise ConfigError(_no_webhook_secret(platform))

        pusher.check_signature(body, headers, secret=secret)
        updates = updates_in(pusher, body, headers)
    except ConfigError:
        # Your set-up, not this request. See the class docstring.
        raise
    except SocialChimpError as error:
        return Reply.for_error(error)

    # Outside the try on purpose. A handler that failed is not something
    # to turn into a tidy reply: it goes up, the framework answers 500,
    # and the network sends the update again. Anything handed on before
    # the failure is skipped second time round if you gave the dispatcher
    # a `SeenUpdates`, which is what that is for.
    for update in updates:
        await webhooks.deliver(update)

    return Reply.json({"ok": True})

setup_check `async` ¶

setup_check(
    platform: str, params: Mapping[str, str]
) -> Reply

Answer the one-off check a network makes before it will send us anything.

Meta does a GET at the same address with a token you chose and a challenge to echo back. Get it wrong and it says the URL could not be verified, without saying why.

Parameters:

Name	Type	Description	Default
`platform`	`str`	Which network.	required
`params`	`Mapping[str, str]`	The query values from the check.	required

Returns:

Type	Description
`Reply`	The challenge as plain text, or 403 if the token was not ours.

Raises:

Type	Description
`ConfigError`	If no setup token is stored for this network. See the class docstring for why that is raised and not answered.

Source code in src/socialchimp/contrib/shared.py

async def setup_check(self, platform: str, params: Mapping[str, str]) -> Reply:
    """Answer the one-off check a network makes before it will send us anything.

    Meta does a GET at the same address with a token you chose and a
    challenge to echo back. Get it wrong and it says the URL could not
    be verified, without saying why.

    Args:
        platform: Which network.
        params: The query values from the check.

    Returns:
        The challenge as plain text, or 403 if the token was not ours.

    Raises:
        ConfigError: If no setup token is stored for this network. See
            the class docstring for why that is raised and not answered.
    """
    # Outside the try, because it is the one thing here that is your
    # set-up rather than this request, and it is meant to get out.
    expected = self._setup_tokens.get(platform)
    if expected is None:
        message = (
            f"No setup token is stored for {platform}, so there is "
            f"nothing to check this against. Add it to the "
            f"setup_tokens given to Routes - it is the value you "
            f"typed into that network's dashboard."
        )
        raise ConfigError(message)

    try:
        return Reply.text(answer_setup_check(params, expected_token=expected))
    except SignatureError:
        # 403 rather than the 401 a bad webhook signature gets. Meta's
        # own setup flow expects it, and this is not a signed request -
        # it is a token quoted back at us.
        return Reply.json({"error": "Refused."}, status=403)

Reply `dataclass` ¶

Reply(
    status: int,
    body: bytes,
    content_type: str = "application/json",
    headers: Mapping[str, str] = dict(),
)

What a route decided to answer, before any framework is involved.

Plain bytes and a status, so the same decision can become a FastAPI Response, a Flask one or a Django one without being decided three times.

Attributes:

Name	Type	Description
`status`	`int`	The HTTP status code.
`body`	`bytes`	Exactly what to send, already encoded.
`content_type`	`str`	What to say the body is.
`headers`	`Mapping[str, str]`	Anything else to send, such as where to redirect to.

json `classmethod` ¶

json(
    data: Mapping[str, object], *, status: int = 200
) -> Reply

Answer with a JSON object.

Parameters:

Name	Type	Description	Default
`data`	`Mapping[str, object]`	What to send.	required
`status`	`int`	The status code.	`200`

Returns:

Type	Description
`Reply`	The reply.

Source code in src/socialchimp/contrib/shared.py

@classmethod
def json(cls, data: Mapping[str, object], *, status: int = 200) -> Reply:
    """Answer with a JSON object.

    Args:
        data: What to send.
        status: The status code.

    Returns:
        The reply.
    """
    return cls(status=status, body=json.dumps(data).encode())

text `classmethod` ¶

text(words: str, *, status: int = 200) -> Reply

Answer with plain text.

Parameters:

Name	Type	Description	Default
`words`	`str`	What to send.	required
`status`	`int`	The status code.	`200`

Returns:

Type	Description
`Reply`	The reply.

Source code in src/socialchimp/contrib/shared.py

@classmethod
def text(cls, words: str, *, status: int = 200) -> Reply:
    """Answer with plain text.

    Args:
        words: What to send.
        status: The status code.

    Returns:
        The reply.
    """
    return cls(
        status=status,
        body=words.encode(),
        content_type="text/plain; charset=utf-8",
    )

redirect `classmethod` ¶

redirect(url: str) -> Reply

Send the person's browser somewhere else.

Parameters:

Name	Type	Description	Default
`url`	`str`	Where to send them.	required

Returns:

Type	Description
`Reply`	The reply.

Source code in src/socialchimp/contrib/shared.py

@classmethod
def redirect(cls, url: str) -> Reply:
    """Send the person's browser somewhere else.

    Args:
        url: Where to send them.

    Returns:
        The reply.
    """
    return cls(status=302, body=b"", headers={"Location": url})

for_error `classmethod` ¶

for_error(error: SocialChimpError) -> Reply

Turn one of our errors into an answer.

Parameters:

Name	Type	Description	Default
`error`	`SocialChimpError`	What went wrong.	required

Returns:

Type	Description
`Reply`	The reply, with the status `status_for` chose.

Source code in src/socialchimp/contrib/shared.py

@classmethod
def for_error(cls, error: SocialChimpError) -> Reply:
    """Turn one of our errors into an answer.

    Args:
        error: What went wrong.

    Returns:
        The reply, with the status `status_for` chose.
    """
    if isinstance(error, SignatureError):
        # Every one of these is answered the same way, on purpose. Saying
        # which check failed - missing header, wrong digest, too old -
        # only helps whoever is guessing. See `errors.SignatureError`.
        return cls.json({"error": "Refused."}, status=401)

    headers: dict[str, str] = {}
    if isinstance(error, RateLimitError) and error.retry_after is not None:
        # Rounded up, because a client that waits the rounded-down number
        # of seconds arrives a moment early and is refused again.
        headers["Retry-After"] = str(math.ceil(error.retry_after))

    return cls(
        status=status_for(error),
        body=json.dumps({"error": str(error), "platform": error.platform}).encode(),
        headers=headers,
    )

status_for ¶

status_for(error: SocialChimpError) -> int

Return the status code that fits one of our errors.

Parameters:

Name	Type	Description	Default
`error`	`SocialChimpError`	What went wrong.	required

Returns:

Type	Description
`int`	The status to answer with. Anything we have no particular answer for
`int`	is 500, on the basis that an error we did not plan for is our
`int`	problem and not the caller's.

Source code in src/socialchimp/contrib/shared.py

def status_for(error: SocialChimpError) -> int:
    """Return the status code that fits one of our errors.

    Args:
        error: What went wrong.

    Returns:
        The status to answer with. Anything we have no particular answer for
        is 500, on the basis that an error we did not plan for is our
        problem and not the caller's.
    """
    for kind, status in _STATUSES:
        if isinstance(error, kind):
            return status
    return 500

read_form ¶

read_form(body: bytes) -> dict[str, str]

Read the values out of a form's body.

Used instead of each framework's own form parsing, so that all three behave identically and none of them needs an extra package installed to read an ordinary HTML form.

Parameters:

Name	Type	Description	Default
`body`	`bytes`	The raw body of a form post.	required

Returns:

Type	Description
`dict[str, str]`	The values, by name.

Source code in src/socialchimp/contrib/shared.py

def read_form(body: bytes) -> dict[str, str]:
    """Read the values out of a form's body.

    Used instead of each framework's own form parsing, so that all three
    behave identically and none of them needs an extra package installed to
    read an ordinary HTML form.

    Args:
        body: The raw body of a form post.

    Returns:
        The values, by name.
    """
    return dict(parse_qsl(body.decode()))

LoginMemory ¶

Bases: Protocol

Where a half-finished sign-in waits for the person to come back.

Signing in is two requests. The first one is handed something the second one needs - the secret half of a PKCE pair, which server the person named, and later the resume token from ChooseAccount. socialchimp cannot keep any of that for you: the person can be sent away by one web worker and come back to another, so anything held in one process works on your laptop and fails in production.

Everything is filed under the sign-in's state, which is the one value that makes the round trip through the network.

Back this with whatever your app already has - a session, a Redis key with a short life, a small table. InMemoryLoginMemory is here to try things out with.

keep `async` ¶

keep(state: str, data: RawData) -> None

Write down what the rest of this sign-in will need.

Parameters:

Name	Type	Description	Default
`state`	`str`	The sign-in's state, which is the key.	required
`data`	`RawData`	What to keep. Plain JSON-shaped data.	required

Source code in src/socialchimp/contrib/shared.py

async def keep(self, state: str, data: RawData) -> None:
    """Write down what the rest of this sign-in will need.

    Args:
        state: The sign-in's state, which is the key.
        data: What to keep. Plain JSON-shaped data.
    """
    ...

look_up `async` ¶

look_up(state: str) -> RawData | None

Read back what was kept for one sign-in.

Parameters:

Name	Type	Description	Default
`state`	`str`	The sign-in's state.	required

Returns:

Type	Description
`RawData \| None`	What was kept, or `None` if there is nothing under that state.

Source code in src/socialchimp/contrib/shared.py

async def look_up(self, state: str) -> RawData | None:
    """Read back what was kept for one sign-in.

    Args:
        state: The sign-in's state.

    Returns:
        What was kept, or `None` if there is nothing under that state.
    """
    ...

forget `async` ¶

forget(state: str) -> None

Throw away one sign-in's notes. Quiet if there are none.

Parameters:

Name	Type	Description	Default
`state`	`str`	The sign-in's state.	required

Source code in src/socialchimp/contrib/shared.py

async def forget(self, state: str) -> None:
    """Throw away one sign-in's notes. Quiet if there are none.

    Args:
        state: The sign-in's state.
    """
    ...

InMemoryLoginMemory ¶

InMemoryLoginMemory(max_size: int = _DEFAULT_MEMORY_SIZE)

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

Fine for trying things out and for tests. Not fine in production: two web workers do not share it, so a person sent away by one and returning to another is told their sign-in has expired, and every restart loses every sign-in in flight.

Use your session, or a Redis key, or a small table instead.

What is kept is capped, so that abandoned sign-ins cannot fill up the process. Once it is full the oldest are forgotten first.

Start with nothing remembered.

Parameters:

Name	Type	Description	Default
`max_size`	`int`	How many half-finished sign-ins to hold before forgetting the oldest.	`_DEFAULT_MEMORY_SIZE`

Source code in src/socialchimp/contrib/shared.py

def __init__(self, max_size: int = _DEFAULT_MEMORY_SIZE) -> None:
    """Start with nothing remembered.

    Args:
        max_size: How many half-finished sign-ins to hold before
            forgetting the oldest.
    """
    self._max_size = max_size
    self._kept: OrderedDict[str, RawData] = OrderedDict()

keep `async` ¶

keep(state: str, data: RawData) -> None

Write down what the rest of this sign-in will need.

Parameters:

Name	Type	Description	Default
`state`	`str`	The sign-in's state, which is the key.	required
`data`	`RawData`	What to keep.	required

Source code in src/socialchimp/contrib/shared.py

async def keep(self, state: str, data: RawData) -> None:
    """Write down what the rest of this sign-in will need.

    Args:
        state: The sign-in's state, which is the key.
        data: What to keep.
    """
    self._kept[state] = data
    while len(self._kept) > self._max_size:
        self._kept.popitem(last=False)

look_up `async` ¶

look_up(state: str) -> RawData | None

Read back what was kept for one sign-in.

Parameters:

Name	Type	Description	Default
`state`	`str`	The sign-in's state.	required

Returns:

Type	Description
`RawData \| None`	What was kept, or `None`.

Source code in src/socialchimp/contrib/shared.py

async def look_up(self, state: str) -> RawData | None:
    """Read back what was kept for one sign-in.

    Args:
        state: The sign-in's state.

    Returns:
        What was kept, or `None`.
    """
    return self._kept.get(state)

forget `async` ¶

forget(state: str) -> None

Throw away one sign-in's notes.

Parameters:

Name	Type	Description	Default
`state`	`str`	The sign-in's state.	required

Source code in src/socialchimp/contrib/shared.py

async def forget(self, state: str) -> None:
    """Throw away one sign-in's notes.

    Args:
        state: The sign-in's state.
    """
    self._kept.pop(state, None)

Framework helpers¶

Django¶

urls ¶

get_client cached ¶

orm_storage ¶

Request ¶

View ¶

MakeResponse ¶

MakePath ¶

Exempt ¶

Settings ¶

FastAPI¶

router ¶

Flask¶

blueprint ¶

run ¶

Shared by all three¶

Routes ¶

start async ¶

finish async ¶

choose async ¶

webhook async ¶

setup_check async ¶

Reply dataclass ¶

json classmethod ¶

text classmethod ¶

redirect classmethod ¶

for_error classmethod ¶

status_for ¶

read_form ¶

LoginMemory ¶

keep async ¶

look_up async ¶

forget async ¶

InMemoryLoginMemory ¶

keep async ¶

look_up async ¶

forget async ¶

get_client `cached` ¶

start `async` ¶

finish `async` ¶

choose `async` ¶

webhook `async` ¶

setup_check `async` ¶

Reply `dataclass` ¶

json `classmethod` ¶

text `classmethod` ¶

redirect `classmethod` ¶

for_error `classmethod` ¶

keep `async` ¶

look_up `async` ¶

forget `async` ¶

keep `async` ¶

look_up `async` ¶

forget `async` ¶