在 Kubernetes 上使用 Flask 搭建 Python 微服務

微服務遵循領域驅動設計（DDD），與開發平台無關。Python 微服務也不例外。Python3 的面向對象特性使得按照 DDD 對服務進行建模變得更加容易。本系列的第 10 部分演示了如何將用戶管理系統的查找服務作為 Python 微服務部署在 Kubernetes 上。

微服務架構的強大之處在於它的多語言性。企業將其功能分解為一組微服務，每個團隊自由選擇一個平台。

我們的用戶管理系統已經分解為四個微服務，分別是添加、查找、搜索和日誌服務。添加服務在 Java 平台上開發並部署在 Kubernetes 集群上，以實現彈性和可擴展性。這並不意味著其餘的服務也要使用 Java 開發，我們可以自由選擇適合個人服務的平台。

讓我們選擇 Python 作為開發查找服務的平台。查找服務的模型已經設計好了（參考 2022 年 3 月份的文章），我們只需要將這個模型轉換為代碼和配置。

Pythonic 方法

Python 是一種通用編程語言，已經存在了大約 30 年。早期，它是自動化腳本的首選。然而，隨著 Django 和 Flask 等框架的出現，它的受歡迎程度越來越高，現在各種領域中都在應用它，如企業應用程序開發。數據科學和機器學習進一步推動了它的發展，Python 現在是三大編程語言之一。

許多人將 Python 的成功歸功於它容易編碼。這只是一部分原因。只要你的目標是開發小型腳本，Python 就像一個玩具，你會非常喜歡它。然而，當你進入嚴肅的大規模應用程序開發領域時，你將不得不處理大量的 if 和 else，Python 變得與任何其他平台一樣好或一樣壞。例如，採用一種面向對象的方法！許多 Python 開發人員甚至可能沒意識到 Python 支持類、繼承等功能。Python 確實支持成熟的面向對象開發，但是有它自己的方式 -- Pythonic！讓我們探索一下！

領域模型

AddService 通過將數據保存到一個 MySQL 資料庫中來將用戶添加到系統中。FindService 的目標是提供一個 REST API 按用戶名查找用戶。域模型如圖 1 所示。它主要由一些值對象組成，如 User 實體的Name、PhoneNumber 以及 UserRepository。

圖 1: 查找服務的域模型

讓我們從 Name 開始。由於它是一個值對象，因此必須在創建時進行驗證，並且必須保持不可變。基本結構如所示：

class Name:
    value: str
    def __post_init__(self):
        if self.value is None or len(self.value.strip()) < 8 or len(self.value.strip()) > 32:
            raise ValueError("Invalid Name")

如你所見，Name 包含一個字元串類型的值。作為後期初始化的一部分，我們會驗證它。

Python 3.7 提供了 @dataclass 裝飾器，它提供了許多開箱即用的數據承載類的功能，如構造函數、比較運算符等。如下是裝飾後的 Name 類：

from dataclasses import dataclass

@dataclass
class Name:
    value: str
    def __post_init__(self):
        if self.value is None or len(self.value.strip()) < 8 or len(self.value.strip()) > 32:
            raise ValueError("Invalid Name")

以下代碼可以創建一個 Name 對象：

name = Name("Krishna")

value 屬性可以按照如下方式讀取或寫入：

name.value = "Mohan"
print(name.value)

可以很容易地與另一個 Name 對象比較，如下所示：

other = Name("Mohan")
if name == other:
    print("same")

如你所見，對象比較的是值而不是引用。這一切都是開箱即用的。我們還可以通過凍結對象使對象不可變。這是 Name 值對象的最終版本：

from dataclasses import dataclass

@dataclass(frozen=True)
class Name:
    value: str
    def __post_init__(self):
        if self.value is None or len(self.value.strip()) < 8 or len(self.value.strip()) > 32:
            raise ValueError("Invalid Name")

PhoneNumber 也遵循類似的方法，因為它也是一個值對象：

