我有一个树数据结构,我想通过GraphQL API返回.
结构不是特别大(小到足以在一次调用中返回它不成问题).
未设置结构的最大深度.
我已经将结构建模为:
type Tag{
id: String!
children: [Tag]
}
当人们想要将标签带到任意深度时,就会出现问题.
要让所有孩子(例如)级别3,可以编写如下查询:
{
tags {
id
children {
id
children {
id
}
}
}
}
有没有办法编写查询以将所有标记返回到任意深度？
如果不是建议的方法来建模如上所述的结构,如GraphQL API.
我需要处理包含带有罗马尼亚语字符的文本的 XML 文件（像“licen?e ?im?rci”这样的文本）。
处理后，此文本最终会出现在 html 文档中。如果在最后一步我使用以下内容输出内容：
doc.write(path, encoding='utf-8', method='xml' )
注意：doc上面是一个实例lxml.etree.ElementTree
我得到类似的东西：
<span id="123">licen?e ?i m?rci</span>
一切看起来都很好（即罗马尼亚字符不会转换为 html 实体）。
如果我确实使用 html 编码方法
doc.write(path, encoding='utf-8', method='html' )
我得到类似的东西：
<span id="128927">licenţe şi mărci</span>
虽然文档在浏览器中看起来不错，但文档中的文本也被另一个进程使用，该进程提取文本并将其用于一些完全被 html 实体混淆的差异化作业。
虽然使用 thexml作为方法参数etree.ElementTree.write似乎解决了我当前的问题，但感觉不对，因为我真的想生成html文件并且在其他上下文中可能表现不佳。
有没有办法将文档编写为 html（使用method='html'）而无需 lxml 用 html 实体替换非 ascii 字符？
在@mzjn 的问题之后，事实证明这个故事有点复杂。
这是重现问题的最小代码：
from lxml import etree
from os import path
def process( method, out_name ):
# the doc
x = etree.XML('<span id="123">licen?e …我试图了解如何使用双向流处理 grpc api（使用 Python API）。
假设我有以下简单的服务器定义：
syntax = "proto3";
package simple;
service TestService {
rpc Translate(stream Msg) returns (stream Msg){}
}
message Msg
{
string msg = 1;
}
假设将从客户端发送的消息异步发送（由于用户选择了一些 ui 元素）。
为客户端生成的 python 存根将包含一个方法Translate，该方法将接受生成器函数并返回一个迭代器。
我不清楚的是我将如何编写生成器函数来返回用户创建的消息。在等待消息时睡在线程上听起来不是最好的解决方案。
我想指定(作为类型提示)当前定义的类型作为方法的返回类型.
这是一个例子:
class Action(Enum):
ignore = 0
replace = 1
delete = 2
@classmethod
# I would like something like
# def idToObj(cls, elmId: int)->Action:
# but I cannot specify Action as the return type
# since it would generate the error
# NameError: name 'Action' is not defined
def idToObj(cls, elmId: int):
if not hasattr(cls, '_idToObjDict'):
cls._idToObjDict = {}
for elm in list(cls):
cls._idToObjDict[elm.value] = elm
return cls._idToObjDict[elmId]
理想情况下,我希望指定类似的东西
def idToObj(cls, elmId: int)->Action:
谢谢.
我已将 django.channels 添加到 django 项目中，以支持长时间运行的进程，这些进程通过 websockets 通知用户进度。
除了长时间运行的进程的实现似乎没有异步响应之外，一切似乎都运行良好。
为了进行测试，我创建了一个AsyncConsumer识别“run”和“isBusy”两种类型的消息。
“运行”消息处理程序设置“忙标志”，发回“进程正在运行”消息，异步等待20 秒重置“忙标志”，然后发回“进程完成消息”
“isBusy”消息返回带有忙标志状态的消息。
我的期望是，如果我发送一条运行消息，我将立即收到一条“进程正在运行”消息，20 秒后我将收到一条“进程完成”消息。这按预期工作。
我还希望如果我发送“isBusy”消息，我将立即收到带有标志状态的响应。
观察到的行为如下：
下面是 Channel 监听器的实现：
class BackgroundConsoleConsumer(AsyncConsumer):
def __init__(self, scope):
super().__init__(scope)
self.busy = False
async def run(self, message):
print("run got message", message)
self.busy = True
await self.channel_layer.group_send('consoleChannel',{
"type":"consoleResponse",
"text":"running please wait"
})
await asyncio.sleep(20)
self.busy = False
await self.channel_layer.group_send('consoleChannel',{
"type":"consoleResponse",
"text": "finished running"
})
async def isBusy(self,message):
print('isBusy got message', message)
await self.channel_layer.group_send('consoleChannel',{ …