我有一个异步API,我用它连接并发送邮件到SMTP服务器,它有一些设置和拆除它.所以它很适合使用contextmanagerPython 3的contextlib.

虽然,我不知道它是否可能写,因为它们都使用生成器语法来编写.

这可能会证明问题(包含yield-base和async-await语法的混合,以演示异步调用和上下文管理器的收益之间的差异).

@contextmanager
async def smtp_connection():
    client = SMTPAsync()
    ...

    try:
        await client.connect(smtp_url, smtp_port)
        await client.starttls()
        await client.login(smtp_username, smtp_password)
        yield client
    finally:
        await client.quit()

目前python中有可能出现这种情况吗？with as如果是,我将如何使用声明？如果没有,我可以用另一种方式实现这一点 - 也许使用旧式上下文管理器？

我希望能够从许多异步协同程序中获益.Asyncio as_completed有点接近我正在寻找的东西(即我希望任何协同程序能够随时返回到调用者然后继续),但这似乎只允许常规协同程序返回一次.

这是我到目前为止所拥有的:

import asyncio


async def test(id_):
    print(f'{id_} sleeping')
    await asyncio.sleep(id_)
    return id_


async def test_gen(id_):
    count = 0
    while True:
        print(f'{id_} sleeping')
        await asyncio.sleep(id_)
        yield id_
        count += 1
        if count > 5:
            return


async def main():
    runs = [test(i) for i in range(3)]

    for i in asyncio.as_completed(runs):
        i = await i
        print(f'{i} yielded')


if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())
    loop.close()

