2

Пишу бота (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, т.е. все как положено.

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

3
  • 1
    Как именно проявляется «не работает», какую-то ошибку пишет или что?
    – andreymal
    1 авг 2022 в 16:43
  • Ну а вообще как минимум очевидно, что внутри функции getdata не существует переменной userid
    – andreymal
    1 авг 2022 в 16:43
  • Обратите внимание на эту тему Что делать с ответами на мой вопрос
    – Dmitry
    5 авг 2022 в 9:12

2 ответа 2

3

Как верно подметили в комментариях - внутри 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)  # здесь будет словарь
                # Дальнейшие операции со словарем

3

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

Также, если у вас не большой блок данных и не поток, а просто 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]

Ваш ответ

By clicking “Отправить ответ”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

Всё ещё ищете ответ? Посмотрите другие вопросы с метками или задайте свой вопрос.