I'm testing the Stripe payment API. I couldn't find a Python wrapper that provides asynchronous requests so I'm writing one. My first goal was to implement the async equivalent of stripe.PaymentIntent.create.

I used aiohttp client for my tests. One implementation detail: the Stripe API can take HTTP requests containing tables and maps. They are respectively encoded as "table[]=1,2" and "map[a]=1&map[b]=2". The aiohttp client doesn't seem to have any rewriting tool for this kind of parameters. I could have passed parameters as key value pairs: params={"table[]": "1,2"} but it seemed a bit ugly to me. Instead I decided to have input parameters of the form params={"table": [1, 2]} and have them automatically translated to their string forms on the fly. The same goes for maps, and {"map": {a:1, b:2}} is automatically translated to "map[a]=1&map[b]=2".

Here is my current implementation:

import asynctools
from aiohttp_params import session_complex_params
from aiohttp import client_exceptions
import uuid

API_KEY = "" # Your Stripe API key
ROOT_URL = "https://api.stripe.com/v1/payment_intents"

class Stripe(asynctools.AbstractSessionContainer):
    def __init__(self):

    async def session_hook(self, session):
        return session_complex_params(session)

    async def create_payment(self, session=None, **kwargs):
        params = {
            "payment_method_types": ["card"],
            "metadata": {
                "order_id": 6735,
        async with session.post(ROOT_URL, params=params, headers=stripe_header()) as response:
            return await response.json()

def stripe_header():
    return {
        "Idempotency-Key": str(uuid.uuid4()),
        "Authorization": f"Bearer {API_KEY}",

async def main():
    async with Stripe() as stripe:
            x = await stripe.create_payment(amount="1000" , currency="eur", receipt_email="me@web.com")
        except client_exceptions.ClientResponseError as e:
            print("Error", e)

if __name__ == "__main__":
    import asyncio

The asynctools module simply allows to turn a class into an asynchronous aiohttp session manager (by inheriting asynctools.AbstractSessionContainer). Inside the class, the session can be retrieved by any method using the decorator @asynctools.attach_session.

It then allows to simply use Stripe as a :

async with Stripe() as stripe:
    pass # Unique aiohttp session used for all requests
# aiohttp session closed

Finally, the code from aiohttp_params is the following:

def session_complex_params(session):
    session.post = patch_params(session.post)
    return session

def patch_params(request_method):

    def decorated_request(*args, **kwargs):
            kwargs["params"] = format_params(kwargs["params"])
        except KeyError:
        return request_method(*args, **kwargs)

    return decorated_request

def format_params(params):
    return '&'.join(format_param(*item) for item in params.items())

def format_param(k, v):
    if isinstance(v, list):
        return format_list_param(k, v)
    if isinstance(v, dict):
        return format_dict_param(k, v)
    return format_simple_param(k, v)

def format_dict_param(key, dic):
    return '&'.join(f"{key}[{k}]={str(v)}" for k, v in dic.items())

def format_list_param(key, li):
    li = map(str, li)
    return f"{key}[]={','.join(li)}"

def format_simple_param(key, value):
    return f"{key}={value}"

Is there anything important missing, or any improvement you think I could add ? Note that I just discovered the Stripe API so there is a good chance I'm not using it properly either. Any comment is welcome.


Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.