如何使用Python構建和部署微服務

如何使用Python構建和部署微服務

在現代軟體開發中,微服務已成為一種舉足輕重的架構,可實現複雜系統的可擴充套件性、靈活性和高效管理。

微服務是執行特定任務的小型獨立應用程式,允許靈活部署和擴充套件。這種模組化的軟體設計方法鬆散了元件之間的耦合,提高了整個開發過程的靈活性和可管理性。

本文概述了微服務、微服務的功能以及使用 Python 建立微服務的過程。文章還演示了使用 Dockerfile 將微服務部署到雲伺服器。

什麼是微服務?

微服務是應用程式中獨立自主的服務,每個服務都能滿足特定的業務需求。它們通過輕量級應用程式介面或訊息代理進行通訊,形成一個綜合系統。

與完全根據需求進行擴充套件的單體系統不同,微服務允許擴充套件單個高流量元件。這種架構便於故障管理和功能更新,消除了單體系統的侷限性。

使用微服務有以下幾個好處:

  • 靈活性和可擴充套件性-對單個服務進行解耦,可以增加執行高流量特定服務例項的節點數量。
  • 程式碼模組化-每個服務都可以使用獨立的技術堆疊,這意味著您可以為每個服務選擇最佳的開發工具。

不過,微服務架構也面臨一些挑戰:

  • 監控多個服務-由於特定服務的例項被部署並分佈在多個節點上,因此監控系統中的單個服務變得非常困難。這種困難在網路故障或其他系統問題時尤為明顯。
  • 成本-由於管理多個服務的相關成本,開發微服務應用程式的成本可能比構建單體系統高得多。每個服務都需要自己的基礎設施和資源,這可能會導致成本高昂–尤其是在系統擴充套件時。

如何使用 Python 設計微服務

既然您已經瞭解了使用微服務架構的好處,那麼現在就該用 Python 構建一個微服務架構了。

在這個示例中,假設您想構建一個電子商務 Web 應用程式。該網站有幾個元件,包括產品目錄、訂單列表、支付處理系統和日誌,您需要將每個元件作為獨立的服務來實現。此外,您還需要建立服務與服務之間的通訊方法,以便在這些服務(如 HTTP)之間高效地傳輸資料。

讓我們使用 Python 構建一個微服務來管理產品目錄。該微服務將從指定來源獲取產品資料,並以 JSON 格式返回資料。

前提條件

要學習本教學,請確保您具備以下條件:

1. 建立您的專案

首先,為專案建立一個名為 flask-microservice 的資料夾,並將當前目錄放入專案目錄。

接下來,執行 python3 --version 確認 Python 是否已正確安裝到電腦上。

執行下面的命令安裝 virtualenv ,為 Flask 微服務建立一個隔離的開發環境:

Plain text
Copy to clipboard
Open code in new window
EnlighterJS 3 Syntax Highlighter
pip3 install virtualenv
pip3 install virtualenv
pip3 install virtualenv

執行以下程式建立虛擬環境:

Plain text
Copy to clipboard
Open code in new window
EnlighterJS 3 Syntax Highlighter
virtualenv venv
virtualenv venv
virtualenv venv

最後,根據計算機作業系統使用以下命令之一啟用虛擬環境:

Plain text
Copy to clipboard
Open code in new window
EnlighterJS 3 Syntax Highlighter
# Windows:
.\venv\Scripts\activate
# Unix or macOS:
source venv/bin/activate
# Windows: .\venv\Scripts\activate # Unix or macOS: source venv/bin/activate
# Windows: 
.\venv\Scripts\activate
# Unix or macOS:
source venv/bin/activate

2. 設定 Flask 伺服器

在根目錄下建立 requirements.txt 檔案,並新增這些依賴項。

Plain text
Copy to clipboard
Open code in new window
EnlighterJS 3 Syntax Highlighter
flask
requests
flask requests
flask
requests

在終端上執行 pip3 命令安裝依賴項。

Plain text
Copy to clipboard
Open code in new window
EnlighterJS 3 Syntax Highlighter
pip install -r requirements.txt
pip install -r requirements.txt
pip install -r requirements.txt

