标签: python-asyncio

是否可以从 Python (3.4)（即在asyncio事件循环的上下文中）对 Microsoft SQL Server 执行异步查询？

以下是一个异步程序框架，其中（异步）SQL 查询应安装到函数中do_it：

import asyncio
import contextlib


@asyncio.coroutine
def do_it():
    # TODO: Make an asynchronous MS SQL query, but how??
    fut = asyncio.Future()
    fut.set_result(None)
    return fut


with contextlib.closing(asyncio.SelectorEventLoop()) as loop:
    asyncio.set_event_loop(loop)
    loop.run_until_complete(do_it())

print('Finished')

我尝试使用 TOR 代理进行抓取，并且在一个线程中一切正常，但这很慢。我尝试做一些简单的事情：

def get_new_ip():
    with Controller.from_port(port = 9051) as controller:
        controller.authenticate(password="password")
        controller.signal(Signal.NEWNYM)
        time.sleep(controller.get_newnym_wait())


def check_ip():
    get_new_ip()
    session = requests.session()
    session.proxies = {'http': 'socks5h://localhost:9050', 'https': 'socks5h://localhost:9050'}
    r = session.get('http://httpbin.org/ip')
    r.text


with Pool(processes=3) as pool:
    for _ in range(9):
        pool.apply_async(check_ip)
    pool.close()
    pool.join()

当我运行它时，我看到输出：

{"origin": "95.179.181.1, 95.179.181.1"}
{"origin": "95.179.181.1, 95.179.181.1"}
{"origin": "95.179.181.1, 95.179.181.1"}
{"origin": "151.80.53.232, 151.80.53.232"}
{"origin": "151.80.53.232, 151.80.53.232"}
{"origin": "151.80.53.232, 151.80.53.232"}
{"origin": "145.239.169.47, 145.239.169.47"}
{"origin": "145.239.169.47, 145.239.169.47"}
{"origin": "145.239.169.47, 145.239.169.47"}

为什么会发生这种情况，我如何为每个线程提供自己的 IP？顺便说一下，我尝试过像 TorRequests、TorCtl 这样的库，结果是一样的。

我知道TOR在发布新IP之前似乎有延迟，但是为什么相同的IP会进入不同的进程？

我正在尝试使用 aiohttp 包编写一些异步 GET 请求，并且已经弄清楚了大部分内容，但是我想知道处理失败（作为异常返回）时的标准方法是什么。

到目前为止我的代码的总体思路（经过一些试验和错误，我遵循这里的方法）：

import asyncio
import aiofiles
import aiohttp
from pathlib import Path

with open('urls.txt', 'r') as f:
    urls = [s.rstrip() for s in f.readlines()]

async def fetch(session, url):
    async with session.get(url) as response:
        if response.status != 200:
            response.raise_for_status()
        data = await response.text()
    # (Omitted: some more URL processing goes on here)
    out_path = Path(f'out/')
    if not out_path.is_dir():
        out_path.mkdir()
    fname = url.split("/")[-1]
    async with aiofiles.open(out_path / f'{fname}.html', 'w+') as f:
        await f.write(data)

async def fetch_all(urls, …

我试图从我的 Flask-socketio 服务器调用的函数

from flask_socketio import emit
import asyncio

async def myfunc():
     for i in range(10):
         j = 1*3
         await emit('update', {'j':j})

在我的服务器功能中，我正在运行

loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
task = asyncio.gather(myfunc())
loop.run_until_complete(task)

我在成功发出的循环的第一次迭代中遇到错误。

File "path\to\Python\Python37-32\Lib\threading.py", line 917, in _bootstrap_inner
    self.run()
  File "path\to\Python\Python37-32\Lib\threading.py", line 865, in run
    self._target(*self._args, **self._kwargs)
  File "path\to\lib\site-packages\socketio\server.py", line 636, in _handle_event_internal
    r = server._trigger_event(data[0], namespace, sid, *data[1:])
  File "path\to\lib\site-packages\socketio\server.py", line 665, in _trigger_event
    return self.handlers[namespace][event](*args)
  File "path\to\lib\site-packages\flask_socketio\__init__.py", line 280, in _handler
    *args)
  File "path\to\lib\site-packages\flask_socketio\__init__.py", line 694, …

我正在寻找在来自 concurrent.futures 的线程池执行程序中运行的异步任务和方法/函数之间通信的最佳解决方案。在以前的同步项目中，我会使用queue.Queue该类。我认为任何方法都应该是线程安全的，因此asyncio.queue不会起作用。

