码厩技术博客

#5 GitHub 观察: asyncio

根据 asyncio stars:>1000 的数据自动生成本文。

#4 转载：Python 异步编程与数据库

这是大神 zzzeek 2015 年发表的一篇文章，详细介绍了关于 SQLAlchemy 与异步编程的一些事情。解答了我关于如何实现异步编程的一些疑惑。
我曾反复阅读这篇文章好多遍，以求能够更加准确地领会到大佬阐述的意思。我认为每个 Python 的使用者都应该阅读阅读。

#3 AsyncIO 异步执行命令 [编辑中]

:) 本文正在编辑中，暂时不提供浏览...

#2 RuntimeError: Event loop is closed

开发过程中发现报错：

Traceback (most recent call last):
  ...
  File "/usr/local/python3.7.3/lib/python3.7/asyncio/streams.py", line 353, in close
    return self._transport.close()
  File "/usr/local/python3.7.3/lib/python3.7/asyncio/selector_events.py", line 690, in close
    self._loop.call_soon(self._call_connection_lost, None)
  File "/usr/local/python3.7.3/lib/python3.7/asyncio/base_events.py", line 719, in call_soon
    self._check_closed()
  File "/usr/local/python3.7.3/lib/python3.7/asyncio/base_events.py", line 508, in _check_closed
    raise RuntimeError('Event loop is closed')
RuntimeError: Event loop is closed

经过检查，发现是我在代码中调用了 asyncio.run(协程方法(*args, **kwargs))
总是有些同步代码中需要调用协程方法，没有办法。

改成这样就可以了：

asyncio.get_event_loop().run_until_complete(协程方法(*args, **kwargs))

如果是在线程中就改成：

asyncio.new_event_loop().run_until_complete(协程方法(*args, **kwargs))

否则会报：RuntimeError: There is no current event loop in thread 'Thread-2'.

线程中 run_sync

RuntimeError: This event loop is already running

attached to a different loop

Traceback (most recent call last):
  File "/opt/fuckTheWorld/core/framework.py", line 194, in fuck
    value = asyncio.new_event_loop().run_until_complete(lego.get(key))
  File "/usr/local/python3.7.3/lib/python3.7/asyncio/base_events.py", line 616, in run_until_complete
    return future.result()
  File "/opt/fuckTheWorld/core/lego.py", line 30, in execute
    data = await super().execute(*args, **options)
  File "/opt/fuckTheWorld/lego/client.py", line 476, in execute
    return await conn.retry.call_with_retry(
  File "/opt/fuckTheWorld/lego/retry.py", line 51, in call_with_retry
    return await do()
  File "/opt/fuckTheWorld/lego/client.py", line 455, in _send_command_parse_response
    return await self.parse_response(conn, command_name, **options)
  File "/opt/fuckTheWorld/lego/client.py", line 494, in parse_response
    response = await connection.read_response()
  File "/opt/fuckTheWorld/lego/connection.py", line 934, in read_response
    await self.disconnect()
  File "/opt/fuckTheWorld/lego/connection.py", line 822, in disconnect
    await self._writer.wait_closed()  # type: ignore[union-attr]
  File "/usr/local/python3.7.3/lib/python3.7/asyncio/streams.py", line 359, in wait_closed
    await self._protocol._get_close_waiter(self)
RuntimeError: Task <Task pending name='Task-45' coro=<Fuck.execute() running at /opt/fuckTheWorld/core/lego.py:30> cb=[_run_until_complete_cb() at /usr/local/python3.7.3/lib/python3.7/asyncio/base_events.py:184]> got Future <Future pending> attached to a different loop

#1 Werkzeug WSGI 框架

• PyPI：https://pypi.org/project/Werkzeug/
• 文档：https://werkzeug.palletsprojects.com/en/latest/
• 代码：https://github.com/pallets/werkzeug/

示例

from werkzeug.wrappers import Request, Response
from werkzeug.routing import Map, Rule

class Application:
    def __init__(self):
        self.url_map = Map([
            Rule('/', endpoint='hello')
        ])

    def hello(self, request):
        text = 'Hello, World!'
        response = Response(text, content_type='text/plain')
        return response

    @Request.application
    def __call__(self, request):
        adapter = self.url_map.bind_to_environ(request.environ)
        endpoint, values = adapter.match()

        handler = getattr(self, endpoint)
        return handler(request, **values)

if __name__ == '__main__':
    from werkzeug.serving import run_simple
    app = Application()
    run_simple('localhost', 5000, app)

Werkzeug 与协程（async/await）

Werkzeug 1.x 开始已经提供了对协程的支持。
PS：WSGI 规范是基于同步的设计，没有较好的适配异步编程，所以后面一些更年轻的框架选择了 ASGI（Async Server Gateway Interface）。
PS：2014/03/01, WSGI
PS：2021/11/06, 体验 ASGI

import asyncio
from werkzeug.wrappers import Request, Response
from werkzeug.routing import Map, Rule

class Application:
    def __init__(self):
        self.url_map = Map([
            Rule('/', endpoint='hello')
        ])

    async def hello(self, request):
        text = 'Hello, World!'
        response = Response(text, content_type='text/plain')
        return response

    @Request.application
    async def __call__(self, request):
        adapter = self.url_map.bind_to_environ(request.environ)
        endpoint, values = adapter.match()

        handler = getattr(self, endpoint)
        response = await handler(request, **values)
        return response

if __name__ == '__main__':
    from werkzeug.serving import run_simple
    app = Application()
    run_simple('localhost', 5000, app)

附：版本历史