Асинхронное получение json ответа от сервиса

Question

Пишу бота (aiogram, везде используется async). Для простоты получения/обработки данных с быстрой возможностью редактирования я решил написать web-api. Написал на expressjs (выводит данные из БД), дописал на php легкую веб морду (вносит данные в БД).

В настоящий момент api полностью работает, возвращает данные как и положено.

Появилась необходимость ограничить круг пользователей, после чего в это же API я решил добавить и id telegram аккаунтов в новую таблицу, которые могут запускать бота (довольно простое решение). Но я столкнулся с тем, что ни одну библиотеку (aiohttp/httpx) не могу настроить через async.

Имею следующую структуру:
Класс, через который я хотел обращаться к API (не работает, причину понять не могу)

class access():
    async def getdata(path):
        client = httpx.AsyncClient()
        async with client.stream('GET', path) as response:
            for record in response:
                if record["userid"] == userid:
                    return userid

Пример класса с функцией, где подключаю код выше:

class userlist():
    async def user_all(user_id):
        if user_id in await access.getdata("http://localhost/users"):
            return True
        else:
            return False

Если просто стучусь через браузер на http://localhost/users то получаю все нужные мне id в формате [{"userid":12345678},{"userid":12345679}], Content-Type: application/json; charset=utf-8, т.е. все как положено.

Если что, на питоне пишу не долго, поэтому прошу прощения за возможные ошибки, но литературу/документации штудировал пол дня.

Answer 1

Как верно подметили в комментариях - внутри getdata не существует переменной userid, но думаю что IDE и так подсветила вам это, а код приведен, можно сказать, упрощенный, поэтому думаю вопрос не в этом. В вашем же случае вы пишите await access.getdata, то есть вы вызываете у класса-ссылки метод getdata, хотя должны сначала создать объект, а затем уже вызвать метод getdata, то есть к access нужно дописать еще (). Плюс в методе был пропущен self, если он не нужен - сделайте метод статическим, обернув его декоратором @staticmethod.

На оф. сайте aiohttp есть неплохие примеры, как делать async запросы.

А вот пример, как можно получить json-данные через корутину:

from aiohttp import ClientSession


class Access:
    async def get_data(self, path):
        async with ClientSession() as session:
            async with session.get(path) as response:
                data = await response.json() 
                print(data)  # здесь будет словарь
                # Дальнейшие операции со словарем

Answer 2

Я делаю без классов, потому что в них тут особого смысла нет. Вам лучше не использовать их, пока вы не изучите по ним теорию. В целом, если вам не нужно хранить в объекте класса какие-то данные, общие для объекта класса, достаточно использовать просто функции. Если нужно структурировать - просто разделите код на отдельные файлы (модули), и используйте импорт для доступа к нужной функции.

Также, если у вас не большой блок данных и не поток, а просто JSON, то нет смысла использовать метод stream, достаточно просто делать GET запрос с помощью await client.get(path).

import asyncio
import httpx


async def getdata(path):
    async with httpx.AsyncClient() as client:
        response = await client.get(path)

        # Выбрасываем исключение, если запрос не был успешен
        response.raise_for_status()

        # тут в response.text должен получиться ответ сервера в виде строки '[{"userid":12345678},{"userid":12345679}]'
        # Нужно вытащить оттуда данные в виде списка, делаем это с помощью метода .json()
        data = response.json()

        # Дальше, если в списке только словари вида {"userid":12345678} 
        # (т.е. данных другого вида в нем нет,
        # поэтому не нужны никакие дополнительные проверки),
        # и нужно вытащить только id, то из каждого словаря
        # вытаскиваем значение по ключу "userid",
        # и формируем формируем новый список из этих id
        result = []
        for record in data:
            result.append(record["userid"])
        # Более коротко можно записать как
        # result = [record["userid"] for record in data]
        return result


async def user_all(user_id):
    # Т.к. оператор in возвращает True и False,
    # можно просто вернуть это значение, не оборачивая в if
    return user_id in await getdata("http://localhost:5000/users")  # URL замените на свой


async def main():
    print(await user_all(12345678))


if __name__ == "__main__":
    asyncio.run(main())

Добавляю тестовый сервер на FastAPI:

import fastapi
import uvicorn

app = fastapi.FastAPI()


@app.get("/users")
def users():
    return [{"userid": 12345678}, {"userid": 12345679}]


if __name__ == '__main__':
    uvicorn.run(app, port=5000)

Запускаем, видим что скрипт напечатал True:

Реализация функции getdata через aiohttp, все примерно так же, только session.get() нужно оборачивать в конструкцию async with и делать await при получении json:

import aiohttp


async def getdata(path):
    async with aiohttp.ClientSession() as session:
        async with session.get(path) as response:
            response.raise_for_status()
            data = await response.json()
            return [record["userid"] for record in data]

БЛОГ НА HUSL

Асинхронное получение json ответа от сервиса

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