@dataclass(frozen=True)
class PhoneNumber:
    value: int
    def __post_init__(self):
        if self.value < 9000000000:
            raise ValueError("Invalid Phone Number")

User 類是一個實體，不是一個值對象。換句話說，User 是可變的。以下是結構：

from dataclasses import dataclass
import datetime

@dataclass
class User:
    _name: Name
    _phone: PhoneNumber
    _since: datetime.datetime

    def __post_init__(self):
        if self._name is None or self._phone is None:
            raise ValueError("Invalid user")
        if self._since is None:
            self.since = datetime.datetime.now()

你能觀察到 User 並沒有凍結，因為我們希望它是可變的。但是，我們不希望所有屬性都是可變的。標識欄位如 _name 和 _since 是希望不會修改的。那麼，這如何做到呢？

Python3 提供了所謂的描述符協議，它會幫助我們正確定義 getter 和 setter。讓我們使用 @property 裝飾器將 getter 添加到 User 的所有三個欄位中。

@property
def name(self) -> Name:
    return self._name

@property
def phone(self) -> PhoneNumber:
    return self._phone

@property
def since(self) -> datetime.datetime:
    return self._since

phone 欄位的 setter 可以使用 @<欄位>.setter 來裝飾：

@phone.setter
def phone(self, phone: PhoneNumber) -> None:
    if phone is None:
        raise ValueError("Invalid phone")
    self._phone = phone

通過重寫 __str__() 函數，也可以為 User 提供一個簡單的列印方法：

def __str__(self):
    return self.name.value + " [" + str(self.phone.value) + "] since " + str(self.since)

這樣，域模型的實體和值對象就準備好了。創建異常類如下所示：

class UserNotFoundException(Exception):
    pass

域模型現在只剩下 UserRepository 了。Python 提供了一個名為 abc 的有用模塊來創建抽象方法和抽象類。因為 UserRepository 只是一個介面，所以我們可以使用 abc 模塊。

任何繼承自 abc.ABC 的類都將變為抽象類，任何帶有 @abc.abstractmethod 裝飾器的函數都會變為一個抽象函數。下面是 UserRepository 的結構：

from abc import ABC, abstractmethod

class UserRepository(ABC):
    @abstractmethod
    def fetch(self, name:Name) -> User:
        pass

UserRepository 遵循倉儲模式。換句話說，它在 User 實體上提供適當的 CRUD 操作，而不會暴露底層數據存儲語義。在本例中，我們只需要 fetch() 操作，因為 FindService 只查找用戶。

因為 UserRepository 是一個抽象類，我們不能從抽象類創建實例對象。創建對象必須依賴於一個具體類實現這個抽象類。數據層 UserRepositoryImpl 提供了 UserRepository 的具體實現：

class UserRepositoryImpl(UserRepository):
    def fetch(self, name:Name) -> User:
        pass

由於 AddService 將用戶數據存儲在一個 MySQL 資料庫中，因此 UserRepositoryImpl 也必須連接到相同的資料庫去檢索數據。下面是連接到資料庫的代碼。注意，我們正在使用 MySQL 的連接庫。

from mysql.connector import connect, Error

class UserRepositoryImpl(UserRepository):
    def fetch(self, name:Name) -> User:
        try:
            with connect(
                    host="mysqldb",
                    user="root",
                    password="admin",
                    database="glarimy",
                ) as connection:
                with connection.cursor() as cursor:
                    cursor.execute("SELECT * FROM ums_users where name=%s", (name.value,))
                    row = cursor.fetchone()
                    if cursor.rowcount == -1:
                        raise UserNotFoundException()
                    else:
                        return User(Name(row[0]), PhoneNumber(row[1]), row[2])
        except Error as e:
            raise e

在上面的片段中，我們使用用戶 root / 密碼 admin 連接到一個名為 mysqldb 的資料庫伺服器，使用名為 glarimy 的資料庫（模式）。在演示代碼中是可以包含這些信息的，但在生產中不建議這麼做，因為這會暴露敏感信息。