接下來,在根目錄下新建一個資料夾,命名為 services。在該資料夾中建立一個新檔案 products.py,並新增下面的程式碼來設定 Flask 伺服器。

Plain text
Copy to clipboard
Open code in new window
EnlighterJS 3 Syntax Highlighter
import requests
import os
from flask import Flask, jsonify
app = Flask(__name__)
port = int(os.environ.get('PORT', 5000))
@app.route("/")
def home():
return "Hello, this is a Flask Microservice"
if __name__ == "__main__":
app.run(debug=True, host="0.0.0.0", port=port)
import requests import os from flask import Flask, jsonify app = Flask(__name__) port = int(os.environ.get('PORT', 5000)) @app.route("/") def home(): return "Hello, this is a Flask Microservice" if __name__ == "__main__": app.run(debug=True, host="0.0.0.0", port=port)
import requests
import os
from flask import Flask, jsonify
app = Flask(__name__)
port = int(os.environ.get('PORT', 5000))
@app.route("/")
def home():
return "Hello, this is a Flask Microservice"
if __name__ == "__main__":
app.run(debug=True, host="0.0.0.0", port=port)

在上面的程式碼中,設定了一個基本的 Flask 伺服器。它初始化了一個 Flask 應用程式,為根 URL(” / “)定義了一條路由,並在訪問時顯示 "Hello, this is a Flask Microservice"的訊息。伺服器在指定的埠(從環境變數中獲取或預設為埠 5000 )上執行,並以除錯模式啟動,以便隨時處理傳入的請求。

3. 定義 API 端點

配置好伺服器後,為微服務建立一個 API 端點,以便從公開可用的 API 中獲取產品資料。將此程式碼新增到 products.py 檔案中:

Plain text
Copy to clipboard
Open code in new window
EnlighterJS 3 Syntax Highlighter
BASE_URL = "https://dummyjson.com"
@app.route('/products', methods=['GET'])
def get_products():
response = requests.get(f"{BASE_URL}/products")
if response.status_code != 200:
return jsonify({'error': response.json()['message']}), response.status_code
products = []
for product in response.json()['products']:
product_data = {
'id': product['id'],
'title': product['title'],
'brand': product['brand'],
'price': product['price'],
'description': product['description']
}
products.append(product_data)
return jsonify({'data': products}), 200 if products else 204
BASE_URL = "https://dummyjson.com" @app.route('/products', methods=['GET']) def get_products(): response = requests.get(f"{BASE_URL}/products") if response.status_code != 200: return jsonify({'error': response.json()['message']}), response.status_code products = [] for product in response.json()['products']: product_data = { 'id': product['id'], 'title': product['title'], 'brand': product['brand'], 'price': product['price'], 'description': product['description'] } products.append(product_data) return jsonify({'data': products}), 200 if products else 204
BASE_URL = "https://dummyjson.com"
@app.route('/products', methods=['GET'])
def get_products():
response = requests.get(f"{BASE_URL}/products")
if response.status_code != 200:
return jsonify({'error': response.json()['message']}), response.status_code
products = []
for product in response.json()['products']:
product_data = {
'id': product['id'],
'title': product['title'],
'brand': product['brand'],
'price': product['price'],
'description': product['description']
}
products.append(product_data)
return jsonify({'data': products}), 200 if products else 204

上面的程式碼在 Flask 伺服器中建立了一個 /products 端點。當通過 GET 請求訪問時,它會從一個虛擬應用程式介面獲取產品資料。如果成功,它會處理檢索到的資料,提取產品詳細資訊,並以 JSON 格式返回資訊。如果出現錯誤或沒有可用資料,它會以適當的錯誤訊息和狀態程式碼進行響應。

4. 測試微服務

至此,您已經成功建立了一個簡單的微服務。要啟動服務,請啟動開發伺服器,它將在 http://localhost:5000 開始執行。

Plain text
Copy to clipboard
Open code in new window
EnlighterJS 3 Syntax Highlighter
flask --app services/products run
flask --app services/products run
flask --app services/products run

然後使用 Postman 客戶端向 /products 端點發出 GET 請求,你應該會看到與下面截圖類似的響應。