替换runs = [test(i) for i in range(3)]为runs = [test_gen(i) for i …

我试图通过使用模块中的signature()函数,基于我在一些Python函数中提供的类型注释生成一些JavaScript inspect.

当类型是一个简单的内置类时,这部分可以正常工作:

import inspect

def my_function() -> dict:
    pass

signature = inspect.signature(my_function)
signature.return_annotation is dict  # True

虽然我不知道如何打开和检查更复杂的注释,例如:

from typing import List
import inspect

def my_function() -> List[int]:
    pass

signature = inspect.signature(my_function)
signature.return_annotation is List[int]  # False

再次引用自定义类的类似问题:

def my_function() -> List['User']:
    pass
...
signature.return_annotation  # typing.List[_ForwardRef('User')]

我想要的是这样的 - 所以我可以在生成JavaScript时适当地分支:

type = signature.return_annotation... # list
member_type = signature.return_annotation... # int / 'User'

谢谢.

我正在尝试找到一种很好的方法来获取一个2d numpy数组并将列和行名称作为结构化数组附加.例如:

import numpy as np

column_names = ['a', 'b', 'c']
row_names    = ['1', '2', '3']

matrix = np.reshape((1, 2, 3, 4, 5, 6, 7, 8, 9), (3, 3))

# TODO: insert magic here

matrix['3']['a']  # 7

我已经能够使用像这样设置列:

matrix.dtype = [(n, matrix.dtype) for n in column_names]

这让我做,matrix[2]['a']但现在我想重命名行,所以我可以做matrix['3']['a'].

我在尝试实现新的defaultdict代理对象时遇到了一些麻烦.该文档是有点恐慌,所以我不知道如何正确地去了解这一点.

我想defaultdict在Manager实例中添加一个类型列表.你不能Manager.register在股票上使用这个方法,multiprocessing.Manager所以我从中创建了自己的存根管理器multiprocessing.mangers.BaseManager

class Manager(BaseManager):
    pass

然后我创建了我的子类multiprocessing.managers.BaseProxy来容纳defaultdict(我最初尝试使用另一个子类defaultdict,BaseProxy但是它似乎没有用.但这是我现在拥有的:

class ProxyDefaultDict(BaseProxy):
    def __init__(self, default_factory=None, *args, **kwargs):
        self.__data = defaultdict(default_factory)
        super().__init__(*args, **kwargs)

    def _callmethod(self, methodname, args=(), kwds={}):
        return getattr(self.__data, methodname)(*args, **kwds)

    def _getvalue(self):
        return self.__data.copy()

    def __repr__(self):
        return self.__data.__repr__()

    def __str__(self):
        return self.__data.__str__()

Manager.register('defaultdict', ProxyDefaultDict)

最终目标是拥有一个共享字典,可以跨进程和线程安全地共享键控锁.以下是我将如何初始化图像的示例:

if __name__ == '__main__':
    m = Manager()
    d = m.defaultdict(m.Lock)
    with d['named_lock']:
        print('holding the lock')

但是,我遇到了一些问题:

1. BaseManager的子类似乎只能通过上下文管理器进行初始化,即 …

我正在尝试使用Nuitka编译一个简单的独立python应用程序.我只使用标准库就可以做到这一点.

我现在想要引入一些额外的依赖,但我现在收到来自Nuitka的错误,我无法解释.

我试图将numpy导入项目,看起来像:

结构体:

• NPG
  • __init__.py
  • __main__.py
• Pipfile
• Pipfile.lock

这是 __main__.py

import numpy as np


def main():
    print('hello Nuitka')
    print('hello numpy', np.__version__)


if __name__ == '__main__':
    main()

然后我运行python -m nuitka npg/__main__.py --standalone并得到错误:

Nuitka:WARNING:/usr/local/lib/python3.6/site-packages/numpy/testing/_private/pytesttester.py:125: Cannot find 'pytest' in package 'numpy.testing._private' as absolute import (tried pytest).
Nuitka:WARNING:/usr/local/lib/python3.6/site-packages/setuptools/command/egg_info.py:20: Cannot find 'setuptools.extern.six.moves' in package 'setuptools.command' as absolute import (tried setuptools.extern.six.moves).
Nuitka:WARNING:/usr/local/lib/python3.6/site-packages/setuptools/command/egg_info.py:20: Cannot find 'setuptools.extern.six' in package 'setuptools.command' as absolute import (tried setuptools.extern.six).
Nuitka:WARNING:/usr/local/lib/python3.6/site-packages/setuptools/glob.py:13: Cannot find 'setuptools.extern.six' in package 'setuptools' as absolute import …

我一直在玩SwiftUI和Combine，感觉好像有一种方法可以在视图中保留现有的@State属性并创建一个新属性。

例如，我有一个密码创建视图，其中包含用户的密码和passwordConfirm字段。我想采用这两个@State属性并派生一个新的@State，我可以在我的视图中使用它来断言输入是否有效。因此，为了简单起见：不等于空值。

苹果公司的文档说，有一个发行人正在装订，尽管我似乎看不到它。

这是一些无效的伪代码：

import SwiftUI
import Combine

struct CreatePasswordView : View {
    @State var password = ""
    @State var confirmation = ""
    lazy var valid = {
        return self.$password.publisher()
            .combineLatest(self.$confirmation)
            .map { $0 != "" && $0 == $1 }
    }

    var body: some View {
        SecureField($password, placeholder: Text("password"))

        SecureField($confirmation, placeholder: Text("confirm password"))

        NavigationButton(destination: NextView()) { Text("Done") }
            .disabled(!valid)
    }
}

发现任何人。/如果可能的话，采取适当的方法？

更新Beta 2：

从beta 2版本开始可用，因此该代码的前半部分现在可以使用。在View中使用结果发布者的后半部分，我仍然没有弄清楚（disabled(!valid)）。

import SwiftUI
import Combine

struct CreatePasswordView : View { …

我正在使用SQLAlchemy和两个MySQL数据库.其中一个是我的机器本地托管的开发数据库,​​另一个是ClearDB在Heroku上提供的用于生产的MySQL服务器.

我与数据库打开了一个长时间运行的会话,同时它与另一个服务执行同步操作.在我的本地机器上,这很好,但在生产中我得到错误(2013年,"在查询期间丢失了与MySQL服务器的连接").

我已经读过其他帖子,说它可能是请求的大小太大或需要调整的池刷新变量.我不相信事务有效负载相对那么大,并且pool_recycle在调用SQLAlachemy时设置变量create_engine似乎不起作用.

有没有其他人遇到过这个问题,或者能够帮助我缩小这个错误的根本原因 - 这似乎是一个问题,我不知道从哪里开始.

根据评论中的要求,两个系统都返回相同的值select @@interactive_timeout, @@wait_timeout:28800,28800.

谢谢

我正在尝试在 Python 3.7 中合并一堆异步生成器，同时仍然在迭代中添加新的异步生成器。我目前正在使用aiostream合并我的生成器：

from asyncio import sleep, run
from aiostream.stream import merge

async def go():
    yield 0
    await sleep(1)
    yield 50
    await sleep(1)
    yield 100

async def main():
    tasks = merge(go(), go(), go())

    async for v in tasks:
        print(v)

if __name__ == '__main__':
    run(main())

但是，一旦循环开始，我需要能够继续添加到正在运行的任务中。就像是。

from asyncio import sleep, run
from aiostream.stream import merge

async def go():
    yield 0
    await sleep(1)
    yield 50
    await sleep(1)
    yield 100

async def main():
    tasks = merge(go(), go(), go())

    async for …

我正在尝试根据当前日期配置两个对象之间的关系。假设我有一个Person对象和一堆Event对象的关系。如果Event对象上有一个DateTime( start)，我想与今天的所有事件建立关系。

到目前为止，我有：

class Person:
    id = Column(Integer, primary_key=True)
    todays_events = relationship('Event', primaryjoin='and_(Person.id == Event.person_id, cast(Event.start, Date) == "2016-04-23"')

这有效，但我找不到我需要用“2016-04-23”替换日期字符串以获得等效的CURDATE().

有谁知道我在找什么？

谢谢。

我正在编写一个使用Tornado Web tornado.httpclient.AsyncHTTPClient发出请求的库，该库为我的代码提供了以下async接口：

async def my_library_function():
    return await ...

如果用户提供了kwarg，我想使此接口可选地为串行接口，例如：serial=True。尽管显然不能async从普通函数中调用用关键字定义的函数await。这将是理想的-尽管目前在该语言中几乎是不可能的：

async def here_we_go():
    result = await my_library_function()
    result = my_library_function(serial=True)

我无法在网上找到任何东西，有人为此提出了一个不错的解决方案。我不想重新实现基本相同的代码，而不会出现awaits混乱。

这是可以解决的问题，还是需要语言的支持？

解决方案（尽管使用Jesse的-如下所述）

Jesse下面的解决方案几乎是我要使用的解决方案。我最终还是通过使用装饰器获得了我最初想要的界面。像这样：

import asyncio
from functools import wraps


def serializable(f):
    @wraps(f)
    def wrapper(*args, asynchronous=False, **kwargs):
        if asynchronous:
            return f(*args, **kwargs)
        else:
            # Get pythons current execution thread and use that
            loop = asyncio.get_event_loop()
            return loop.run_until_complete(f(*args, **kwargs))
    return wrapper

这为您提供了此界面：

result = await my_library_function(asynchronous=True)
result = my_library_function(asynchronous=False)

我理智地在python的异步邮件列表中检查了这一点，我很幸运地得到Guido的响应，因此他礼貌地将其拒绝了： …

浏览 docopt 文档和示例，我似乎找不到这个功能，但我觉得它应该存在，所以我想我会要求确认一下。

我正在使用 Python 的 docopt 并希望能够允许任意选项。该用例是一个命令行模板实用程序 - 因此任意键值都会很方便。

"""Templator

Usage:
    templator <template> [--arbitrary-option=<value>]...
"""

希望这个例子能说明我所追求的。也许类似的东西--*=<value>是另一种书写方式。

编辑：

这是我当前的解决方案，它采用键值对。然而，它们是用空格分隔的，因此可能很难弄清楚长语句的键和值是什么。

templator <template> (<key> <value>)...

然后在 python 脚本中（对于对此解决方案感兴趣的任何人）

arguments = docopt(__doc_)
arbitrary_kwargs = dict(zip(arguments['<key>'], arguments['<value>']))

key=value如果可能的话，拥有语法将是理想的选择。