我使用的是aiohttp和sqlalchemy,和我创建了一个单身,帮助我,当我需要的SQLAlchemy(代码如下)的实例连接.不幸的是,每隔一段时间我就会收到以下错误(我通过重启服务器"解决"):

12月11日09:35:29 ip-xxx-xxx-xxx-xxx gunicorn [16513]:sqlalchemy.exc.StatementError:(sqlalchemy.exc.InvalidRequestError)在无效事务回滚之前无法重新连接[SQL:'.. .\nFROM ... \nWHERE ... =%(username_1)s \n LIMIT%(param_1)s'] [参数:[{}]]```

有没有办法修复当前的代码？谢谢 :)

CONNECTION_DETAILS = {
    'driver': 'pymysql',
    'dialect': 'mysql',
    'host': os.environ.get('HOST'),
    'port': 3306,
    'user': 'master',
    'password': os.environ.get('PASSWORD'),
    'database': 'ourdb',
    'charset': 'utf8'
}

_instance = None

def __new__(cls, *args, **kwargs):
    if not cls._instance:
        con_str = '{dialect}+{driver}://{user}:{password}@' \
                  '{host}:{port}/{database}?charset={charset}'\
            .format(**cls.CONNECTION_DETAILS)
        try:
            engine = sqlalchemy.create_engine(con_str)

            Session = scoped_session(sessionmaker(bind=engine))
            session = Session()  # Create the ORM handle
        except sqlalchemy.exc.OperationalError:
            logger.exception('Establishing database connection error.')

        cls._instance = super().__new__(cls)
        logger.debug("Returning …

我正在编写一个使用 aiohttp 执行 http 请求的类。根据文档，我不应该为每个请求创建一个 ClientSession，所以我想重用同一个会话。

代码：

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

   # async defs methods from here 

做的时候

TestApi()

我收到错误：Unclosed client session。

持久化ClientSession对象的解决方案是什么？

如何使用aiohttp提供单个静态文件(而不是整个目录)？

静态文件服务似乎与UrlDispatcher.add_static()一起烘焙到路由系统中,但这只服务于整个目录.

(我知道我最终应该使用类似nginx的东西在生产环境中提供静态文件.)

因为Python 3.5引入了async with在推荐的语法文档的aiohttp改变.现在要获得一个网址,他们建议:

import aiohttp
import asyncio

async def fetch(session, url):
    with aiohttp.Timeout(10):
        async with session.get(url) as response:
            return await response.text()

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    with aiohttp.ClientSession(loop=loop) as session:
        html = loop.run_until_complete(
            fetch(session, 'http://python.org'))
        print(html)

如何修改此设置以获取网址集合而不仅仅是一个网址？

在旧asyncio示例中,您将设置一个任务列表,例如

    tasks = [
            fetch(session, 'http://cnn.com'),
            fetch(session, 'http://google.com'),
            fetch(session, 'http://twitter.com')
            ]

我试图将这样的列表与上面的方法结合起来但是失败了.

这个线程中的 aiohttp 服务器示例失败并显示RuntimeError: There is no current event loop in thread 'Thread-1'.错误：

import threading
from aiohttp import web


def aiohttp_server():
    def say_hello(request):
        return web.Response(text='Hello, world')

    app = web.Application(debug=True)
    app.add_routes([web.get('/', say_hello)])
    web.run_app(app)


t = threading.Thread(target=aiohttp_server)
t.start()

如何在线程中运行 aiohttp 服务器？

我试图了解如何保护数据在通过服务器和工作服务器之间的开放网络后被更改

在我的脑海中,我在想它应该遵循以下内容:

|server|---send_job----->|worker|
|      |<--send_results--|      |
|      |                 |      |
|      |-send_kill_req-->|      |

很明显,我不希望有人改变我send_job做一些邪恶的事情,我不希望有人偷看我的结果.

所以我有一个超级简单的aiohttp客户端/服务器设置,我正在尝试实现,ssl但我完全迷失了.

下面是我尝试过的最基本的东西,但我也试过通过以下方式实现我自己的ssl证书:

openssl req -x509 -sha256 -nodes -days 365 -newkey rsa:2048 -keyout domain_srv.key -out domain_srv.crt

以及遵循文档,但我仍然无法做出get任何回应.

我如何正确实现ssl_context以使其工作？!

server.py

from aiohttp import web
import msgpack
import ssl

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)])

ssl_context = ssl.create_default_context(ssl.Purpose.SERVER_AUTH)
web.run_app(app, ssl_context=ssl_context)

client.py …

尝试从URL下载和处理jpeg.我的问题是不适合的网址有些证书验证失败,因为这些网址都是老可能不再是值得信赖的,但是,当我try...except...的SSLCertVerificationError,我仍然得到回溯.

系统:Linux 4.17.14-arch1-1-ARCH,python 3.7.0-3,aiohttp 3.3.2

最小的例子:

import asyncio
import aiohttp
from ssl import SSLCertVerificationError

async def fetch_url(url, client):
    try:
        async with client.get(url) as resp:
            print(resp.status)
            print(await resp.read())
    except SSLCertVerificationError as e:
        print('Error handled')

async def main(urls):
    tasks = []
    async with aiohttp.ClientSession(loop=loop) as client:
        for url in urls:
            task = asyncio.ensure_future(fetch_url(url, client))
            tasks.append(task)
        return await asyncio.gather(*tasks)

loop = asyncio.get_event_loop()
loop.run_until_complete(main(['https://images.photos.com/']))

输出:

SSL handshake failed on verifying the certificate
protocol: <asyncio.sslproto.SSLProtocol object at 0x7ffbecad8ac8>
transport: <_SelectorSocketTransport fd=6 …

我正在运行以下代码,通过aiohttp发出5个请求:

import aiohttp
import asyncio

def fetch_page(url, idx):
    try:
        url = 'http://google.com'
        response = yield from aiohttp.request('GET', url)

        print(response.status)
    except Exception as e:
        print(e)

def main():
    try:
        url = 'http://google.com'
        urls = [url] * 5

        coros = []
        for idx, url in enumerate(urls):
            coros.append(asyncio.Task(fetch_page(url, idx)))

        yield from asyncio.gather(*coros)
    except Exception as e:
        print(e)

if __name__ == '__main__':
    try:
        loop = asyncio.get_event_loop()
        loop.run_until_complete(main())
    except Exception as e:
        print(e)

输出:

200
200
200
200
200
Exception ignored in: Exception ignored in: Exception …

我在Ubuntu 16上使用python 3.5.

我正在尝试使用aiohttp来编写一个简单的客户端.

这是我的代码.我从这里拿走了它.这是第一个代码示例,禁用了ssl检查:

import aiohttp
import asyncio
import async_timeout

async def fetch(session, url):
    with async_timeout.timeout(10):
        async with session.get(url) as response:
            return await response.text()

async def main(loop):
    conn = aiohttp.TCPConnector(verify_ssl=False)
    async with aiohttp.ClientSession(loop=loop, connector=conn) as session:
        html = await fetch(session, 'http://www.google.com')
        print(html)

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

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

对于某些网站,此代码有效.对于其他人,包括http://python.org或http://google.com不起作用.相反,代码会生成此错误:

aiohttp.errors.ClientOSError: [Errno 101] Cannot connect to host google.com:80 ssl:False [Can not connect to google.com:80 [Network is unreachable]]

我尝试了一个简单的requests脚本,如下所示:

import requests …

我正在运行一个一次性的Fargate任务，该任务运行一个小的python脚本。任务定义被配置为用于awslogs将日志发送到Cloudwatch，但是我面临一个非常奇怪的间歇性问题。

日志有时会出现在新创建的Cloudwatch流中，有时却不会。我尝试删除部分代码，而现在，这就是我所拥有的。

当我删除asyncio / aiohttp提取逻辑时，打印语句通常出现在Cloudwatch日志中。虽然由于问题是断断续续的，但我不能100％肯定会一直发生。

但是，由于包含了获取逻辑，有时在Fargate任务退出后，我会得到完全为空的日志流。没有日志显示“作业开始”，“作业结束”或“将文件放入S3”。也没有错误日志。尽管如此，当我检查S3存储桶时，仍创建了具有相应时间戳的文件，表明脚本确实运行完毕。我无法理解这是怎么可能的。

dostuff.py

#!/usr/bin/env python3.6

import asyncio
import datetime
import time

from aiohttp import ClientSession
import boto3


def s3_put(bucket, key, body):
    try:
        print(f"Putting file into {bucket}/{key}")
        client = boto3.client("s3")
        client.put_object(Bucket=bucket,Key=key,Body=body)
    except Exception:
        print(f"Error putting object into S3 Bucket: {bucket}/{key}")
        raise


async def fetch(session, number):
    url = f'https://jsonplaceholder.typicode.com/todos/{number}'
    try:
        async with session.get(url) as response:
            return await response.json()
    except Exception as e:
        print(f"Failed to fetch {url}")
        print(e)
        return None


async def fetch_all():
    tasks = []
    async …