从零开始学python | 使用 gRPC 的 Python 微服务 II
从零开始学python | 使用 gRPC 的 Python 微服务 I
生产就绪的 Python 微服务
此时，您的开发机器上运行了一个 Python 微服务架构，非常适合测试。在本节中，您将使其在云中运行。
码头工人
Docker是一项了不起的技术，它可以让您将一组进程与同一台机器上的其他进程隔离开来。您可以拥有两组或更多组具有自己的文件系统、网络端口等的进程。你可以把它想象成一个 Python 虚拟环境，而且对于整个系统来说更加安全。
Docker 非常适合部署 Python 微服务，因为您可以打包所有依赖项并在隔离的环境中运行微服务。当您将微服务部署到云时，它可以与其他微服务在同一台机器上运行，而不会相互影响。这样可以更好地利用资源。
本教程不会深入探讨 Docker，因为它需要一整本书才能涵盖。相反，您只需掌握将 Python 微服务部署到云所需的基础知识。有关 Docker 的更多信息，您可以查看Python Docker 教程。
在开始之前，如果您想在您的机器上进行操作，请确保已安装 Docker。您可以从官方网站下载。
您将创建两个 Docker镜像，一个用于 Marketplace 微服务，另一个用于 Recommendations 微服务。图像基本上是一个文件系统加上一些元数据。本质上，您的每个微服务都将拥有一个自己的迷你 Linux 环境。它可以在不影响实际文件系统的情况下写入文件，并在不与其他进程冲突的情况下打开端口。
要创建图像，您需要定义一个Dockerfile. 您总是从一个包含一些基本内容的基本图像开始。在这种情况下，您的基本映像将包含一个 Python 解释器。然后将文件从开发机器复制到 Docker 映像中。您还可以在 Docker 映像中运行命令。这对于安装依赖项很有用。
建议 Dockerfile
您将首先创建 Recommendations 微服务 Docker 映像。创建recommendations/Dockerfile并添加以下内容：
1FROM python
2
3RUN mkdir /service
4COPY protobufs/ /service/protobufs/
5COPY recommendations/ /service/recommendations/
6WORKDIR /service/recommendations
7RUN python -m pip install --upgrade pip
8RUN python -m pip install -r requirements.txt
9RUN python -m grpc_tools.protoc -I ../protobufs --python_out=. \
10 --grpc_python_out=. ../protobufs/recommendations.proto
11
12EXPOSE 50051
13ENTRYPOINT [ "python", "recommendations.py" ]
这是逐行演练：
-
第 1 行使用基本的 Linux 环境和最新版本的 Python 初始化您的映像。此时，您的映像具有典型的 Linux 文件系统布局。如果你要进去看看，那就得
/bin，/home以及所有你所期望的基本文件。 -
第 3 行创建了一个新目录 at
/service以包含您的微服务代码。 -
第 4 行和第 5行将
protobufs/和recommendations/目录复制到/service. -
第 6 行给了 Docker 一条
WORKDIR /service/recommendations指令，这有点像cd在镜像里面做一个。您提供给 Docker 的任何路径都将相对于该位置，并且当您运行命令时，它将在该目录中运行。 -
第 7 行更新
pip以避免有关旧版本的警告。 -
第 8 行告诉 Docker
pip install -r requirements.txt在镜像内运行。这会将所有grpcio-tools文件以及您可能添加的任何其他包添加到映像中。请注意，您没有使用虚拟环境，因为它是不必要的。此映像中唯一运行的将是您的微服务，因此您无需进一步隔离其环境。 -
第 9行运行
python -m grpc_tools.protoc命令从 protobuf 文件生成 Python 文件。您/service在图像中的目录现在看起来像这样：/service/ | ├── protobufs/ │ └── recommendations.proto | └── recommendations/ ├── recommendations.py ├── recommendations_pb2.py ├── recommendations_pb2_grpc.py └── requirements.txt -
第 12 行告诉 Docker 你将在 port 上运行一个微服务
50051，并且你想把它暴露在镜像之外。 -
第 13 行告诉 Docker 如何运行你的微服务。
现在您可以从您的Dockerfile. 从包含所有代码的目录中运行以下命令 - 不是在recommendations/目录中，而是在其上一级：
$ docker build . -f recommendations/Dockerfile -t recommendations
这将为 Recommendations 微服务构建 Docker 映像。当 Docker 构建映像时，您应该会看到一些输出。现在你可以运行它：
$ docker run -p 127.0.0.1:50051:50051/tcp recommendations
您不会看到任何输出，但您的 Recommendations 微服务现在正在 Docker 容器中运行。当你运行一个镜像时，你会得到一个容器。您可以多次运行该映像以获得多个容器，但仍然只有一个映像。
该-p 127.0.0.1:50051:50051/tcp选项告诉 Docker 将机器端口上的TCP 连接转发50051到50051容器内的端口。这使您可以灵活地转发机器上的不同端口。
例如，如果您运行的两个容器都在 port 上运行 Python 微服务50051，那么您将需要在主机上使用两个不同的端口。这是因为两个进程不能同时打开同一个端口，除非它们在不同的容器中。
市场 Dockerfile
接下来，您将构建您的 Marketplace 映像。创建marketplace/Dockerfile并添加以下内容：
1FROM python
2
3RUN mkdir /service
4COPY protobufs/ /service/protobufs/
5COPY marketplace/ /service/marketplace/
6WORKDIR /service/marketplace
7RUN python -m pip install --upgrade pip
8RUN python -m pip install -r requirements.txt
9RUN python -m grpc_tools.protoc -I ../protobufs --python_out=. \
10 --grpc_python_out=. ../protobufs/recommendations.proto
11
12EXPOSE 5000
13ENV FLASK_APP=marketplace.py
14ENTRYPOINT [ "flask", "run", "--host=0.0.0.0"]
这与 Recommendations 非常相似，但Dockerfile有一些不同：
- 第 13 行用于
ENV FLASK_APP=marketplace.py设置FLASK_APP图像内部的环境变量。Flask 需要这个来运行。 - 第 14 行添加
--host=0.0.0.0到flask run命令中。如果你不添加这个，那么 Flask 将只接受来自 localhost 的连接。
但是等等，你不是还在运行所有东西localhost吗？嗯，不是真的。当您运行 Docker 容器时，默认情况下它与您的主机是隔离的。localhost容器内部与localhost外部不同，即使在同一台机器上。这就是为什么你需要告诉 Flask 接受来自任何地方的连接。
继续并打开一个新终端。您可以使用以下命令构建您的 Marketplace 映像：
$ docker build . -f marketplace/Dockerfile -t marketplace
这将创建 Marketplace 图像。您现在可以使用以下命令在容器中运行它：
$ docker run -p 127.0.0.1:5000:5000/tcp marketplace
您不会看到任何输出，但您的 Marketplace 微服务正在运行。
联网
不幸的是，即使您的 Recommendations 和 Marketplace 容器都在运行，如果您现在http://localhost:5000在浏览器中访问，您会收到错误消息。您可以连接到 Marketplace 微服务，但它无法再连接到 Recommendations 微服务。容器是隔离的。
幸运的是，Docker 提供了一个解决方案。您可以创建一个虚拟网络并将两个容器添加到其中。您还可以为他们提供 DNS 名称，以便他们可以找到彼此。
下面，您将创建一个名为的网络microservices并在其上运行 Recommendations 微服务。您还将为其指定 DNS 名称recommendations。首先，使用Ctrl+C停止当前正在运行的容器。然后运行以下命令：
$ docker network create microservices
$ docker run -p 127.0.0.1:50051:50051/tcp --network microservices \
--name recommendations recommendations
该docker network create命令创建网络。您只需要执行一次，然后就可以将多个容器连接到它。然后添加‑‑network microservices到docker run命令以在此网络上启动容器。该‑‑name recommendations选项为其提供了 DNS 名称recommendations。
在重新启动市场容器之前，您需要更改代码。这是因为您localhost:50051在以下行中进行了硬编码marketplace.py：
recommendations_channel = grpc.insecure_channel("localhost:50051")
现在您想要连接到recommendations:50051。但是，您可以从环境变量中加载它，而不是再次对其进行硬编码。用以下两行替换上面的行：
recommendations_host = os.getenv("RECOMMENDATIONS_HOST", "localhost")
recommendations_channel = grpc.insecure_channel(
f"{recommendations_host}:50051"
)
这将在环境变量中加载 Recommendations 微服务的主机名RECOMMENDATIONS_HOST。如果未设置，则可以将其默认为localhost. 这允许您直接在您的机器上或在容器内运行相同的代码。
由于您更改了代码，因此您需要重建市场映像。然后尝试在您的网络上运行它：
$ docker build . -f marketplace/Dockerfile -t marketplace
$ docker run -p 127.0.0.1:5000:5000/tcp --network microservices \
-e RECOMMENDATIONS_HOST=recommendations marketplace
这与您之前的运行方式类似，但有两个不同之处：
-
您添加了
‑‑network microservices在与您的 Recommendations 微服务相同的网络上运行它的选项。您没有添加‑‑name选项，因为与 Recommendations 微服务不同，不需要查找 Marketplace 微服务的 IP 地址。提供的端口转发-p 127.0.0.1:5000:5000/tcp就足够了，不需要DNS名称。 -
您添加了
-e RECOMMENDATIONS_HOST=recommendations，它在容器内设置环境变量。这就是将 Recommendations 微服务的主机名传递给代码的方式。
此时，您可以再次localhost:5000在浏览器中尝试，它应该可以正确加载。哈扎！
Docker 撰写
使用 Docker 可以完成所有这些工作真是太神奇了，但它有点乏味。如果您可以运行一个命令来启动所有容器，那就太好了。幸运的是有！它被称为docker-compose，它是 Docker 项目的一部分。
您可以在 YAML 文件中声明您的微服务，而不是运行一堆命令来构建图像、创建网络和运行容器：
1version: "3.8"
2services:
3
4 marketplace:
5 build:
6 context: .
7 dockerfile: marketplace/Dockerfile
8 environment:
9 RECOMMENDATIONS_HOST: recommendations
10 image: marketplace
11 networks:
12 - microservices
13 ports:
14 - 5000:5000
15
16 recommendations:
17 build:
18 context: .
19 dockerfile: recommendations/Dockerfile
20 image: recommendations
21 networks:
22 - microservices
23
24networks:
25 microservices:
通常，您将其放入名为docker-compose.yaml. 将其放在项目的根目录中：
.
├── marketplace/
│ ├── marketplace.py
│ ├── requirements.txt
│ └── templates/
│ └── homepage.html
|
├── protobufs/
│ └── recommendations.proto
|
├── recommendations/
│ ├── recommendations.py
│ ├── recommendations_pb2.py
│ ├── recommendations_pb2_grpc.py
│ └── requirements.txt
│
└── docker-compose.yaml
本教程不会详细介绍语法，因为它在其他地方有很好的文档记录。它实际上只是做与您已经手动完成的相同的事情。但是，现在您只需要运行一个命令来启动您的网络和容器：
$ docker-compose up
运行后，您应该可以再次localhost:5000在浏览器中打开，并且一切正常。
请注意，当容器与 Marketplace 微服务位于同一网络中时，您不需要50051在recommendations容器中公开，因此您可以删除该部分。
注意：使用 开发时docker-compose，如果您更改了任何文件，请运行docker-compose build以重建映像。如果您运行docker-compose up，它将使用旧图像，这可能会令人困惑。
如果您想docker-compose在向上移动之前停下来进行一些编辑，请按Ctrl+C。
测试
要对您的 Python 微服务进行单元测试，您可以实例化您的微服务类并调用其方法。以下是您的RecommendationService实现的基本示例测试：
1# recommendations/recommendations_test.py
2from recommendations import RecommendationService
3
4from recommendations_pb2 import BookCategory, RecommendationRequest
5
6def test_recommendations():
7 service = RecommendationService()
8 request = RecommendationRequest(
9 user_id=1, category=BookCategory.MYSTERY, max_results=1
10 )
11 response = service.Recommend(request, None)
12 assert len(response.recommendations) == 1
这是一个细分：
- 第 6 行像其他任何类一样实例化该类并在其上调用方法。
- 第 11 行传递
None上下文，只要您不使用它，它就可以工作。如果要测试使用上下文的代码路径，则可以模拟它。
集成测试涉及使用多个未模拟的微服务运行自动化测试。所以这有点复杂，但不是太难。添加marketplace/marketplace_integration_test.py文件：
from urllib.request import urlopen
def test_render_homepage():
homepage_html = urlopen("http://localhost:5000").read().decode("utf-8")
assert "<title>Online Books For You</title>" in homepage_html
assert homepage_html.count("<li>") == 3
这会向主页 URL 发出 HTTP 请求，并检查它是否返回了一些带有标题和三个<li>项目符号元素的HTML 。这不是最大的测试，因为如果页面上有更多内容，它就不会很容易维护，但它证明了一点。只有当 Recommendations 微服务启动并运行时，此测试才会通过。您甚至可以通过向它发出 HTTP 请求来测试 Marketplace 微服务。
那么你如何运行这种类型的测试呢？幸运的是，Docker 的好人也提供了一种方法来做到这一点。使用 运行 Python 微服务后docker-compose，您可以使用docker-compose exec. 因此，如果您想在marketplace容器内运行集成测试，可以运行以下命令：
$ docker-compose build
$ docker-compose up
$ docker-compose exec marketplace pytest marketplace_integration_test.py
这将pytest在marketplace容器内运行命令。由于您的集成测试连接到localhost，您需要在与微服务相同的容器中运行它。
部署到 Kubernetes
伟大的！您现在有几个微服务在您的计算机上运行。您可以快速启动它们并对它们运行集成测试。但是您需要让它们进入生产环境。为此，您将使用Kubernetes。
本教程不会深入介绍 Kubernetes，因为这是一个很大的主题，并且其他地方提供了全面的文档和教程。但是，在本节中，您将找到将 Python 微服务迁移到云中的 Kubernetes 集群的基础知识。
注意：要将 Docker 映像部署到云提供商，您需要将 Docker 映像推送到Docker Hub 之类的映像注册表。
以下示例使用本教程中的镜像，这些镜像已经推送到 Docker Hub。如果你想改变它们，或者如果你想创建自己的微服务，那么你需要在 Docker Hub 上创建一个帐户，以便你可以推送图像。如果您愿意，您也可以创建私有注册中心，或者使用其他注册中心，例如Amazon 的 ECR。
Kubernetes 配置
您可以从kubernetes.yaml. 完整的文件有点长，但它由四个不同的部分组成，因此您将一一查看它们：
1---
2apiVersion: apps/v1
3kind: Deployment
4metadata:
5 name: marketplace
6 labels:
7 app: marketplace
8spec:
9 replicas: 3
10 selector:
11 matchLabels:
12 app: marketplace
13 template:
14 metadata:
15 labels:
16 app: marketplace
17 spec:
18 containers:
19 - name: marketplace
20 image: hidan/python-microservices-article-marketplace:0.1
21 env:
22 - name: RECOMMENDATIONS_HOST
23 value: recommendations
这定义了Marketplace 微服务的部署。一个部署告诉Kubernetes如何部署你的代码。Kubernetes 需要四个主要信息：
- 要部署什么 Docker 镜像
- 部署多少个实例
- 微服务需要哪些环境变量
- 如何识别您的微服务
您可以使用标签告诉 Kubernetes 如何识别您的微服务。虽然这里没有显示，但您也可以告诉 Kubernetes 您的微服务需要哪些内存和 CPU 资源。您可以在Kubernetes 文档 中找到许多其他选项。
这是代码中发生的事情：
-
第 9 行告诉 Kubernetes 为您的微服务创建多少个 pod。一个Pod基本上是一个隔离的执行环境，就像一个轻量级的虚拟机，实现为一组容器。设置
replicas: 3为每个微服务提供三个 pod。拥有多个允许冗余，在不停机的情况下实现滚动更新，根据您需要的更多机器进行扩展，并在一台机器出现故障时进行故障转移。 -
第 20 行是要部署的 Docker 映像。您必须在映像注册表上使用 Docker 映像。要在那里获取您的映像，您必须将其推送到映像注册表。当您在 Docker Hub 上登录您的帐户时，有关于如何执行此操作的说明。
Recommendations 微服务的部署非常相似：
24---
25apiVersion: apps/v1
26kind: Deployment
27metadata:
28 name: recommendations
29 labels:
30 app: recommendations
31spec:
32 replicas: 3
33 selector:
34 matchLabels:
35 app: recommendations
36 template:
37 metadata:
38 labels:
39 app: recommendations
40 spec:
41 containers:
42 - name: recommendations
43 image: hidan/python-microservices-article-recommendations:0.1
主要区别在于一种使用名称marketplace，另一种使用recommendations. 您还在部署RECOMMENDATIONS_HOST上设置环境变量，marketplace但不在recommendations部署上设置。
接下来，您为 Recommendations 微服务定义一个服务。部署告诉 Kubernetes 如何部署您的代码，而服务则告诉它如何将请求路由到它。为避免与通常用于谈论微服务的术语服务混淆，您会在引用 Kubernetes 服务时看到大写这个词。
这是 的服务定义recommendations：
44---
45apiVersion: v1
46kind: Service
47metadata:
48 name: recommendations
49spec:
50 selector:
51 app: recommendations
52 ports:
53 - protocol: TCP
54 port: 50051
55 targetPort: 50051
这是定义中发生的事情：
-
第 48 行：当您创建服务时，Kubernetes 本质上会
name在集群内创建一个相同的 DNS 主机名。因此，您集群中的任何微服务都可以向recommendations. Kubernetes 会将此请求转发到您的 Deployment 中的一个 pod。 -
第 51行：此行将服务连接到部署。它告诉 Kubernetes 将请求转发到Deployment
recommendations中的 Pod 之一recommendations。这必须匹配labelsDeployment中的键值对之一。
该marketplace服务是类似的：
56---
57apiVersion: v1
58kind: Service
59metadata:
60 name: marketplace
61spec:
62 type: LoadBalancer
63 selector:
64 app: marketplace
65 ports:
66 - protocol: TCP
67 port: 5000
68 targetPort: 5000
除了名称和端口之外，只有一个区别。您会注意到它type: LoadBalancer仅出现在marketplace服务中。这是因为marketplace需要从 Kubernetes 集群外部访问，而recommendations只需要在集群内部访问。
注意：在具有许多微服务的大型集群中，使用IngressService 比使用Service更常见LoadBalancer。如果您正在企业环境中开发微服务，那么这可能是您要走的路。
查看 Sandeep Dinesh 的文章Kubernetes NodePort vs LoadBalancer vs Ingress？我什么时候应该使用什么？了解更多信息。
您可以通过展开下面的框来查看完整的文件：
完整kubernetes.yaml代码显示隐藏
现在您有了 Kubernetes 配置，下一步就是部署它！
Deploying Kubernetes
您通常使用云提供商部署 Kubernetes。您可以选择许多云提供商，包括Google Kubernetes Engine (GKE)、Amazon Elastic Kubernetes Service (EKS)和DigitalOcean。
如果您在公司部署微服务，那么您使用的云提供商很可能由您的基础设施决定。对于此演示，您将在本地运行 Kubernetes。几乎一切都与使用云提供商相同。
如果您在 Mac 或 Windows 上运行 Docker Desktop，那么它带有本地 Kubernetes 集群，您可以在“首选项”菜单中启用该集群。通过单击系统托盘中的 Docker 图标打开 Preferences，然后找到 Kubernetes 部分并启用它：
如果您在 Linux 上运行，那么您可以安装minikube。按照起始页上的说明进行设置。
创建集群后，您可以使用以下命令部署微服务：
$ kubectl apply -f kubernetes.yaml
如果您想尝试在云中部署到 Kubernetes，DigitalOcean 的设置最简单，并且具有简单的定价模型。您可以注册一个帐户，然后单击几下即可创建 Kubernetes 集群。如果您将默认值更改为仅使用一个节点和最便宜的选项，那么在撰写本文时，成本仅为每小时 0.015 美元。
按照 DigitalOcean 提供的说明下载配置文件kubectl并运行上述命令。然后，您可以单击DigitalOcean 中的Kubernetes按钮以查看在那里运行的服务。DigitalOcean 将为您的LoadBalancer服务分配一个 IP 地址，因此您可以通过将该 IP 地址复制到您的浏览器来访问您的 Marketplace 应用程序。
重要提示：完成后，请销毁您的集群，这样您就不会继续为此付费。您还应该转到 Networking 选项卡并销毁 Load Balancer，它与集群分开但也会产生费用。
到 Kubernetes 的部署到此结束。接下来，您将学习如何监控 Python 微服务。
使用拦截器进行 Python 微服务监控
一旦您在云中拥有一些微服务，您就希望了解它们的运行情况。您要监控的一些内容包括：
- 每个微服务收到多少请求
- 有多少请求导致错误，以及它们引发什么类型的错误
- 每个请求的延迟
- 异常日志，以便您可以稍后进行调试
您将在以下各节中了解执行此操作的几种方法。
为什么不是装饰器
一种可以做到这一点的方法，也是 Python 开发人员最自然的方法，就是为每个微服务端点添加一个装饰器。但是，在这种情况下，使用装饰器有几个缺点：
- 新微服务的开发人员必须记住将它们添加到每个方法中。
- 如果你有很多监控，那么你最终可能会得到一堆装饰器。
- 如果您有一堆装饰器，那么开发人员可能会以错误的顺序堆叠它们。
- 您可以将所有监控合并到一个装饰器中，但这样可能会变得混乱。
这堆装饰器是你想要避免的：
1class RecommendationService(recommendations_pb2_grpc.RecommendationsServicer):
2 @catch_and_log_exceptions
3 @log_request_counts
4 @log_latency
5 def Recommend(self, request, context):
6 ...
在每个方法上都有这堆装饰器是丑陋和重复的，它违反了DRY 编程原则：不要重复自己。装饰器也是写作的挑战，尤其是当他们接受参数时。
拦截器
您将在本教程中采用另一种使用装饰器的方法：gRPC 有一个拦截器概念，它提供类似于装饰器的功能，但方式更简洁。
实现拦截器
不幸的是，gRPC 的 Python 实现有一个相当复杂的拦截器 API。这是因为它非常灵活。但是，有一个grpc-interceptor包可以简化它们。为了完全披露，我是作者。
将它添加到您的recommendations/requirements.txtwith 中pytest，您将很快使用它：
grpc-interceptor ~= 0.12.0
grpcio-tools ~= 1.30
pytest ~= 5.4
然后更新您的虚拟环境：
$ python -m pip install recommendations/requirements.txt
您现在可以使用以下代码创建拦截器。您不需要将此添加到您的项目中，因为它只是一个示例：
1from grpc_interceptor import ServerInterceptor
2
3class ErrorLogger(ServerInterceptor):
4 def intercept(self, method, request, context, method_name):
5 try:
6 return method(request, context)
7 except Exception as e:
8 self.log_error(e)
9 raise
10
11 def log_error(self, e: Exception) -> None:
12 # ...
log_error()每当调用微服务中未处理的异常时，它就会调用。例如，您可以通过将异常记录到Sentry来实现这一点，以便在它们发生时获得警报和调试信息。
要使用这个拦截器，你可以grpc.server()像这样传递它：
interceptors = [ErrorLogger()]
server = grpc.server(futures.ThreadPoolExecutor(max_workers=10),
interceptors=interceptors)
使用此代码，Python 微服务的每个请求和响应都将通过您的拦截器，因此您可以计算它收到的请求和错误的数量。
grpc-interceptor还为每个 gRPC 状态代码和一个名为ExceptionToStatusInterceptor. 如果微服务引发异常之一，ExceptionToStatusInterceptor则将设置 gRPC 状态代码。这允许您通过将下面突出显示的更改更改为以下内容来简化您的微服务recommendations/recommendations.py：
1from grpc_interceptor import ExceptionToStatusInterceptor
2from grpc_interceptor.exceptions import NotFound
3
4# ...
5
6class RecommendationService(recommendations_pb2_grpc.RecommendationsServicer):
7 def Recommend(self, request, context):
8 if request.category not in books_by_category:
9 raise NotFound("Category not found")
10
11 books_for_category = books_by_category[request.category]
12 num_results = min(request.max_results, len(books_for_category))
13 books_to_recommend = random.sample(books_for_category, num_results)
14
15 return RecommendationResponse(recommendations=books_to_recommend)
16
17def serve():
18 interceptors = [ExceptionToStatusInterceptor()]
19 server = grpc.server(
20 futures.ThreadPoolExecutor(max_workers=10),
21 interceptors=interceptors
22 )
23 # ...
这更具可读性。您还可以从调用堆栈中的许多函数中引发异常，而不必传递context这样您就可以调用context.abort(). 您也不必自己在微服务中捕获异常——拦截器会为您捕获它。
测试拦截器
如果你想编写自己的拦截器，那么你应该测试它们。但是在测试拦截器之类的东西时模拟太多是危险的。例如，您可以调用.intercept()测试并确保它返回您想要的内容，但这不会测试真实的输入，甚至根本不会调用它们。
为了改进测试，您可以使用拦截器运行 gRPC 微服务。该grpc-interceptor包提供了一个框架来做到这一点。下面，您将为ErrorLogger拦截器编写一个测试。这只是一个示例，因此您无需将其添加到您的项目中。如果您要添加它，那么您会将其添加到测试文件中。
以下是为拦截器编写测试的方法：
1from grpc_interceptor.testing import dummy_client, DummyRequest, raises
2
3class MockErrorLogger(ErrorLogger):
4 def __init__(self):
5 self.logged_exception = None
6
7 def log_error(self, e: Exception) -> None:
8 self.logged_exception = e
9
10def test_log_error():
11 mock = MockErrorLogger()
12 ex = Exception()
13 special_cases = {"error": raises(ex)}
14
15 with dummy_client(special_cases=special_cases, interceptors=[mock]) as client:
16 # Test no exception
17 assert client.Execute(DummyRequest(input="foo")).output == "foo"
18 assert mock.logged_exception is None
19
20 # Test exception
21 with pytest.raises(grpc.RpcError) as e:
22 client.Execute(DummyRequest(input="error"))
23 assert mock.logged_exception is ex
这是一个演练：
-
ErrorLogger要模拟的第 3 到 8 行子类log_error()。您实际上并不希望发生日志记录副作用。你只是想确保它被调用。 -
第 15 到 18 行使用
dummy_client()上下文管理器创建连接到真实 gRPC 微服务的客户端。你发送DummyRequest到微服务，它回复DummyResponse. 默认情况下，input的DummyRequest回显到output的DummyResponse。但是，您可以传递dummy_client()特殊情况的字典，如果input匹配其中一个，则它将调用您提供的函数并返回结果。 -
第 21 到 23 行：您测试
log_error()调用时会出现预期的异常。raises()返回另一个引发提供的异常的函数。您设置input为error以便微服务将引发异常。
有关测试的更多信息，您可以阅读使用 Pytest 进行有效 Python 测试和了解 Python Mock 对象库。
在某些情况下，拦截器的替代方案是使用服务网格。它将通过代理发送所有微服务请求和响应，因此代理可以自动记录请求量和错误计数等内容。为了获得准确的错误日志，您的微服务仍然需要正确设置状态代码。因此，在某些情况下，您的拦截器可以补充服务网格。一种流行的服务网格是Istio。
最佳实践
现在你有一个有效的 Python 微服务设置。您可以创建微服务，一起测试它们，将它们部署到 Kubernetes，并使用拦截器监控它们。此时您可以开始创建微服务。但是，您应该记住一些最佳实践，因此您将在本节中学习一些。
Protobuf 组织
通常，您应该将 protobuf 定义与微服务实现分开。客户端几乎可以用任何语言编写，如果您将 protobuf 文件捆绑到Python 轮或类似的东西中，那么如果有人想要 Ruby 或 Go 客户端，他们将很难获得 protobuf 文件。
即使你所有的代码都是 Python，为什么有人需要为微服务安装包来为它编写客户端？
一种解决方案是将您的 protobuf 文件放在与微服务代码不同的 Git 存储库中。许多公司把所有的protobuf的文件全部微服务在一个单一的回购。这使得查找所有微服务、在它们之间共享通用 protobuf 结构以及创建有用的工具变得更加容易。
如果您选择将 protobuf 文件存储在单个 repo 中，则需要注意 repo 保持井井有条，并且绝对应该避免 Python 微服务之间的循环依赖。
Protobuf 版本控制
API 版本控制可能很困难。主要原因是如果你改变一个API并更新微服务，那么可能仍然有客户端使用旧的API。当客户端位于客户的机器上时尤其如此，例如移动客户端或桌面软件。
你不能轻易强迫人们更新。即使可以，网络延迟也会导致竞争条件，并且您的微服务很可能会使用旧 API 获取请求。好的 API 应该向后兼容或版本化。
为了实现向后兼容性，使用 protobufs 版本 3 的 Python 微服务将接受缺少字段的请求。如果您想添加一个新字段，那没关系。您可以先部署微服务，它仍然会接受来自旧 API 的请求，而没有新字段。微服务只需要优雅地处理它。
如果您想进行更大幅度的更改，则需要对API进行版本控制。Protobufs 允许您将 API 放入包命名空间中，其中可以包含版本号。如果您需要彻底更改 API，则可以创建它的新版本。微服务也可以继续接受旧版本。这允许您在逐步淘汰旧版本的同时推出新的 API 版本。
通过遵循这些约定，您可以避免进行重大更改。在公司内部，人们有时会觉得对 API 进行重大更改是可以接受的，因为他们控制着所有客户端。这由您决定，但请注意，进行重大更改需要协调客户端和微服务部署，并且会使回滚复杂化。
在微服务生命周期的早期，当没有生产客户端时，这可能没问题。但是，一旦您的微服务对您公司的健康至关重要，养成只进行不间断更改的习惯是很好的。
Protobuf Linting
确保不会对 protobuf 进行重大更改的一种方法是使用linter。一个流行的是buf。您可以将其设置为CI 系统的一部分，以便您可以检查拉取请求中的重大更改。
类型检查 Protobuf 生成的代码
Mypy 是一个用于静态类型检查 Python 代码的项目。如果您不熟悉 Python 中的静态类型检查，那么您可以阅读Python 类型检查以了解所有相关信息。
生成的代码protoc有点粗糙，而且没有类型注释。如果您尝试使用 Mypy 进行类型检查，那么您将收到很多错误，并且它不会捕获真正的错误，例如字段名称拼写错误。幸运的是，Dropbox 的好人为编译器编写了一个插件protoc来生成类型存根。这些不应与 gRPC 存根混淆。
为了使用它，您可以安装mypy-protobuf包，然后更新命令以生成 protobuf 输出。注意新‑‑mypy_out选项：
$ python -m grpc_tools.protoc -I ../protobufs --python_out=. \
--grpc_python_out=. --mypy_out=. ../protobufs/recommendations.proto
大多数 Mypy 错误应该会消失。您可能仍会收到有关grpc包没有类型信息的错误。您可以安装非官方的gRPC 类型存根或将以下内容添加到您的 Mypy 配置中：
[mypy-grpc.*]
ignore_missing_imports = True
您仍将获得类型检查的大部分好处，例如捕获拼写错误的字段。这对于在将错误投入生产之前捕获错误非常有帮助。
从零开始学python | 使用 gRPC 的 Python 微服务 III
- 点赞
- 收藏
- 关注作者
评论（0）