在 Postman 中測試 HTTP API GET請求

在 Postman 中測試 HTTP API GET請求。

如何在 Python 微服務中實施身份驗證和授權

在構建微服務時,實施強大的安全措施(如身份驗證和授權)非常重要。保證微服務的安全可以確保只有授權使用者才能訪問和使用服務,從而保護敏感資料並防止惡意攻擊。

在微服務中實施安全身份驗證和授權的一種有效方法是 JSON Web 標記(JWT)。

JWT 是一種廣泛使用的開放標準,它提供了一種在客戶端和伺服器之間傳輸身份驗證資訊的安全高效的方法。它們是經過加密和數字簽名的小巧令牌,可與 HTTP 請求一起傳遞。當你在每個請求中包含一個 JWT 時,伺服器就能快速驗證使用者的身份和許可權。

要在微服務中實施 JWT 身份驗證,請執行以下操作:

requirements.txt 檔案中新增 Python 的 pyjwt 軟體包,並使用 pip install -r requirements.txt 重新安裝依賴項。

由於該服務沒有專用資料庫,因此請在專案根目錄下建立 users.json 檔案,以儲存授權使用者列表。將下面的程式碼貼上到檔案中:

Plain text
Copy to clipboard
Open code in new window
EnlighterJS 3 Syntax Highlighter
[
{
"id": 1,
"username": "admin",
"password": "admin"
}
]
[ { "id": 1, "username": "admin", "password": "admin" } ]
[
{   
"id": 1,
"username": "admin",
"password": "admin"
}
]

注:您可以使用我們的資料庫託管服務,為您的微服務輕鬆設定您喜歡的資料庫(PostgreSQLMariaDBRedis 和 MySQL)。

然後,在您的 services/products.py 檔案中,用以下語句替換匯入語句:

Plain text
Copy to clipboard
Open code in new window
EnlighterJS 3 Syntax Highlighter
import requests
from flask import Flask, jsonify, request, make_response
import jwt
from functools import wraps
import json
import os
from jwt.exceptions import DecodeError
import requests from flask import Flask, jsonify, request, make_response import jwt from functools import wraps import json import os from jwt.exceptions import DecodeError
import requests 
from flask import Flask, jsonify, request, make_response
import jwt
from functools import wraps
import json
import os
from jwt.exceptions import DecodeError

匯入這些模組是為了處理 HTTP 請求、建立 Flask 應用程式、管理 JSON 資料、實現基於 JWT 的身份驗證和處理異常,從而在 Flask 伺服器中啟用各種功能。

在 Flask 應用程式例項建立下方新增以下程式碼,以生成用於簽署 JWT 標記的祕鑰。

Plain text
Copy to clipboard
Open code in new window
EnlighterJS 3 Syntax Highlighter
app.config['SECRET_KEY'] = os.urandom(24)
app.config['SECRET_KEY'] = os.urandom(24)
app.config['SECRET_KEY'] = os.urandom(24)

要驗證 JWT,請建立一個裝飾器函式,並在 Flask 伺服器程式碼中的 API 路由上方新增以下程式碼。此裝飾器函式將在使用者訪問受保護路由之前對其進行身份驗證和驗證。

Plain text
Copy to clipboard
Open code in new window
EnlighterJS 3 Syntax Highlighter
def token_required(f):
@wraps(f)
def decorated(*args, **kwargs):
token = request.cookies.get('token')
if not token:
return jsonify({'error': 'Authorization token is missing'}), 401
try:
data = jwt.decode(token, app.config['SECRET_KEY'], algorithms=["HS256"])
current_user_id = data['user_id']
except DecodeError:
return jsonify({'error': 'Authorization token is invalid'}), 401
return f(current_user_id, *args, **kwargs)
return decorated
def token_required(f): @wraps(f) def decorated(*args, **kwargs): token = request.cookies.get('token') if not token: return jsonify({'error': 'Authorization token is missing'}), 401 try: data = jwt.decode(token, app.config['SECRET_KEY'], algorithms=["HS256"]) current_user_id = data['user_id'] except DecodeError: return jsonify({'error': 'Authorization token is invalid'}), 401 return f(current_user_id, *args, **kwargs) return decorated
def token_required(f):
@wraps(f)
def decorated(*args, **kwargs):
token = request.cookies.get('token')
if not token:
return jsonify({'error': 'Authorization token is missing'}), 401
try:
data = jwt.decode(token, app.config['SECRET_KEY'], algorithms=["HS256"])
current_user_id = data['user_id']
except DecodeError:
return jsonify({'error': 'Authorization token is invalid'}), 401
return f(current_user_id, *args, **kwargs)
return decorated

