non async interface

[nbraun-wolf]

Hi,

I have a situation where some bits of code don't support async.
Give the fact that it's using SQL alchemy engine under the hood and I also have already sync support by installing psycopg2 to use it with alembic, I was wondering if this package is exposing some kind of non async interface that could be used when there is unfortunately no other way.

[aminalaee]

If I understand correctly, you want to use databases for sync queries?

As far as I know, that's not possible.

BTW, why do you need that? Sqlalchemy and psycopg2 should be enough for that case.

[nbraun-wolf]

How could I mix async and sync code in a sync callback? I cant use the await keyword in there.

Yes Ideally I would replace the driver, I have looked into way for doing that as well, but there are not good options out there. Paho is very solid and battle proven.

There async wrapper library but it tries to enforce the use of context managers, https://github.com/sbtinstruments/asyncio-mqtt. I didn't manage to really integrate this properly with fastapi. It's a long story.

[aminalaee]

something like this maybe:

def callback():
    loop = asyncio.get_event_loop()
    loop.run_in_executor(None, async_func)

https://docs.python.org/3/library/asyncio-eventloop.html#asyncio.loop.run_in_executor

[nbraun-wolf]

I see, how about doing something like this?

import asyncio
import json

from paho.mqtt.client import Client


def init_client(*topics):
    client = Client("test")
    client.connect_async("localhost")

    @client.connect_callback()
    def on_connect(client, userdata, flags, rc):
        print("Connection returned " + str(rc))
        client.subscribe(*topics)
    
    # get the main event loop (from fastapi)
    loop = asyncio.get_event_loop()

    def topic(sub):
        def decorator(coro):
            @client.topic_callback(sub)
            def handle(client, userdata, msg):
                # run the async callbacks on the main event loop
                asyncio.run_coroutine_threadsafe(coro(client, userdata, msg), loop)

        return decorator

    return client, topic

# async main program (potentially fastapi)
async def main():
    client, topic = init_client("my/topic")

    @topic("my/topic")
    async def print_payload(client, userdata, msg):
        await asyncio.sleep(2)
        print(msg.payload)

    client.loop_start()

    while True:
        print("running...")
        await asyncio.sleep(5)
        client.publish("my/topic", json.dumps({"mqtt": "msg"}))


asyncio.run(main())

[aminalaee]

I just took a quick look, I think it should be fine.

[nbraun-wolf]

Cool, thanks for taking your time and helping me to get on the right track.