fetch() 操作的邏輯非常直觀，它對 ums_users 表執行 SELECT 查詢。回想一下，AddService 正在將用戶數據寫入同一個表中。如果 SELECT 查詢沒有返回記錄，fetch() 函數將拋出 UserNotFoundException 異常。否則，它會從記錄中構造 User 實體並將其返回給調用者。這沒有什麼特殊的。

應用層

最終，我們需要創建應用層。此模型如圖 2 所示。它只包含兩個類：控制器和一個 DTO。

圖 2: 添加服務的應用層

眾所周知，一個 DTO 只是一個沒有任何業務邏輯的數據容器。它主要用於在 FindService 和外部之間傳輸數據。我們只是提供了在 REST 層中將 UserRecord 轉換為字典以便用於 JSON 傳輸：

class UserRecord:
    def toJSON(self):
        return {
            "name": self.name,
            "phone": self.phone,
            "since": self.since
        }

控制器的工作是將 DTO 轉換為用於域服務的域對象，反之亦然。可以從 find() 操作中觀察到這一點。

class UserController:

    def __init__(self):
        self._repo = UserRepositoryImpl()

    def find(self, name: str):
        try:
            user: User = self._repo.fetch(Name(name))
            record: UserRecord = UserRecord()
            record.name = user.name.value
            record.phone = user.phone.value
            record.since = user.since
            return record
        except UserNotFoundException as e:
            return None

