标签: aiohttp

我正在使用 peewee ORM 和 sanic(sanic-crud) 作为应用程序服务器构建 CRUD REST API。一切正常。我为此编写了几个单元测试用例。

但是，我在运行单元测试时遇到了问题。问题是 unittests 启动了 sanic 应用程序服务器并停滞在那里。它根本没有运行单元测试用例。但是，当我手动按 Ctrl+C 时，sanic 服务器将终止并开始执行单元测试。因此，这意味着应该有一种方法可以启动 sanic 服务器并继续运行单元测试并在最后终止服务器。

有人可以请我为sanic应用程序编写单元测试用例的正确方法吗？

我也遵循了官方文档，但没有运气。 http://sanic.readthedocs.io/en/latest/sanic/testing.html

我试过

from restapi import app # the execution stalled here i guess
import unittest
import asyncio
import aiohttp

class AutoRestTests(unittest.TestCase):
    ''' Unit testcases for REST APIs '''

    def setUp(self):
        self.loop = asyncio.new_event_loop()
        asyncio.set_event_loop(None)

    def test_get_metrics_all(self):
        @asyncio.coroutine
        def get_all():
            res = app.test_client.get('/metrics')
            assert res.status == 201
        self.loop.run_until_complete(get_all())

来自restapi.py

app = Sanic(__name__)
generate_crud(app, [Metrics, ...])
app.run(host='0.0.0.0', port=1337, workers=4, debug=True)

我尝试每 X 秒访问一个具有并行和单独会话的网站，然后分析响应中的内容以查看每个会话是否应该继续。但是，一旦代码到达第二个循环，它就会失败。

import asyncio
from aiohttp import ClientSession
import logging
import time

interval = 30
instances = 2
visit_url = 'http://www.example.org'

tasks = []

logging.basicConfig(
    format='%(asctime)s.%(msecs)03d %(message)s',  # Log in format time.milliseconds {message}
    level=logging.INFO,  # Use with logging.info()
    datefmt='%H:%M:%S')  # Display time as Hours:Minutes:Seconds


class StopException(Exception):
    pass


async def quit_app(session, task_, reason):
    logging.info("[{}] {}.".format(task_, reason))
    session.cookies.clear()  # Reset cookies
    session.headers.clear()  # Reset headers
    session.close()  # End HTTP connection
    raise StopException


async def get_status(response):
    if "abow" in response:
        return "success" …

使用Python 3.6和asyncio和aiohttp我写了一个简单的异步程序:

from aiohttp import ClientSession
import asyncio, ssl, time

base_url = 'https://my-base-url.com/api'

async def fetch(session, id):
    query_params = {'qp1':'v1','qp2':'v2', 'id': id}
    async with session.get(base_url, params=query_params, ssl=ssl.SSLContext()) as response:
        res_json = await response.json()
        if response.status == 200:
            time.sleep(2)
            min_rating = res_json.get('minRating')
            max_rating = res_json.get('maxRating')
            print("id = %s, min = %s, max = %s" % (id, min_rating, max_rating))


async def run(ids):
    tasks = []
    async with ClientSession() as session:
        for id in ids:
            task = asyncio.ensure_future(fetch(session, id)) …

我对使用 asyncio/aiohttp 很陌生，但我有一个 Python 脚本，它从 Postgres 表中读取一批 URL:s，下载 URL:s，在每次下载时运行处理函数（与问题无关） ，并将处理结果存回到表中。

简化形式如下所示：

import asyncio
import psycopg2
from aiohttp import ClientSession, TCPConnector

BATCH_SIZE = 100

def _get_pgconn():
    return psycopg2.connect()

def db_conn(func):
    def _db_conn(*args, **kwargs):
        with _get_pgconn() as conn:
            with conn.cursor() as cur:
                return func(cur, *args, **kwargs)
            conn.commit()
    return _db_conn

async def run():
    async with ClientSession(connector=TCPConnector(ssl=False, limit=100)) as session:
        while True:
            count = await run_batch(session)
            if count == 0:
                break

