Linux中國

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

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

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

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

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

Pythonic 方法

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

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

領域模型

AddService 通過將數據保存到一個 MySQL 資料庫中來將用戶添加到系統中。FindService 的目標是提供一個 REST API 按用戶名查找用戶。域模型如圖 1 所示。它主要由一些值對象組成,如 User 實體的NamePhoneNumber 以及 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 文件夾中的代碼和一些配置文件,例如 Dockerfilerequirements.txtkube-find-deployment.yml

domain.py 包含域模型,data.py 包含 UserRepositoryImplapp.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)的體系結構包含 AddServiceFindService,以及存儲和消息傳遞所需的後端服務,如圖 5 所示。可以看到終端用戶使用 ums-add-service 的 IP 地址添加新用戶,使用 ums-find-service 的 IP 地址查找已有用戶。每個 Kubernetes 服務都由三個對應容器的節點支持。還要注意:同樣的 mysqldb 服務用於存儲和檢索用戶數據。

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

其他服務

UMS 系統還包含兩個服務:SearchServiceJournalService。在本系列的下一部分中,我們將在 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

對這篇文章感覺如何?

太棒了
0
不錯
0
愛死了
0
不太好
0
感覺很糟
0
雨落清風。心向陽

    You may also like

    Leave a reply

    您的電子郵箱地址不會被公開。 必填項已用 * 標註

    此站點使用Akismet來減少垃圾評論。了解我們如何處理您的評論數據

    More in:Linux中國