问题标签 [grpc-python]

最近，我一直在使用 MagicMock() 在 python 中模拟 gRPC 调用，但我没有成功模拟我得到的 gRPC 响应消息。这是一段代码python客户端代码

这是我模拟它的方式（仍然是新手学习模拟框架）

Flow 是 python 函数 senor_read() 向远程服务器发出 gRPC 请求并获取读数并传回给调用者。

此测试的目的是使代码在“eif”情况下执行并捕获异常。1）当我执行测试代码时，如果 case 并且我看到一些其他异常被发送给调用者，它不会转到“elf”。我是否在 gRPC 响应模拟中正确设置了状态？

2) 我的模拟步骤对于模拟 gRPC 调用是否正确？

3）此外，当我运行时，我无法在成功时看到打印消息，而只能在失败测试用例打印时看到。另外，实际功能中的打印不打印？

我确实搜索了很多，找不到太多关于在 python 中模拟 gRPC 的信息。

我正在尝试使用 gRPC 在 python 上构建一个网络爬虫。我已经在服务器文件中包含了用于抓取的功能，并使用客户端向用户请求 URL 列表并将其发送到服务器以进行抓取部分。每个网址大约需要 25-30 秒才能被抓取。因此，我想使用多处理来加速该过程，即并行使用 N 个内核从 N 个 URL 中提取信息。我该如何进行？比如说，我有 4 个内核：是否可以在 4 个不同的内核上实现客户端对服务器的 4 次调用？还是应该在每个核心上分别创建一个服务器-客户端对？或者我可以创建 4 个具有不同通道端口的服务器实例并在 4 个内核上执行它们吗？

我对这一切都很陌生。所以，我可以在这方面使用任何形式的帮助。

对于我们新的开放实验室设备自动化标准（https://gitlab.com/SiLA2/sila_python），我们希望以两种模式运行设备（=gRPC 服务器）：模拟模式和真实模式（具有相同的设置远程调用，但在第一种情况下，它应该只返回模拟响应，在第二种情况下它应该与硬件通信。

我的第一个想法是在单独的模块中创建两个几乎相同的 gRPC 服务程序 python 类，如下例所示：

在hello_sim.py中：

在hello_real.py中：

然后，在 server.py 中创建 gRPC 服务器后，我可以通过在 gRPC 服务器上重新注册 servcier 来在模拟和实模式之间切换，例如：

服务器.py

能够调用真正的硬件代码；或通过交换对服务对象的引用，例如：

但是这两种策略都不起作用:(

从 gRPC 客户端以模拟模式调用服务器时，我希望它应该回复（根据示例）：“模拟”

gRPC_client.py

并在切换到实模式（通过任何机制）“真实”之后：

在不完全关闭 gRPC 服务器（以及失去与客户端的连接）的情况下实现这种模式切换的最简洁和优雅的解决方案是什么？

非常感谢您提前提供的帮助！

PS：（关闭 gRPC 服务器并重新注册当然可以，但这不是我们想要的。）

我正在尝试使用 dart(flutter) 客户端和带有 gRPC api 的 python 服务器进行实时图像分析。如何在 gRPC 中使用 dart-client 设置异步图像流？（如 websockets）

已经使用 grpc 创建了一个 rpc 服务。当我尝试在 grpc 调用（预测）中调用张量流预测函数时。我收到与张量流相关的错误，并发现 tf.global_variables() 变为空。这是代码

帮助将不胜感激！

我确实阅读了 [this similar question][1] 的答案，但它无助于为我解决问题。

我的设置：

• 远程 gRPC 服务；
• 直接在主机上.py运行的客户端。

在该配置中，一切正常。但是，如果我remote gRPC service在本地 Docker 容器中启动它（.py客户端仍然在本地运行）：

我用来运行 gRPC 服务的命令：sudo docker run --rm -it -u dud --net=host --entrypoint=/usr/local/bin/application COOL_APP

.py这是我的客户的一段代码：

我收到以下错误（AFAIK 这意味着我的.py客户端无法连接到host:port我的 Docker 服务）：

我已经尝试在我的客户端中同时设置localhost:9000、0.0.0.0:9000和，但没有成功。:9000.py

我不确定这是否有意义，但是当我运行时：

即，我的终端没有显示任何内容tcp:9000（我运行上面的命令来检查是否真的有东西在听localhost:9000）。

更新：当我运行hello-world[container][2] 时，-p 9000:9000我收到一个不同的错误：

我创建了一个简单的服务器拦截器，它根据 JWT 令牌检索用户。

但现在我想让它可用于我服务的所有方法。

目前我正在使用装饰器。但我想避免不得不装饰所有的方法。以防万一，只装饰那些不需要用户的。

有人可以给我一个线索吗？

这是我的代码：

而且我希望我的方法能够以这样的方式访问用户。

我想在 python 中使用带有 grpc 调用的谷歌云 pub/sub。我该如何实施？有人可以为此提供良好的资源吗？

在尝试使用 docker 运行 tensorflow-serving 时，我收到以下错误，使用带有以下代码的 gRPC 发出客户端请求：

关于我的环境的信息：

操作系统：macOS 虚拟环境：Anaconda 3 Python 3.6 gRPC/tools 1.19

你能帮我解决这个问题吗？

我正在尝试做的事情：当我对对象的状态进行更新时，应该通过 gRPC 流向所有 gRPC 客户端提供更新。重要的是每个客户端都获得每次更新，并且他们只获得一次。

我期望发生的事情：当我立即执行 event.set() 和 event.clear() 时，所有客户端都将运行一次，产生新的状态。

实际发生的情况：客户端缺少更新。例如，我的服务功能发送了 10 个版本更新。在客户端，它缺少这些更新，我会查看它在哪里更新 1 2 然后错过 3 或其他更新，然后再次开始获取它们。

服务器版本 1，这不起作用，因为客户端缺少一些更新：

服务器版本 2，此版本有效，但我认为存在竞争条件： 在第二个版本中，我没有使用 threading.Event，而是使用布尔值 new_update，它在所有线程之间共享。在 serve 函数中，我将其设置为 true，然后所有线程都将其设置为 False。

我相信第二个版本之所以有效，是因为它依赖于 CPython 的全局解释器锁，确保没有线程会同时改变 new_update。我不喜欢这个解决方案，我有什么选择？另外，我知道我可以创建一个队列或列表并存储所有更改，然后跟踪每个连接的客户端所在的位置，我不想分配内存来执行此操作。