此裝飾器函式會檢查傳入 HTTP 請求中的 JWT 授權令牌,該令牌應存在於請求頭或 cookie 中。如果令牌缺失或無效,裝飾器將 unauthorized status code 傳送未授權狀態程式碼訊息作為響應。

相反,如果存在有效令牌,裝飾器會在解碼後提取使用者 ID。這一過程只允許授權使用者訪問受保護的應用程式介面端點,從而保障了應用程式介面端點的安全。

使用下面的程式碼定義一個用於使用者身份驗證的 API 端點。

Plain text
Copy to clipboard
Open code in new window
EnlighterJS 3 Syntax Highlighter
with open('users.json', 'r') as f:
users = json.load(f)
@app.route('/auth', methods=['POST'])
def authenticate_user():
if request.headers['Content-Type'] != 'application/json':
return jsonify({'error': 'Unsupported Media Type'}), 415
username = request.json.get('username')
password = request.json.get('password')
for user in users:
if user['username'] == username and user['password'] == password:
token = jwt.encode({'user_id': user['id']}, app.config['SECRET_KEY'],algorithm="HS256")
response = make_response(jsonify({'message': 'Authentication successful'}))
response.set_cookie('token', token)
return response, 200
return jsonify({'error': 'Invalid username or password'}), 401
with open('users.json', 'r') as f: users = json.load(f) @app.route('/auth', methods=['POST']) def authenticate_user(): if request.headers['Content-Type'] != 'application/json': return jsonify({'error': 'Unsupported Media Type'}), 415 username = request.json.get('username') password = request.json.get('password') for user in users: if user['username'] == username and user['password'] == password: token = jwt.encode({'user_id': user['id']}, app.config['SECRET_KEY'],algorithm="HS256") response = make_response(jsonify({'message': 'Authentication successful'})) response.set_cookie('token', token) return response, 200 return jsonify({'error': 'Invalid username or password'}), 401
with open('users.json', 'r') as f:
users = json.load(f)
@app.route('/auth', methods=['POST'])
def authenticate_user():
if request.headers['Content-Type'] != 'application/json':
return jsonify({'error': 'Unsupported Media Type'}), 415
username = request.json.get('username')
password = request.json.get('password')
for user in users:
if user['username'] == username and user['password'] == password:
token = jwt.encode({'user_id': user['id']}, app.config['SECRET_KEY'],algorithm="HS256")
response = make_response(jsonify({'message': 'Authentication successful'}))
response.set_cookie('token', token)
return response, 200
return jsonify({'error': 'Invalid username or password'}), 401

為了驗證和授權使用者, /auth API 端點會根據允許的使用者列表檢查 POST 請求的 JSON 有效負載中的憑據。如果憑據有效,它就會使用使用者 ID 和應用程式祕鑰生成一個 JWT 令牌,並在響應中將令牌設定為 cookie。現在,使用者可以使用此令牌進行後續的 API 請求。

建立 /auth 端點後,使用 Postman 向 http://localhost:5000/auth 傳送 HTTP POST 請求。在請求正文中,包含你建立的模擬管理員使用者的憑據。

Postman 請求顯示請求正文

Postman 請求顯示請求正文。

如果請求成功,API 將生成一個 JWT 令牌,將其設定在 Postman 的 cookies 中,併傳送驗證成功的響應。

最後,更新 GET API 端點,使用下面的程式碼檢查並驗證 JWT 標記:

