我有一个函数，它接受常规和异步函数（不是协程，而是返回协程的函数）。
它在内部使用asyncio.iscoroutinefunction() test来查看它获得了哪种类型的功能。
最近，当我尝试创建部分异步函数时，它崩溃了。
在这个演示中，ptest 不被识别为一个协程函数，即使它返回一个协程，即ptest() 是一个协程。
import asyncio
import functools
async def test(arg): pass
print(asyncio.iscoroutinefunction(test)) # True
ptest = functools.partial(test, None)
print(asyncio.iscoroutinefunction(ptest)) # False!!
print(asyncio.iscoroutine(ptest())) # True
问题原因很清楚，但解决方案却不是。
如何动态创建通过测试的部分异步函数？
或者
如何测试包裹在部分对象中的 func ？
任何一个答案都可以解决问题。
我需要编写两个程序,这些程序将作为父进程及其子进程运行.父进程生成子进程,然后通过连接到子进程stdin和stdout的一对管道进行通信.通信是点对点的,这就是我需要asyncio的原因.一个简单的读/回放循环是行不通的.
我写过父母.没问题,因为asyncio我提供了所需的一切create_subprocess_exec().
但是我不知道如何在孩子中创建类似的流读取器/写入器.我没想到会有任何问题.因为已经创建了管道,并且在子进程启动时可以使用文件描述符0和1.没有连接是打开的,不需要生成任何进程.
这是我不努力的尝试:
import asyncio
import sys
_DEFAULT_LIMIT = 64 * 1024
async def connect_stdin_stdout(limit=_DEFAULT_LIMIT, loop=None):
if loop is None:
loop = asyncio.get_event_loop()
reader = asyncio.StreamReader(limit=limit, loop=loop)
protocol = asyncio.StreamReaderProtocol(reader, loop=loop)
r_transport, _ = await loop.connect_read_pipe(lambda: protocol, sys.stdin)
w_transport, _ = await loop.connect_write_pipe(lambda: protocol, sys.stdout)
writer = asyncio.StreamWriter(w_transport, protocol, reader, loop)
return reader, writer
问题是我有两个运输工具,我应该有一个.该函数失败,因为它尝试将协议的传输设置两次:
await loop.connect_read_pipe(lambda: protocol, sys.stdin)
await loop.connect_write_pipe(lambda: protocol, sys.stdout)
# !!!! assert self._transport is None, 'Transport already set'
我试图将伪协议传递给第一行,但这行也不正确,因为需要两个传输,而不仅仅是一个:
writer = …我已经分类dict并且需要检测它的所有修改.
(我知道我无法检测到存储值的就地修改.没关系.)
我的代码:
def __setitem__(self, key, value):
super().__setitem__(key, value)
self.modified = True
def __delitem__(self, key):
super().__delitem__(key)
self.modified = True
问题是它只适用于简单的分配或删除.它未检测到所做的更改pop(),popitem(),clear()和update().
为什么__setitem__和__delitem__当项目被添加或删除绕过？我是否还必须重新定义所有这些方法(pop等等)？
语言规范允许使用大写的 F“字符串”以及大写的 R“原始”或 B“字节”，但今天我第一次看到它在答案中使用。
不知道为什么我们有大写和小写，因为“应该有一种 - 最好只有一种 - 明显的方法来做到这一点”。
是否有任何规则何时优先使用大写而不是通常的 f"string"、r"raw" 或 b"bytes"？
我犯了这个错误：
key, value = 'K', 999
msg = (
f"key={key}, "
"value={value}" # needs to be prefixed with f as well
)
# key=K, value={value}
并开始想知道 Python 如何处理复杂的文字连接情况。
我们假设一个字符串是 f 字符串（格式化字符串文字），另一个字符串是普通字符串文字，如上例所示。Python 会在编译时连接这两个字符串吗？如果是，结果是什么？
我想问一下asyncio.Condition。我不熟悉这个概念，但我从学生时代就知道并理解锁、信号量和队列。
我找不到很好的解释或典型的用例，只是这个例子。我看了一下源码。核心功能是通过期货的先进先出实现的。每个等待的协程都会添加一个新的未来并等待它。另一个协程可能会调用notify()它从 FIFO 设置一个或多个期货的结果，并唤醒相同数量的等待协程。到目前为止真的很简单。
但是，实现和使用比这更复杂。等待协程必须首先获取与条件关联的锁，以便能够等待（并wait()在等待时释放它）。此外，通知程序必须获得一个锁才能通知（）。这导致with在每个操作之前声明：
async with condition:
# condition operation (wait or notify)
否则会发生RuntimeError。
我不明白拥有这个锁的意义。我们需要用锁保护哪些资源？在 asyncio 中，事件循环中总是只有一个协程在执行，没有线程中已知的“临界区”。
这个锁真的需要（为什么？）还是只是为了与线程代码兼容？
我的第一个想法是为了兼容性，但在这种情况下，为什么他们在保留使用的同时不删除锁？即制作
async with condition:
基本上是可选的无操作。
我最近提升了我的Linux发行版.Python 3.5被Python 3.6取代.
我安装的所有站点包pip3仍然在/usr/lib/python3.5/site-packages目录中,Python现在没有找到它们,因为它看起来.../python3.6/site-packages很明显.
我看到目录内容,我可以再次手动安装它们,但这并不像我这样做的正确方法.我可以将内容移动到新目录,但同样,这在我看来也是错误的.
我该如何正确处理？
我应该pip3 freeze在升级之前准备一份清单吗？
我尝试搜索,但关键字可能过于笼统,并得到了许多无关的答案.
我想通过子类化为现有的等待类添加一个新功能.
让我们从一个非常简单的基类开始,创建在短暂睡眠后异步返回99的对象.子类应该只为结果添加+1.
我找不到super()用于引用基类的正确方法.
import asyncio
class R99:
def __await__(self):
loop = asyncio.get_event_loop()
fut = loop.create_future()
loop.call_later(0.5, fut.set_result, 99)
return fut.__await__()
class R100(R99):
async def add1(self):
v = await R99()
#v = await super().__await__() # <== error
return v + 1
def __await__(self):
return self.add1().__await__()
async def test():
print(await R99())
print(await R100())
asyncio.get_event_loop().run_until_complete(test())
我想将上下文变量用于类似的目的，就像在这个问题和接受的答案中一样：Python 中的上下文变量
这对应f3a()于本例中：
import contextvars
user_id = contextvars.ContextVar("user_id_var")
def test():
user_id.set("SOME-DATA")
f2()
def f2():
f3a()
f3b()
def f3a():
print(user_id.get())
def f3b():
ctx = contextvars.copy_context()
for key, value in ctx.items():
if key.name == 'user_id_var':
print(value)
break
test()
但是该函数需要user_id全局变量来获取值。如果它在不同的模块中，则需要导入它。
我的想法是，如果一个函数知道存在一个上下文并且它知道变量名，那应该就是它所需要的。我写了f3b，但正如你所看到的，我必须搜索所有变量，因为上下文变量不支持按名称查找。实现了按变量查找，但是如果我有变量，我可以直接从中获取值（f3a案例）
恐怕我不明白为什么它是这样设计的。为什么商定的名称不是密钥？如果在某种框架中设置了上下文，然后由应用程序代码使用，那么这两个函数将位于不同的模块中，而没有公共模块全局变量。我能找到的例子对我没有帮助。有人可以解释上下文变量 API 背后的基本原理吗？
我在服务器上添加了一个小的调试工具。它记录从获得的堆栈跟踪traceback.format_stack()
它包含一些不完整的行，如下所示：
File "/home/...../base/loop.py", line 361, in run
self.outputs.fd_list, (), sleep)
这不是很有帮助。
源代码行360和361：
rlist, wlist, unused = select.select(self.inputs.fd_list,
self.outputs.fd_list, (), sleep)
如果只有一行可以作为堆栈跟踪的一部分，那么我想说带有函数名称（此处为select.select）的行360 是正确的，因为堆栈是通过调用函数创建的。
无论如何，我希望打印整个（逻辑）行。或至少某些上下文（例如之前的2行）。那可能吗？我的意思是当然要付出足够的努力。
尝试添加行继续符\，但未成功。
EPILOGUE：基于Jean-FrançoisFabre的回答和他的代码，我将使用此功能：
def print_trace():
for fname, lnum, func, line in traceback.extract_stack()[:-1]:
print('File "{}", line {}, in {}'.format(fname, lnum, func))
try:
with open(fname) as f:
rl = f.readlines()
except OSError:
if line is not None:
print(" " + line + " <===")
continue
first = max(0, lnum-3)
# read 2 lines …