Как указать *args в классе python

Question

работаю с классами в 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()

Answer 1

Вы хотите выбрать удобный способ передачи данных для подключения к базе данных, будь то 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())

БЛОГ НА HUSL

Как указать *args в классе python

Ответы (1 шт):