主页 / 问题 / 1583999
Accepted
Handler
Asked:2024-06-15 15:33:17 +0800 CST2024-06-15 15:33:17 +0800 CST

如何在Python类中指定*args

  • 772

我正在使用 Python 中的类，并且需要实现指定不同数据的能力。就我而言，这是一个 dsn 连接或通过参数指定端口、登录名和密码。

所以我无法做出选择；如果未指定 dsn，则将读取其他参数，反之亦然。

class DB:
    def __init__(self, dsn=None, **kwargs):
        self.user = kwargs['user']
        self.password = kwargs['host']
        self.database = kwargs['host']
        self.host = kwargs['host']
        self.connection = None

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

    async def execute(self, query, *args):
        if self.connection is None:
            await self.connect()
        result = await self.connection.fetch(query, *args)
        return result

    async def close(self):
        if self.connection is not None:
            await self.connection.close()
python

1 个回答

  1. Best Answer

    您想要选择一种方便的方式来传递数据以连接到数据库，无论是 DSN 还是单独传递每个参数。

    为此，您需要稍微更正您的代码：

    import asyncpg
import asyncio


class DB:
    # Получаем переданные параметры с помощью метода .get()
    # Не забываем указывать порт.
    def __init__(self, dsn=None, **kwargs): 
        self.dsn = dsn
        self.user = kwargs.get('user')
        self.password = kwargs.get('password')
        self.database = kwargs.get('database')
        self.host = kwargs.get('host')
        self.port = kwargs.get('port')
        self.connection = None


    async def connect(self):
        if self.dsn:  # Проверяем наличие переданного DSN
            self.connection = await asyncpg.connect(dsn=self.dsn)
        else:  # Если его нет, то проверяем наличие всех других параметров
            if not all([self.user, self.password, self.database, self.host, self.port]):
                raise ValueError("Не все данные были переданы.")
            self.connection = await asyncpg.connect(
                user=self.user,
                password=self.password,
                database=self.database,
                host=self.host,
                port=self.port
            )

    async def execute(self, query, *args):
        if self.connection is None:
            await self.connect()
        result = await self.connection.fetch(query, *args)
        return result

    async def close(self):
        if self.connection is not None:
            await self.connection.close()


async def main():
    # db = DB(dsn='postgresql://postgres:[email protected]:5432/dbname')
    db = DB(user='postgres', password='password12345', database='dbname', host='host.com', port='5432')

    result = await db.execute('SELECT * FROM table_name;')

    for i in result:
        print(i)

    await db.close()


if __name__ == '__main__':
    asyncio.run(main())
    • 2

相关问题