find() 操作接收一個字元串作為用戶名，然後將其轉換為 Name 對象，並調用 UserRepository 獲取相應的 User 對象。如果找到了，則使用檢索到的 User 對象創建UserRecord`。回想一下，將域對象轉換為 DTO 是很有必要的，這樣可以對外部服務隱藏域模型。

UserController 不需要有多個實例，它也可以是單例的。通過重寫 __new__，可以將其建模為一個單例。

class UserController:
    def __new__(self):
        if not hasattr(self, 『instance』):
            self.instance = super().__new__(self)
        return self.instance

    def __init__(self):
        self._repo = UserRepositoryImpl()

    def find(self, name: str):
        try:
            user: User = self._repo.fetch(Name(name))
            record: UserRecord = UserRecord()
            record.name = user.name.getValue()
            record.phone = user.phone.getValue()
            record.since = user.since
            return record
        except UserNotFoundException as e:
            return None

我們已經完全實現了 FindService 的模型，剩下的唯一任務是將其作為 REST 服務公開。

REST API

FindService 只提供一個 API，那就是通過用戶名查找用戶。顯然 URI 如下所示：

GET /user/{name}

此 API 希望根據提供的用戶名查找用戶，並以 JSON 格式返回用戶的電話號碼等詳細信息。如果沒有找到用戶，API 將返回一個 404 狀態碼。

我們可以使用 Flask 框架來構建 REST API，它最初的目的是使用 Python 開發 Web 應用程序。除了 HTML 視圖，它還進一步擴展到支持 REST 視圖。我們選擇這個框架是因為它足夠簡單。創建一個 Flask 應用程序：

from flask import Flask
app = Flask(__name__)

然後為 Flask 應用程序定義路由，就像函數一樣簡單：

@app.route(&apos;/user/<name>&apos;)
def get(name):
    pass

注意 @app.route 映射到 API /user/<name>，與之對應的函數的 get()。

如你所見，每次用戶訪問 API 如 http://server:port/user/Krishna 時，都將調用這個 get() 函數。Flask 足夠智能，可以從 URL 中提取 Krishna 作為用戶名，並將其傳遞給 get() 函數。

get() 函數很簡單。它要求控制器找到該用戶，並將其與通常的 HTTP 頭一起打包為 JSON 格式後返回。如果控制器返回 None，則 get() 函數返回合適的 HTTP 狀態碼。

from flask import jsonify, abort

controller = UserController()
record = controller.find(name)
if record is None:
    abort(404)
else:
    resp = jsonify(record.toJSON())
    resp.status_code = 200
    return resp

最後，我們需要 Flask 應用程序提供服務，可以使用 waitress 服務：

from waitress import serve
serve(app, host="0.0.0.0", port=8080)

在上面的片段中，應用程序在本地主機的 8080 埠上提供服務。最終代碼如下所示：

from flask import Flask, jsonify, abort
from waitress import serve

app = Flask(__name__)

@app.route(&apos;/user/<name>&apos;)
def get(name):
    controller = UserController()
    record = controller.find(name)
    if record is None:
        abort(404)
    else:
        resp = jsonify(record.toJSON())
        resp.status_code = 200
        return resp

serve(app, host="0.0.0.0", port=8080)

部署

FindService 的代碼已經準備完畢。除了 REST API 之外，它還有域模型、數據層和應用程序層。下一步是構建此服務，將其容器化，然後部署到 Kubernetes 上。此過程與部署其他服務妹有任何區別，但有一些 Python 特有的步驟。

在繼續前進之前，讓我們來看下文件夾和文件結構：

+ ums-find-service
+ ums
- domain.py
- data.py
- app.py
- Dockerfile
- requirements.txt
- kube-find-deployment.yml

如你所見，整個工作文件夾都位於 ums-find-service 下，它包含了 ums 文件夾中的代碼和一些配置文件，例如 Dockerfile、requirements.txt 和 kube-find-deployment.yml。

domain.py 包含域模型，data.py 包含 UserRepositoryImpl，app.py 包含剩餘代碼。我們已經閱讀過代碼了，現在我們來看看配置文件。

第一個是 requirements.txt，它聲明了 Python 系統需要下載和安裝的外部依賴項。我們需要用查找服務中用到的每個外部 Python 模塊來填充它。如你所見，我們使用了 MySQL 連接器、Flask 和 Waitress 模塊。因此，下面是 requirements.txt 的內容。

Flask==2.1.1
Flask_RESTful
mysql-connector-python
waitress

第二步是在 Dockerfile 中聲明 Docker 相關的清單，如下：

FROM python:3.8-slim-buster

WORKDIR /ums
ADD ums /ums
ADD requirements.txt requirements.txt
RUN pip3 install -r requirements.txt

EXPOSE 8080
ENTRYPOINT ["python"]
CMD ["/ums/app.py"]

總的來說，我們使用 Python 3.8 作為基線，除了移動 requirements.txt 之外，我們還將代碼從 ums 文件夾移動到 Docker 容器中對應的文件夾中。然後，我們指示容器運行 pip3 install 命令安裝對應模塊。最後，我們向外暴露 8080 埠（因為 waitress 運行在此埠上）。

為了運行此服務，我們指示容器使用使用以下命令：

python /ums/app.py

一旦 Dockerfile 準備完成，在 ums-find-service 文件夾中運行以下命令，創建 Docker 鏡像：

docker build -t glarimy/ums-find-service

它會創建 Docker 鏡像，可以使用以下命令查找鏡像：

docker images

嘗試將鏡像推送到 Docker Hub，你也可以登錄到 Docker。

docker login
docker push glarimy/ums-find-service

最後一步是為 Kubernetes 部署構建清單。

在之前的文章中，我們已經介紹了如何建立 Kubernetes 集群、部署和使用服務的方法。我假設仍然使用之前文章中的清單文件來部署添加服務、MySQL、Kafka 和 Zookeeper。我們只需要將以下內容添加到 kube-find-deployment.yml 文件中：

apiVersion: apps/v1
kind: Deployment
metadata:
name: ums-find-service
labels:
app: ums-find-service
spec:
replicas: 3
selector:
matchLabels:
app: ums-find-service
template:
metadata:
labels:
app: ums-find-service
spec:
containers:
- name: ums-find-service
image: glarimy/ums-find-service
ports:
- containerPort: 8080
apiVersion: v1
kind: Service
metadata:
name: ums-find-service
labels:
name: ums-find-service
spec:
type: LoadBalancer
ports:
- port: 8080
selector:
app: ums-find-service

上面清單文件的第一部分聲明了 glarimy/ums-find-service 鏡像的 FindService，它包含三個副本。它還暴露 8080 埠。清單的後半部分聲明了一個 Kubernetes 服務作為 FindService 部署的前端。請記住，在之前文章中，mysqldb 服務已經是上述清單的一部分了。

運行以下命令在 Kubernetes 集群上部署清單文件：

kubectl create -f kube-find-deployment.yml

部署完成後，可以使用以下命令驗證容器組和服務：

kubectl get services

輸出如圖 3 所示：

圖 3: Kubernetes 服務

它會列出集群上運行的所有服務。注意查找服務的外部 IP，使用 curl 調用此服務：

curl http://10.98.45.187:8080/user/KrishnaMohan

注意：10.98.45.187 對應查找服務，如圖 3 所示。

如果我們使用 AddService 創建一個名為 KrishnaMohan 的用戶，那麼上面的 curl 命令看起來如圖 4 所示：

圖 4: 查找服務

用戶管理系統（UMS）的體系結構包含 AddService 和 FindService，以及存儲和消息傳遞所需的後端服務，如圖 5 所示。可以看到終端用戶使用 ums-add-service 的 IP 地址添加新用戶，使用 ums-find-service 的 IP 地址查找已有用戶。每個 Kubernetes 服務都由三個對應容器的節點支持。還要注意：同樣的 mysqldb 服務用於存儲和檢索用戶數據。

圖 5: UMS 的添加服務和查找服務

其他服務

UMS 系統還包含兩個服務：SearchService 和 JournalService。在本系列的下一部分中，我們將在 Node 平台上設計這些服務，並將它們部署到同一個 Kubernetes 集群，以演示多語言微服務架構的真正魅力。最後，我們將觀察一些與微服務相關的設計模式。

via: https://www.opensourceforu.com/2022/09/python-microservices-using-flask-on-kubernetes/

作者：Krishna Mohan Koyya 選題：lkxed 譯者：MjSeven 校對：wxy

本文由 LCTT 原創編譯，Linux中國榮譽推出

本文轉載來自 Linux 中國: https://github.com/Linux-CN/archive

對這篇文章感覺如何？

太棒了

不錯

愛死了

不太好

感覺很糟

Rain

雨落清風。心向陽

在 Kubernetes 上使用 Flask 搭建 Python 微服務

Pythonic 方法

領域模型

應用層

REST API

部署

其他服務

對這篇文章感覺如何？

在 Linux 中創建 LVM 分區的分步指南

Xubuntu 22.10：熱門新功能

Leave a reply 取消回復

More in:Linux中國

如何通過 VLC 使用字幕

Unix 桌面：在 Linux 問世之前

Valve 對於 Ubuntu 的 Snap 版本的 Steam 並不滿意：原因何在

Wine 9.0 發布，實驗性地加入了 Wayland 驅動

中文操作系統論壇

關注 LinuxStory

開源學村

編程類開放書籍薈萃

如何殺死 Linux 中的殭屍進程

2022年，從學習Rust開始

5本學習 TeX 的最佳免費書籍

Makefile 簡介

Linux 內核補丁提交初體驗

使用 Linux 命令行解決Wordle 問題

Linux 的前世今生 – 1

特別關注

Самое Популярное Казино В России: Обзор Площадки «Azino777 Вход»

在 Ubuntu 24.04 上安裝和配置 VNC 遠程桌面

學校教師和計算機愛好者開發的法語 Linux 發行版：PrimTux 8 正式發布

LinuxStory

加入 LinuxStory 交流群組

投票調查

最熱標籤

Pythonic 方法

領域模型

應用層

REST API

部署

其他服務

分享

對這篇文章感覺如何？

You may also like

Leave a reply 取消回復

More in:Linux中國

中文操作系統論壇

關注 LinuxStory

開源學村

特別關注

最新文章

最熱標籤