async def run_batch(session):
    tasks = []
    for url in get_batch():
        task = asyncio.ensure_future(process_url(url, …

默认情况下,asyncio同步运行协同程序.如果它们包含阻塞IO代码,它们仍会等待它返回.解决这个问题的方法是loop.run_in_executor()将代码转换为线程.如果线程在IO上阻塞,则另一个线程可以开始执行.所以你不要浪费时间等待IO调用.

如果您在asyncio没有执行者的情况下使用,那么您将失去这些加速.所以我想知道,为什么你必须明确地使用执行器.为什么不默认启用它们？(在下文中,我将重点关注http请求.但它们实际上只是作为一个例子.我对一般原则感兴趣.)

经过一番搜索,我找到了aiohttp.它是一个基本上提供asyncio和组合的库requests:非阻塞HTTP调用.使用执行程序,asyncio并且requests表现得非常像aiohttp.是否有理由实施新库,您是否因使用执行程序而支付性能损失？

回答了这个问题:为什么asyncio总是不使用执行程序？ Mikhail Gerasimov向我解释过,执行程序会启动操作系统线程并且它们会变得昂贵.因此,不将它们作为默认行为是有道理的.aiohttp比requests在执行程序中使用模块更好,因为它提供只有协同程序的非阻塞代码.

这让我想到了这个问题.aiohttp将自己宣传为:

用于asyncio和Python的异步HTTP客户端/服务器.

所以aiohttp是基于asyncio？为什么不asyncio提供只有协同程序的非阻塞代码呢？这将是理想的默认值.

或者aiohttp实现了这个新的事件循环(没有OS线程)本身？在那种情况下,我不明白他们为什么要宣传自己为基础asyncio.Async/await是一种语言功能.Asyncio是一个事件循环.如果aiohttp有自己的事件循环,那么应该有很少的交集asyncio.实际上,我认为这样的事件循环将比http请求更大.

我有一个将在其中包含aiohttp.ClientSession对象的类。

通常当您使用

async with aiohttp.ClientSession() as session:  
   # some code

由于调用了会话的__aexit__方法，因此该会话将关闭。

我无法使用上下文管理器，因为我想在对象的整个生命周期中保持会话的持久性。

这有效：

import asyncio
import aiohttp

class MyAPI:
    def __init__(self):
        self.session = aiohttp.ClientSession()

    def __del__(self):
        # Close connection when this object is destroyed
        print('In __del__ now')
        asyncio.shield(self.session.__aexit__(None, None, None))



async def main():
    api = MyAPI()

asyncio.run(main())

但是，如果在某些地方引发了异常，则在完成__aexit__方法之前将关闭事件循环。我该如何克服？

堆栈跟踪：

Traceback (most recent call last):
  File "/home/ron/.PyCharm2018.3/config/scratches/async.py", line 19, in <module>
    asyncio.run(main())
  File "/usr/local/lib/python3.7/asyncio/runners.py", line 43, in run
    return loop.run_until_complete(main)
  File "/usr/local/lib/python3.7/asyncio/base_events.py", line 568, in run_until_complete
    return future.result()
  File "/home/ron/.PyCharm2018.3/config/scratches/async.py", line …

当使用编写异步爬虫asyncio和aiohttpPython中，我一直有一个疑问：为什么你必须使用async with，并可以很容易地报告错误，如果你不使用它们。

虽然aiohttp也有方法request，但是可以支持调用更简单的api。我想知道有什么区别。还是requests挺喜欢模块的，不知道能不能像requests模块一样简单使用。

我想编写一个简单的脚本来检查网站是否已启动。如果不是，我想使用 Python 的 aiohttp 模块捕获 http 返回错误代码。在下面的示例中，我传入了一个假网站“ http://www.googlesr2332.com ”而不是返回 http 错误，我得到以下信息：

Traceback (most recent call last):
  File "/home/runner/.local/share/virtualenvs/python3/lib/python3.7/site-packages/aiohttp/connector.py", l
ine 967, in _create_direct_connection    traces=traces), loop=self._loop)
  File "/home/runner/.local/share/virtualenvs/python3/lib/python3.7/site-packages/aiohttp/connector.py", l
ine 830, in _resolve_host
    self._resolver.resolve(host, port, family=self._family)  File "/home/runner/.local/share/virtualenvs/python3/lib/python3.7/site-packages/aiohttp/resolver.py", li
ne 30, in resolve
    host, port, type=socket.SOCK_STREAM, family=family)
  File "/usr/local/lib/python3.7/asyncio/base_events.py", line 784, in getaddrinfo
    None, getaddr_func, host, port, family, type, proto, flags)
  File "/usr/local/lib/python3.7/concurrent/futures/thread.py", line 57, in run
    result = self.fn(*self.args, **self.kwargs)
  File "/usr/local/lib/python3.7/socket.py", line 748, in getaddrinfo
    for res in _socket.getaddrinfo(host, …

在 python aiohttp 中，我们可以在 inClientSession或 a (例如) 中设置超时session.get。https://docs.aiohttp.org/en/stable/client_quickstart.html

假设我们这样做

async with aiohttp.ClientSession(timeout=<customized timeout>) as session: 
    async with session.get(<url1>): 
        xxx
    async with session.get(<url2>): 
        xxx

该customized timeout是整个async with aiohttp.ClientSession()或每个async with session.get？

我正在使用aiohttp、asyncio和codetiming来发出并发请求。我最近将 Python 升级到 3.9.0 版，但出现 RuntimeError: 程序运行后事件循环已关闭。我正在尝试使用队列数据结构发出异步请求...

import asyncio
import aiohttp
from codetiming import Timer
async def task(name, work_queue):
    timer = Timer(text=f"Task {name} elapsed time: {{:.1f}}")
    async with aiohttp.ClientSession() as session:
        while not work_queue.empty():
            url = await work_queue.get()
            print(f"Task {name} getting URL: {url}")
            timer.start()
            async with session.get(url) as response:
                await response.text()
            timer.stop()
async def main():
    # Create the queue of work
    work_queue = asyncio.Queue()
    # Put some work in the queue
    for url in …