Write a wrapper to REST API with asyncio

Question

I want understand how to correctly structure a functional asyncio-based program.

The code below wraps two external APIs to provide the client a simple send_weather() function. Its structure reflects my current understanding of ansynchronous programming and the very little I've grasped of functional programming.

import asyncio
from functools import partial

import aiohttp


SUCCESSFUL_PUSH = {'deleted': True}  # Value specific to fake push service


async def _post(headers, url, data):
    async with aiohttp.ClientSession(headers=headers) as session:
        async with session.post(url, json=data, verify_ssl=False) as response:
            return await response.json()


async def _get(headers, url):
    async with aiohttp.ClientSession(headers=headers) as session:
        async with session.get(url, verify_ssl=False) as response:
            return await response.json()


def make_comms(headers):
    get = partial(_get, headers)
    post = partial(_post, headers)
    return get, post


async def _notify(post, message, user):
    url = f"https://api.keen.io/dev/null?user={user}"  # Fake push service
    return await post(url, message)


async def _find_city(get, city):
    url = f"https://www.metaweather.com/api/location/search?query={city}"
    obj_list = await get(url)
    if obj_list:
        return obj_list[0]
    return None


async def _city_weather(get, city_id):
    url = f"https://www.metaweather.com/api/location/{city_id}"
    return await get(url)


async def _send_weather(get, post, city, user):
    city_data = await _find_city(get, city)
    if city_data is None:
        print("Could not find city")
        return
    forecast_data = await _city_weather(get, city_data["woeid"])
    weather = forecast_data["consolidated_weather"][0]["weather_state_name"]
    message = f"Current weather in {city}: {weather}"
    reply = await _notify(post, {"message": message}, user)
    if reply == SUCCESSFUL_PUSH:
        print(f"Current weather in {city}: {weather}")
    else:
        print(f"Push notification to {user} failed: {reply}")


def init(get, post):
    return partial(_send_weather, get, post)


if __name__ == "__main__":
    HEADERS = {"Content-Type": "application/json"}
    get, post = make_comms(HEADERS)
    send_weather = init(get, post)
    asyncio.run(send_weather("Glasgow", "Bob"))

How to structure an asynchronous functional program?

Answer 1

• Calling aiohttp.ClientSession in _post and _get is really bad. Don't do this.
• There is no main function, this just makes the code harder to convert into a setuptools application.

• All these functools.partial only hinder readability rather than enhance it.

  Just make a class.

• The code isn't very functional. And I can only imagine what a purist would have to say about _send_weather. There are so many side effects in one little, hard to read, function.

• There is no separation of concerns, _find_city and _city_weather are solely around getting a forecast, whilst _notify is sending a notification. You've mangled them into one _send_weather function. Not great.
• Since the code isn't functional, and this is possibly the worst project to learn functional programming on. Let's just make it so that the code is easy to read.

1. Make a WeatherForecast class.

   1. At instantiation, it only takes a session.
      This is so we can remove all those partials with self.get.
   2. Include get into this class, however without the session creation part.
   3. Include find_city as find_cities, and don't mutate the output.
   4. Include city_weather.

   5. Change send_weather to get_weather and delegate calls to find_city and city_weather.

      It is not allowed to print here. All it does is get the weather for the wanted city.

      This is a helper function and so mutates the values to get the two other functions to work and to get the output as we desire. Nothing else.

2. Remove the session creation part from post.

3. Make a function notify_user that notifies the user of city's weather.

   This is as simple as using two f-strings, to build the URL and the message.

4. Change _send_weather so that it takes a session, city, and user.

   To make the function easy to read, we are limited to three things in the function.

   1. Calling another function.
   2. Performing a simple if else to control data flow.
   3. Formatting output to the user - basic f-strings.

Here's how I performed the above. Which is as functional as your code. And you could argue that my functions are more pure than yours, as get_weather only gets weather. notify_user notifies the user. And send_weather only prints and delegates to other impure functions. None of them are pure and can't be because they all have side effects. Which pretty much makes implementing this in 'pure FP' a joke.

import asyncio
from functools import partial

import aiohttp

SUCCESSFUL_PUSH = {'deleted': True}  # Value specific to fake push service


class WeatherForecast:
    def __init__(self, session):
        self.session = session

    async def get(self, headers, url):
        async with self.session.get(url, verify_ssl=False) as response:
            return await response.json()

    async def find_cities(self, city):
        return await self.get(f"https://www.metaweather.com/api/location/search?query={city}")

    async def city_weather(self, city_id):
        return await self.get(f"https://www.metaweather.com/api/location/{city_id}")

    async def get_weather(self, city_name):
        cities = await self.find_city(city_name)
        if not cities:
            return None
        forecast = await self.city_weather(cities[0]["woeid"])
        return forecast["consolidated_weather"][0]["weather_state_name"]


async def post(session, url, data):
    async with session.post(url, json=data, verify_ssl=False) as response:
        return await response.json()


def notify_user(session, city, weather, user):
    return await post(
        session,
        f"https://api.keen.io/dev/null?user={user}",
        {"message": f"Current weather in {city}: {weather}"},
    )


def send_weather(session, city, user):
    weather = await WeatherForecast(session).get_weather(city)
    if weather is None:
        print("Could not get forecast")
    else:
        reply = notify_user(session, city, weather, user)
        if reply == SUCCESSFUL_PUSH:
            print(f"Current weather in {city}: {weather}")
        else:
            print(f"Push notification to {user} failed: {reply}")


async def main():
    headers = {"Content-Type": "application/json"}
    async with aiohttp.ClientSession(headers=headers) as session:
        send_weather(session, "Glasgow", "Bob")


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