Universall's questions

任务

我想编写将数据添加到最终响应的中间件。也就是说，这个答案：

{
    "user": ...,
    "some_field": ...
}

它应该变成这样：

{
    "some_additional_field": ...,
    "data": {
        "user": ...,
        "some_field": ...
    }
}

为此我编写了中间件：

class ResponseFormatterMiddleware:
    async def _post_process_response(self, response: Response) -> Response:
        response.body = response.render({
            "some_additional_field": 123,
            "data": response.body
        })

        return response

    async def __call__(self, request: Request, call_next) -> Response:
        response = await call_next(request)
        return await self._post_process_response(response)

并添加它：

app.middleware("http")(ResponseFormatterMiddleware())

但我收到一个错误：

AttributeError: '_StreamingResponse' object has no attribute 'body'

之后我发现我收到的输入不是fastapi.Request, but starlette.middleware.base._CachedRequest，但是执行控制器后我收到的输入不是fastapi.Response, but starlette.middleware.base._StreamingResponse。

问题）

是否可以像 django 中那样做 +- ？也就是说，使用中间件将附加信息写入请求中，然后以某种方式更改最终响应。

是否有可能以某种方式使用对象fastapi而不是starlette中间件？

有一个抽象类：

class Parent(Generic[T]):
    @abstractmethod
    def test(self) -> List[T]:  ...

我需要创建一个抽象子类，它在方法中将test返回一个包含子类的列表V（例如[Cls1, Cls2, Cls3, ...]）：

V = TypeVar("V")


class Child(Parent[V]):
    @abstractmethod
    def test(self) -> List[V]: ...

这样稍后我就可以指定将返回什么类型的子类test：

class IntChild(Child[int]):  # test должен возвращать список подклассов int

我尝试过的：

如果bound你传入它type，它将V被指定为后代类type（即元类），这不适合我

如果Child.test你写in ，那么方法类型与父方法（ B ， a in ）def test(self) -> List[Type[V]]: ...不匹配，这也不好：Child List[Type[V]]Parent List[V]

error: Return type "list[type[V]]" of "test" incompatible with return type "list[V]" in supertype "Parent"  [override]

问题：如何指示方法test不应Child返回类型的对象列表V，而是子类列表V

我有来自服务器的 json 响应作为字符串。我将它分解为一个typejson.Unmarshal的变量。bodymap[string]interface{}

var body map[string]interface{}
if err := json.Unmarshal(bodyByte, &body); err != nil {
    log.Fatal(err)
}

现在我需要从中获取第body["items"]一个元素，但是有一个问题：

Invalid operation: 'body["items"][0]' (type 'interface{}' does not support indexing)

我在 go 中写的不多（主要是 python），所以我有一个问题：How to tell go that bodythere are elements with different types ( map[string]string, []string, stringetc)

最近我对结合 python 和 go 产生了兴趣（例如，加快一些任务）。我找到了几个关于这个主题的网站，但事实证明你需要将它们结合起来c。我寻求帮助：向不知道c在哪里做的人解释ctrl+ c，ctrl+v以便从 python 调用用 go 编写的函数并给出结果。

PS：网站：1、2

我玩socket。我有点想通了正常的版本，现在我想写一个异步的。例如一个简单的回显服务器。有一个代码：

    async def _run(self) -> None:
        self.server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.server_socket.bind((settings.HOST, settings.PORT))
        self.server_socket.listen(8)
        self.server_socket.setblocking(False)

        loop = asyncio.get_event_loop()

        while True:
            client, address = await loop.sock_accept(self.server_socket)
            print(f"Connected {address}")
            loop.create_task(self.handle_client(client, address))

    async def handle_client(self, client_socket: socket.socket, address: tuple) -> None:
        while True:
            data = client_socket.recv(1024).decode(encoding=settings.ENCODING)  # BlockingIOError: [WinError 10035] Операция на незаблокированном сокете не может быть завершена немедленно
            if not data:
                break
            client_socket.send(data.upper().encode(encoding=settings.ENCODING))
            print(data)

它因错误而中断：

BlockingIOError: [WinError 10035] Операция на незаблокированном сокете не может быть завершена немедленно

我大致理解为什么：需要self.server_socket.setblocking(False)实现异步并且客户端还没有读取过去的数据或者根本还没有准备好。client_socket.recv如果你把before time.sleep，那么一切都开始工作，但这显然不是一个解决方案。

我怎样才能异步（以便在那一刻处理其他客户端）等待我可以读/写东西的那一刻？