我见过有人扩展queue.Queue课程来做类似的事情：

class async_queue(Queue):
  async def aput(self, item):
    self.put_nowait(item)

  async def aget(self):
    resp = await asyncio.get_event_loop().run_in_executor( None, self.get )
    return resp

有没有更好的办法？

当使用asyncio.create_subprocess_exec一个asyncio.subprocess.process被返回。该文档表明不存在poll或is_alive类型方法。它似乎wait或communicate提供了查看进程是否正在运行的唯一方法，但它们正在阻止调用，并且通信的 asyncio 版本没有超时选项。

有没有一种好方法来检查asyncio子进程是否以非阻塞方式活着？

我能想到的最好的is_alive样式函数是：

import asyncio

async def is_alive(proc):
    try:
        await asyncio.wait_for(proc.wait(), 0.001)
    except asyncio.TimeoutError:
        return True
    else:
        return False

虚拟用例：

async def foo():
    proc = await asyncio.create_subprocess_exec('sleep', '5')
    i = 0
    res = True
    while res:
        res = await is_alive(proc)
        print(f"[{i}] is_alive: {res}")
        # ... do foo stuff while we wait ...
        await asyncio.sleep(1)
        i += …

我希望我的机器人在 on_ready 事件中上线时发送一条消息。该行在 (on_message) 中工作，但我无法使其在 (on_ready) 中发送某些内容

@client.event
async def on_ready():
    print('We have logged in as {0.user}'.format(client))
    await message.channel.send('The bot is online ')

最近一直在阅读关于 python 并发realpython - python concurrency

我的主要关注asyncio点是相当新的。

使用执行异步活动的代码块asyncio并aiohttp在直接运行时运行良好。

但是，当我将代码添加到我的烧瓶蓝图中时，它会引发此错误：

RuntimeError: There is no current event loop in thread 'Thread-2'

出于演示目的，我制作了一个演示烧瓶应用程序。万一有人想测试一下。

主文件

from flask import Flask
from my_blueprint import my_blueprint

#Define flask app
app = Flask(__name__)

#load blueprints
app.register_blueprint(my_blueprint,url_prefix='/demo')

#start flask
if __name__ == '__main__':
    app.run(debug=True)

my_blueprint.py

from flask import Blueprint,request, jsonify,abort,make_response
from flask import make_response
import asyncio
import time
import aiohttp

my_blueprint = Blueprint('my_blueprint', __name__)

@my_blueprint.route('/',methods=['GET'])
def home():
    #code block
    async def download_site(session, …

我尝试编写简单的非阻塞 http 服务器。我无法管理所有例程（Task1、Task2 和服务器）同时运行。不管我做什么服务器块。

import asyncio
from aiohttp import web


async def Task1():
    for i in range(100):
        print ('Task-1',i)
        await asyncio.sleep(1)


async def Task2():
    for i in range(100):
        print ('Task-2',i)
        await asyncio.sleep(2)

async def handle(request):
    name = request.match_info.get('name', "Anonymous")
    text = "Hello, " + name
    return web.Response(text=text)



app = web.Application()
app.add_routes([web.get('/', handle), web.get('/{name}', handle)])


loop=asyncio.new_event_loop()
loop.create_task(Task1())
loop.create_task(Task2())
loop.create_task(web.run_app(app)) #with this line commented task1/2 works
loop.run_forever()

服务器运行时，应在终端上打印预期结果。但是我得到了终端输出或服务器正在运行（从末尾开始评论第二行）但两者都想要。

Task-1 0
Task-2 0
Task-1 1
Task-2 1
Task-1 2
Task-1 3
Task-2 2 …

在有关 Context Vars 的 Python 文档中，描述了 Context::run 方法以启用在上下文中执行可调用对象的更改，以便可调用对象对上下文执行的更改包含在复制的上下文中。如果你需要执行协程怎么办？为了实现相同的行为，你应该怎么做？

就我而言，我想要的是这样的东西来处理可能嵌套事务的事务上下文：

my_ctxvar = ContextVar("my_ctxvar")

async def coro(func, transaction):
    token = my_ctxvar.set(transaction)
    r = await func()
    my_ctxvar.reset(token)  # no real need for this, but why not either
    return r

async def foo():
    ctx = copy_context()
    # simplification to one case here: let's use the current transaction if there is one
    if tx_owner := my_ctxvar not in ctx:
        tx = await create_transaction()
    else:
        tx = my_ctxvar.get()
    
    try:
        r = await ctx.run(coro)  # not …