Pythonを使用してNacos Configuration Centerを制御する方法を教えます

みなさんこんにちは、An Guoです！

Nacos は Alibaba のオープンソース プロジェクトであり、クラウド ネイティブ アプリケーション向けの動的なサービス検出、構成管理、サービス管理プラットフォームの構築に使用されます。

コア機能には、サービス検出、サービスヘルス監視、動的構成サービス、動的DNSサービス、サービスおよびメタデータ管理が含まれます。

Python プロジェクトで Nacos 動的構成サービスを使用する場合はどうすればよいですか?

1. インストールの依存関係

nacos-sdk-python プロジェクトは、Nacos OpenAPI の Python 実装であり、Nacos 構成ファイルのデータの変更を監視するために使用できます。

 # 安装以来pip3 install nacos-sdk-python # Nacos配置文件为yaml的依赖pip3 install pyyaml

プロジェクトアドレス: https://github.com/nacos-group/nacos-sdk-python

2. 基本的な使い方（yaml）

YAML設定ファイルを例に挙げます

まず、Nacos接続情報（接続情報、名前空間、ユーザー名、パスワード）を使用してNacosクライアント接続オブジェクトを作成します。

 import nacos # 连接地址SERVER_ADDRESSES = "192.*.*.*" SERVER_PORT = '8848' # 命名空间NAMESPACE = "public" # 账号信息USERNAME = 'nacos' PASSWORD = 'nacos' # 创建一个连接对象client = nacos.NacosClient(f'{SERVER_ADDRESSES}:{SERVER_PORT}', namespace=NAMESPACE, username=USERNAME, password=PASSWORD

次に、グループ名とサービスIDでサービスの構成を解析し、YAML形式でデータを解析します。

 import yaml # 初始化def init(data_id, group): config = client.get_config(data_id, group) # 配置数据解析（YAML） config_data = yaml.load(config, Loader=yaml.FullLoader) # 通过键路径，解析出数据result = config_data['arg1']['arg2'] print(result) # 服务id（键） data_id = "service_name" # 分组名称，默认为：DEFAULT_GROUP group = "DEFAULT_GROUP" # 初始化解析init(data_id, group)

最後に、グループ名とサービス ID、および Nacos 接続情報を使用してリスナー イベントを追加し、Nacos 構成が変更されたときにプログラムが変更されたデータを適時に取得できるようにします。

 # Nacos数据变动时触发def nacos_data_change_callback(config): # 数据解析nacos_data = yaml.load(config['content'], Loader=yaml.FullLoader) # 读取键值result = nacos_data['arg1']['arg2'] print(result) # 监听Nacos数据变动def add_nacos_listener(data_id, group): client.add_config_watcher(data_id=data_id, group=group, cb=nacos_data_change_callback) # 添加监听事件add_nacos_listener(data_id, group)

3. プロパティファイル

YAML設定ファイルとの違いは

• YAMLはインデントとコロンを使用して階層を示す
• プロパティは等号を使用してキーと値のペアを接続します

Nacos 構成ファイルを監視する機能では、解析ロジックを変更するだけで済みます。

 import nacos # 解析Properties配置文件（Nacos） # 初始化def init(data_id, group): # 换行符进行分割，存入列表中config_list = client.get_config(data_id, group).split("\n") properties = {} for config_item in config_list: # 过滤有用的键值对if config_item.find('=') > 0: strs = config_item.replace('\n', '').split('=') properties[strs[0]] = strs[1] # 配置的地址address = properties['address'] print(address) # Nacos数据变动时触发def nacos_data_change_callback(config): config_list = config['content'].split("\n") properties = {} for config_item in config_list: # 过滤有用部分if config_item.find('=') > 0: strs = config_item.replace('\n', '').split('=') properties[strs[0]] = strs[1] # 配置的地址address = properties['address'] print("Nacos数据变动了，address：", address)

4. Python Web + ナコス

Python Web アプリケーションで Nacos の動的構成を組み合わせる場合は、以下の手順に従う必要があります。

ここではFastAPIを例に挙げて説明します

まず、Nacosクライアント接続オブジェクトとグローバル変数を定義します。

PS: グローバル変数はテストデモンストレーションに使用されます

import nacos client = nacos.NacosClient(f'{SERVER_ADDRESSES}:{SERVER_PORT}', namespace=NAMESPACE, username=USERNAME, password=PASSWORD) # 定义一个全局变量arg1 = ''

次に、FastAPIオブジェクトをインスタンス化し、アプリケーションの起動時にasyncioを使用してリスナーイベントを作成します。

 from fastapi import FastAPI import nacos import uvicorn import asyncio app = FastAPI() # 运行时触发@app.on_event("startup") async def startup_event(): asyncio.create_task(event_listener()) if __name__ == '__main__': uvicorn.run("demo_fastapi:app", host="0.0.0.0", port=8000, reload=True)

監視イベントでは、グループ名とサービスIDを使用してデータを初期化および監視します。

 # Nacos初始化async def init(data_id, group): global arg1 # 换行符进行分割，存入列表中config_list = client.get_config(data_id, group).split("\n") ... # 配置的地址arg1 = properties['address'] print("arg1:", arg1) # Nacos数据变动时触发def nacos_data_change_callback(config): global arg1 config_list = config['content'].split("\n") ... # 配置的地址arg1 = properties['address'] print("arg1:", arg1) async def event_listener(): data_id = "service_name" group = "DEFAULT_GROUP" # 初始化await init(data_id, group) # Nacos配置监听，用于数据变动监听client.add_config_watcher(data_id=data_id, group=group, cb=nacos_data_change_callback)

最後に、変数の値を取得するための簡単なインターフェースを定義します。

 # 定义一个全局变量arg1 = '' @app.get("/") async def index(): global arg1 return {"message": arg1}

これにより、Nacos 設定ファイル内のデータが変更された場合、インターフェースを通じて最新のデータをリアルタイムで取得できるようになります。

5. 落とし穴を防ぐ

nacos-sdk-python プロジェクトで説明されているように、作者はせいぜい Python 3.7 および Nacos 1.3.2 との互換性しか実現しませんでした。

実際のテストでは、プログラムはWindowsでは正常に動作しましたが、MacまたはLinuxに配置すると、TypeError：'_thread.RLock'オブジェクトをピクルできませんというエラーが発生していることがわかりました。

ここでは、ソースコード nacos/clinet.py を書き直して Windows 以外のシステムと互換性を持たせ、RLock を使用して実装する必要があります。

PS: 記事の最後にあるキーワードを送信してソースコードを取得し、直接置き換えることができます。

詳細は次号をご覧ください

https://github.com/nacos-group/nacos-sdk-python/pull/125