PS我知道asyncore，但我想从头开始写一切，并从内部了解一切是如何运作的

我有一个任务：按标题查找一堆 youtube 视频（我正在编写一个不和谐的机器人）。为此，我使用youtubesearchpython，但处理 100 首歌曲可能需要 + - 2 分钟，这不适合我，我决定使用 . 将 20 首曲目分配到 5 个进程中multiprocessing。问题是我multiprocessing.Pool没有从主文件（run.py=> my_bot.load_extension('Bot')=> import parser.Parser=> multiprocessing.Pool）运行，因此我运行了 5 个机器人副本。我听说要使用if __name__ == "__main__":，但在我的情况下它不起作用（因为文件不是主文件）。如何在 5 个进程中只运行我需要的功能？

PS这是我的代码：

    async def get_songs(songs: Tuple[str, ...], *, by: str) -> list:
        result = []
        with multiprocessing.Pool(processes=5) as pool:
            step = len(songs) // 5
            args = [{"songs": songs[index: index + step], "by": by} for index in range(0, len(songs), step)]
            for res in pool.map(get_song_process, args):
                result.extend(res)
        return result

PS2：我有一个想法，写一个单独的文件并将任务拆分为其中的进程，但我认为这里有更好的解决方案

我寻求建议/澄清：在哪里使用生成器更好，在哪里使用迭代器。

假设我们有一个迭代器：

class MyIter1:
    def __init__(self, num):
        self.num = num

    def __iter__(self):
        return self

    def __next__(self):
        if self.num > 0:
            self.num -= 1
            return self.num
        raise StopIteration

还有 for 循环：

for i in MyIter1(10):
    print(i)

在循环中，MyIter1调用 y 方法__next__ 并使用它返回的值。

还有一个生成器：

def gen(num):
    for x in range(num):
        if num > 0:
            num -= 1
            yield num

并对其进行相同的 for 循环：

for i in gen(10):
    print(i)

在循环中，gen调用ynext()并使用它返回的值。

以下是问题：

这两种方法有何不同（在我看来它们大致相同），最好在哪里使用它们以及在什么情况下它们会有显着差异（最好有例子）。

PS我对生成器很熟悉，也了解它们的工作原理，它们是用于流式数据处理的，但是我不完全明白有什么区别，因为这也可以用迭代器来完成，只需在它的__next__()时候调用它是必须的。

我正在写一个不和谐的机器人，但我遇到了这个问题：为了处理命令的延迟，discord.ext.commands.CommandOnCooldown我需要一个方法on_command_error，但它不适用于特定的错误，但会同时出现，因此，当另一个（真正的）错误来了，它也以异常on_command_error的形式进入，我不知道是什么原因造成的，因为我只得到文本。有没有办法以某种方式从异常对象中获取有关导致它的行的信息。到目前为止，我的代码如下所示：

@commands.Cog.listener()
async def on_command_error(self, ctx, error: Exception):
    if isinstance(error, commands.CommandOnCooldown):
        await self.error_message(ctx, f'This command is on cooldown, you can use it in {round(error.retry_after, 2)}')
    else:
        print(error)

我有一个需要大量计算的复杂任务。我想优化它并在线程之间分配它。测试空白：

import threading

test = 0


def my_func():
    c = 43243423 ** 543536  # 'сложные вычисления...'
    global test
    test += c
    print(f"{threading.current_thread().name} finished")
    barrier.wait()


barrier = threading.Barrier(6)
for i in range(5):
    threading.Thread(target=my_func).start()

print("created")
barrier.wait()
print(test)

在输出中我得到：

Thread-1 finished
Thread-2 finished
Thread-3 finished
Thread-4 finished
createdThread-5 finished

根据我的想法，所有线程都应该并行执行计算，因此它们应该在大约 1 时刻完成，但这不会发生。如何让它们同时工作？

我有一个不和谐的 py 机器人。是否有可能以某种方式使其可以添加到 2 个不同的服务器并异步工作？（现在，当他从第一台服务器发出命令时，他开始在另一台服务器上播放音乐）这是部分代码：

class MyBot(commands.Cog):
def __init__(self, bot: commands.Bot):
    self.bot = bot
    self.connect = None
    self.voice_channel = None

@commands.command()
async def join(self, ctx):
    self.voice_channel = discord.utils.get(ctx.guild.voice_channels, name=ctx.message.author.voice.channel.name)
    if not self.connect:
        self.connect = await self.voice_channel.connect()

@commands.command()
async def leave(self, ctx):
    if self.connect:
        await self.connect.disconnect()
        del self.connect