0

我正在尝试为 asyncpg 构建一个上下文管理器，但到目前为止，我得到了AttributeError: __enter__，这就是我尝试过的。我试过在def __enter__没有异步的情况下使用，但不起作用

什么是正确的方法？

import asyncpg
import asyncio


class DatabaseConnection(object):

    def __init__(self, host, port, database, user, password):
        self.connection = None
        self.host = host
        self.port = port
        self.database = database
        self.user = user
        self.password = password

    async def __aenter__(self):
        self.connection = await asyncpg.connect(host=self.host,
                                                port=self.port,
                                                database=self.database,
                                                user=self.user,
                                                password=self.password)
        return self.connection

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        # its executed when connection close
        self.connection.close()

然后尝试使用它

async def get_time():
    with DatabaseConnection(host=DB_HOST, port=DB_PORT,
                            database=DB_NAME, user=DB_USER,
                            password=DB_PASSWORD) as connection:
        values = await connection.fetch('''SELECT * CURRENT_TIME''')
        print(values)


loop = asyncio.get_event_loop()
loop.run_until_complete(get_time())
4

0 回答 0