Plain text
Copy to clipboard
Open code in new window
EnlighterJS 3 Syntax Highlighter
@app.route('/products', methods=['GET'])
@token_required
def get_products(current_user_id):
headers = {'Authorization': f'Bearer {request.cookies.get("token")}'}
response = requests.get(f"{BASE_URL}/products", headers=headers)
if response.status_code != 200:
return jsonify({'error': response.json()['message']}), response.status_code
products = []
for product in response.json()['products']:
product_data = {
'id': product['id'],
'title': product['title'],
'brand': product['brand'],
'price': product['price'],
'description': product['description']
}
products.append(product_data)
return jsonify({'data': products}), 200 if products else 204
@app.route('/products', methods=['GET']) @token_required def get_products(current_user_id): headers = {'Authorization': f'Bearer {request.cookies.get("token")}'} response = requests.get(f"{BASE_URL}/products", headers=headers) if response.status_code != 200: return jsonify({'error': response.json()['message']}), response.status_code products = [] for product in response.json()['products']: product_data = { 'id': product['id'], 'title': product['title'], 'brand': product['brand'], 'price': product['price'], 'description': product['description'] } products.append(product_data) return jsonify({'data': products}), 200 if products else 204
@app.route('/products', methods=['GET'])
@token_required
def get_products(current_user_id):
headers = {'Authorization': f'Bearer {request.cookies.get("token")}'}    
response = requests.get(f"{BASE_URL}/products", headers=headers)
if response.status_code != 200:
return jsonify({'error': response.json()['message']}), response.status_code
products = []
for product in response.json()['products']:
product_data = {
'id': product['id'],
'title': product['title'],
'brand': product['brand'],
'price': product['price'],
'description': product['description']
}
products.append(product_data)
return jsonify({'data': products}), 200 if products else 204

如何使用 Docker 將 Python 微服務容器化

Docker 是一個將應用程式及其依賴關係打包到一個隔離的開發環境中的平臺。將微服務打包到容器中可簡化其在伺服器中的部署和管理流程,因為每個服務都在其容器中獨立執行和執行。

要將微服務容器化,必須從指定在容器中執行應用程式所需的依賴項的 Docker 檔案建立 Docker 映像。在專案根目錄中建立一個 Dockerfile,並新增以下說明:

Plain text
Copy to clipboard
Open code in new window
EnlighterJS 3 Syntax Highlighter
FROM python:3.9-alpine
WORKDIR /app
COPY requirements.txt ./
RUN pip install -r requirements.txt
COPY . .
EXPOSE 5000
CMD ["python", "./services/products.py"]
FROM python:3.9-alpine WORKDIR /app COPY requirements.txt ./ RUN pip install -r requirements.txt COPY . . EXPOSE 5000 CMD ["python", "./services/products.py"]
FROM python:3.9-alpine
WORKDIR /app
COPY requirements.txt ./
RUN pip install -r requirements.txt
COPY . .
EXPOSE 5000
CMD ["python", "./services/products.py"]

在構建映象之前,請檢視這些命令:

  • FROM – 指示 Docker 使用哪個基礎映象。基礎映象是預先構建的例項,包含在容器中執行 Flask 應用程式所需的軟體和依賴項。
  • WORKDIR – 將容器中的指定目錄設定為工作目錄。
  • COPY requirements.txt ./ – 將 requirements.txt 檔案中的依賴項複製到容器的 requirements.txt 檔案中。
  • RUN – 執行指定命令以安裝映像所需的依賴項。
  • COPY . . – 將專案根目錄中的所有檔案複製到容器內的工作目錄。
  • EXPOSE – 指定容器監聽請求的埠。不過,Docker 不會向主機發布該埠。
  • CMD – 指定容器啟動時要執行的預設命令。

接下來,在專案根目錄中新增 .dockerignore 檔案,指定 Docker 映象應排除的檔案。限制映像內容將減少其最終大小和相關的構建時間。

Plain text
Copy to clipboard
Open code in new window
EnlighterJS 3 Syntax Highlighter
/venv
/services/__pycache__/
.gitignore
/venv /services/__pycache__/ .gitignore
/venv
/services/__pycache__/
.gitignore

現在,執行下面的命令來構建 Docker 映象:

Plain text
Copy to clipboard
Open code in new window
EnlighterJS 3 Syntax Highlighter
docker build -t flask-microservice .
docker build -t flask-microservice .
docker build -t flask-microservice .

