4
\$\begingroup\$

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):
        super().__init__(raise_for_status=True)

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

    @asynctools.attach_session
    async def create_payment(self, session=None, **kwargs):
        params = {
            **kwargs,
            "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:
        try:
            x = await stripe.create_payment(amount="1000" , currency="eur", receipt_email="me@web.com")
            print(x)
        except client_exceptions.ClientResponseError as e:
            print("Error", e)


if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

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):
        try:
            kwargs["params"] = format_params(kwargs["params"])
        except KeyError:
            pass
        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.

\$\endgroup\$

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.