pydux-test/pydux_thunk.py

import asyncio
import inspect


loop = None


def middleware(store):
    dispatch = store['dispatch']
    get_state = store['get_state']

    def wrapper(next_):
        def thunk_dispatch(action):
            if callable(action):
                if inspect.iscoroutinefunction(action):
                    print('found async action')
                    return loop.call_soon_threadsafe(asyncio.ensure_future, action(dispatch, get_state))
                else:
                    print('found sync action')
                    return action(dispatch, get_state)
            return next_(action)

        return thunk_dispatch

    return wrapper