最後,映象構建完成後,就可以使用以下命令在 Docker 容器中執行微服務了:

Plain text
Copy to clipboard
Open code in new window
EnlighterJS 3 Syntax Highlighter
docker run -p 5000:5000 flask-microservice
docker run -p 5000:5000 flask-microservice
docker run -p 5000:5000 flask-microservice

該命令將啟動一個執行微服務的 Docker 容器,並將容器上的 5000 埠暴露給主機上的 5000 埠,這樣就可以使用 URL http://localhost:5000 從網路瀏覽器或 Postman 發出 HTTP 請求。

使用 Kinsta 部署 Python 微服務

Kinsta 為網路應用程式和資料庫提供託管解決方案–您可以在生產環境中無縫部署和管理您的 Python 微服務和後端 API。

請按照以下步驟配置您的 Flask 微服務,以便使用 MyKinsta 進行部署:

首先,在根目錄中建立一個新的 Procfile,並新增以下程式碼。它指定了在 Kinsta 的 Gunicorn WSGI HTTP 伺服器(適用於 Python 應用程式)上執行 Flask 微服務的命令。

Plain text
Copy to clipboard
Open code in new window
EnlighterJS 3 Syntax Highlighter
web: gunicorn services.wsgi
web: gunicorn services.wsgi
web: gunicorn services.wsgi

requirements.txt 檔案中,新增 Gunicorn 依賴項:

Plain text
Copy to clipboard
Open code in new window
EnlighterJS 3 Syntax Highlighter
gunicorn==20.1.*
gunicorn==20.1.*
gunicorn==20.1.*

接下來,建立一個新的 services/wsgi.py 檔案並新增以下程式碼。

Plain text
Copy to clipboard
Open code in new window
EnlighterJS 3 Syntax Highlighter
from services.products import app as application
if __name__ == "__main__":
application.run()
from services.products import app as application if __name__ == "__main__": application.run()
from services.products import app as application
if __name__ == "__main__":
application.run()

在專案根資料夾中建立 .gitignore 檔案,並新增以下內容:

Plain text
Copy to clipboard
Open code in new window
EnlighterJS 3 Syntax Highlighter
services/__pycache__
venv
services/__pycache__ venv
services/__pycache__
venv

最後,在 GitHub 上建立一個新倉庫並推送專案檔案。

倉庫準備就緒後,按照以下步驟將 Flask 微服務部署到 Kinsta:

  1. 登入或建立賬戶,檢視 MyKinsta 面板。
  2. 使用 Git 提供商(Bitbucket、GitHub 或 GitLab)授權 Kinsta。
  3. 單擊左側邊欄上的 “Applications“,然後單擊 “Add application“。
  4. 在儀表板上點選 Add Service,然後選擇 Application
  5. 選擇要部署的版本庫和分支。
  6. 為應用程式指定一個唯一的名稱,並選擇一個 data center location
  7. 要配置 Build 環境,請選擇使用 Dockerfile 構建容器映像的選項。
  8. 提供 Dockerfile 的路徑和上下文。
  9. 最後,檢視其他資訊並單擊 “Create application“。

測試微服務

部署過程成功後,點選提供的 URL,在 Postman 中發出 HTTP 請求,測試微服務。向根端點發出 GET 請求。

對微服務 product 端點的 HTTP API GET 請求

對微服務 product 端點的 HTTP API GET 請求。

要進行身份驗證並生成 JWT 令牌,請向 /auth API 端點傳送 POST 請求,並在請求正文中傳遞管理員憑據。

HTTP API POST 請求到微服務 auth 端點

HTTP API POST 請求到微服務 auth 端點。

最後,成功通過身份驗證後,向 /products 端點發出 GET 請求以獲取資料。

對微服務 products 端點的 HTTP API GET 請求

對微服務 products 端點的 HTTP API GET 請求。

小結

隨著應用程式的規模和複雜性不斷增加,採用能讓軟體系統在不佔用可用資源的情況下進行擴充套件的架構模式至關重要。

微服務架構具有可擴充套件性、開發靈活性和可維護性,使您能更輕鬆地管理複雜的應用程式。

評論留言