标签: python-asyncio

这是我尝试过的：

>>> import asyncio
>>> @asyncio.coroutine
... def f():
...   print("RUN")
...   raise Exception("ciao")
... 
>>> asyncio.ensure_future(f())
<Task pending coro=<coro() running at /usr/lib/python3.4/asyncio/coroutines.py:139>>
>>> loop = asyncio.get_event_loop()
>>> loop.run_forever()
RUN

并且不检索堆栈跟踪。如果我运行协程asyncio.run_until_complete(f())就没有问题。

我最近发布了一个有关如何仅允许从VPC内的EC2实例调用AWS Lambda函数的问题。通过将带有“ AWS lambda角色”策略的IAM角色附加到EC2实例，我设法使其正常工作，现在我可以使用boto3调用lambda函数。

现在，我想使用asyncio await语法异步调用lambda函数。我读到lambda函数通过设置提供了一个异步版本InvokeType='Event'，但是实际上使调用立即返回而没有得到函数的结果。

由于函数需要一些时间，因此我想并行启动许多函数，因此我想避免在等待函数返回时阻塞执行。

我尝试使用aiobotocore，但这仅支持基本的“ s3”服务功能。

解决这个问题的最佳方法（谦虚）是使用AWS API Gateway服务通过GET / POST请求调用lambda函数，该请求可以使用aiohttp轻松处理。

但是，我无法使其正常运行。

我在EC2 IAM角色中添加了策略“ AmazonAPIGatewayInvokeFullAccess”，但每次尝试执行以下操作：

import requests
r = requests.get('https://url_to_api_gateway_for_function')

我得到禁止的回应<Response [403]>。

我直接使用lambda函数中的触发器创建了API网关。

我还尝试编辑API网关设置，方法是在函数路径中添加post方法并设置“ AWS_IAM”身份验证，然后将其部署为“ prod”部署。仍然是相同的禁止响应。当我通过“ API网关上的测试屏幕”对其进行测试时，它可以正常工作。

任何想法如何解决这个问题？我错过了一步吗？

我正在使用 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)

我想将async/await语法与 Twisted Deferred.addCallback方法一起使用。但正如文档中所述，addCallback回调是同步调用的。我见过用于此目的的inlineCallbacks装饰器，但我更喜欢使用async/await语法（如果它甚至可能或有意义的话）。

我从pika 文档中获取了原始代码，但我没有成功尝试将其迁移到 async/await 语法：

import pika
from pika import exceptions
from pika.adapters import twisted_connection
from twisted.internet import defer, reactor, protocol, task


async def run_async(connection):
    channel = await connection.channel()
    exchange = await channel.exchange_declare(exchange='topic_link',type='topic')
    queue = await channel.queue_declare(queue='hello', auto_delete=False, exclusive=False)
    await channel.queue_bind(exchange='topic_link', queue='hello', routing_key='hello.world')
    await channel.basic_qos(prefetch_count=1)
    queue_object, consumer_tag = await channel.basic_consume(queue='hello', no_ack=False)
    l = task.LoopingCall(read_async, queue_object)
    l.start(0.01)


async def read_async(queue_object):
    ch,method,properties,body = await queue_object.get()
    if body:
        print(body)
    await …

我究竟做错了什么？当我跑这个,

import asyncio
def oneSecond():
    await asyncio.sleep(1)

我明白了:

File "<string>", line 3
await asyncio.sleep(1)
            ^
SyntaxError: invalid syntax

使用Python 3.3

问题是我的解决方案是否是一种使用来自另一个线程的数据更新 Tkinter-GUI 的保存和 pythonic 方式？是Lock必需的吗？或者怎么能在Queue这里提供帮助？此示例运行良好，但原始应用程序需要处理复杂得多的数据。

请专注于AsyncioThread.create_dummy_data()最小的工作示例。该示例有两个线程。一个运行Tkinter -mainloop，第二个线程运行asyncio -loop。异步循环模拟获取一些数据并tkinter.Label用这些数据刷新一些数据。

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# restrict to Python3.5 or higher because of asyncio syntax

# based on </sf/answers/3354408991/>

from tkinter import *
import asyncio
import threading
import random


class AsyncioThread(threading.Thread):
    def __init__(self, asyncio_loop, theWindow):
        self.asyncio_loop = asyncio_loop
        self.theWindow = theWindow
        self.maxData = len(theWindow.varData)
        threading.Thread.__init__(self)


    def run(self):
        self.asyncio_loop.run_until_complete(self.do_data())


    async def do_data(self):
        """ Creating and starting 'maxData' asyncio-tasks. """ …

我正在尝试使用子进程中的wget异步下载Python中的文件.我的代码看起来像这样:

async def download(url, filename):
    wget = await asyncio.create_subprocess_exec(
        'wget', url,
        'O', filename
    )
    await wget.wait()


def main(url):
    loop = asyncio.get_event_loop()
    future = asyncio.ensure_future(download(url, 'test.zip'), loop=loop)
    print("Downloading..")
    time.sleep(15)
    print("Still downloading...")
    loop.run_until_complete(future)
    loop.close()

我正在尝试做的是见证打印"Downloading .."然后15秒后"仍在下载...",所有这些都是在文件下载开始的时候.我实际看到的是文件的下载仅在代码命中loop.run_until_complete(future)时开始

我的理解是asyncio.ensure_future应该开始运行download协程的代码,但显然我错过了一些东西.

我正在尝试学习 python 异步模块，我在互联网上到处搜索，包括 youtube pycon 和各种其他视频，但我找不到一种方法从一个异步函数（永远运行）获取变量并将变量传递给其他异步函数（永远运行）

演示代码：

async def one():
    while True:
        ltp += random.uniform(-1, 1)
        return ltp

async def printer(ltp):
    while True:
        print(ltp)

所以我有一个图像 url 列表，我想用请求库迭代它们并将所有图像下载到一个目录中。

def get_image(url, image_name):
    path = pathlib.Path('/path/to/some/directory')
    response = requests.get(url, stream=True)
    with open('{}/{}.png'.format(path, image_name), 'wb') as file:
        for block in response.iter_content(1024):
            file.write(block)

for url in urls:
    get_image(url, image_name)

现在，有没有办法创建一个装饰器来使一个函数成为一个回调，以便在为特定的异步请求返回响应时运行？

以下代码输出如下：

1 sec delay, print "1", 
1 sec delay, print "2", 
1 sec delay, print "1", 
1 sec delay, print "2"

如何将其修改为像这样运行：

1 sec delay, print "1", print "1",
1 sec delay, print "2", print "2"

我希望它能够运行，以便for循环的两个实例开始同时执行。在执行每个实例时，它们将同时遇到first（）函数，然后同时遇到second（）函数，因此按上述顺序打印。

码：

import asyncio

async def first():
    await asyncio.sleep(1)
    return "1"

async def second():
    await asyncio.sleep(1)
    return "2"

async def main():     
    for i in range(2):
      result = await first()
      print(result)
      result2 = await second()
      print(result2)


loop = asyncio.get_event_loop()
loop.run_until_complete(main())