Advertisement
Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- async def _get_rfids(self, query, *args) -> Sequence[RFID]:
- async with self._pg.cursor() as cursor:
- await cursor.execute(query, args) # выполняем запрос (postgres код из query, который нам передется)
- result = await cursor.fetchall() # получаем результат от базы данных (в нашем случае методы на лайне 4 и 5 объеденяются в одну строку)
- # То есть в пиви нужно просто вызвать метод и он что-то вернет
- # а в случае прямой работы нужно сделать запрос, а потом ручками запрос на получение данных
- if not result: # Если результат пустой -- вернем ничего
- return tuple()
- def get_rfid(res): # Этот метод просто распаршивает данные в словарь rfid_data
- fields = map(attrgetter('name'), cursor.description) # cursor.description['name'] -- название колоннок в таблице
- rfid_data = dict(zip(fields, res)) # -- теперь мы создаем dict, в котором ключ -- название колонки в таблице БД,
- # а значение -- значение взятое из БД (для текущего res)
- coalesce_int(rfid_data, 'external_apt_id') # -- это какая-то хуйня, которая делает что-то с аргументом (приводит в нормальный вид мб)
- coalesce_int(rfid_data, 'external_rfid_id') # -- аналогично предидущей строке
- return RFID(**rfid_data) # -- вернуть объект класса RFID, инициализируя его из полученного dict rfid_data
- return tuple(map(get_rfid, result)) # вот тут мы применяем метод get_rfid к каждому объекту tuple result
Advertisement
Add Comment
Please, Sign In to add comment
Advertisement