如何在Faust代理中使用异步连接池？

Question

我想让Faust代理写入PostgreSQL表。我想使用asyncpg连接池，但找不到一种干净的方法将其注入到应用程序初始化代码中。

Answer 1

只需将以下功能添加到您的faust应用程序中

class KafkaWorker(faust.App):
    def __init__(self, *args: List, **kwargs: Dict) -> None:
        self.broker : str = kwargs.pop('broker')
        self._db_pool = None

        super().__init__(*args, broker=KafkaWorker._broker_faust_string(self.broker), **kwargs)

    async def db_pool(self) -> Pool:
        ''' '''
        if not self._db_pool:
            logging.warning('kafka.db_pool initialization...')
            self._db_pool = await db.db_pool()
            logging.warning('kafka.db_pool initialization...done ✓')

        return self._db_pool

db.db_pool在哪里

from os import environ

import asyncpg
from asyncpg.pool import Pool
from sqlalchemy.orm import scoped_session, sessionmaker
from sqlalchemy.ext.declarative import declarative_base


session = scoped_session(sessionmaker())
Base = declarative_base()


async def db_pool() -> Pool:
    return await asyncpg.create_pool(
        dsn=environ.get('DB_CNX_STRING', 'postgresql://postgres:postgres@postgres:5432/actions')
    )

然后你可以像这样访问它

@kafka.agent(actions_topic)
async def store_actions(actions: StreamT):
    async for action in actions:
        db_pool = await current_agent().app.db_pool()
        async with db_pool.acquire() as conn:
            try:
                yield await save_action(conn, action.to_representation())
            except StoreException:
                logger.exception(f'Error while inserting action in DB, continuing....')
            finally:
                yield action.id