shenyu-client-python 1.0.1

基于shenyu 2.7.0.2的python客户端

注册方式

1. 全路径注册（默认）

register_all_paths=True（默认）时，向 Admin 注册 context_path/**，网关代理整个上下文下所有 HTTP 路径。

2. 按接口注册（注解）

将 register_all_paths 设为 False，在需要暴露的接口上使用 @register_metadata（别名 @shenyu_client），启动时由 SHENYU_CONFIG.register() 批量上报：

from shenyu_register.annotations import register_metadata

SHENYU_CONFIG.register_all_paths = False

@register_metadata("/health", path_desc="健康检查")
@app.get("/health")
def health():
    return {"status": "ok"}

装饰器写在路由装饰器内侧（更靠近 def），与 Flask 用法一致。path 可为相对 context_path 的路径（如 /health），注册时会拼成 /python-demo/health。

也可用代码声明：SHENYU_CONFIG.add_endpoint("/health", path_desc="健康检查")。

---

import logging
import os
from contextlib import asynccontextmanager

import uvicorn
from fastapi import FastAPI

from shenyu_register.config import SHENYU_CONFIG

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

HTTP_PORT = int(os.getenv("PORT", os.getenv("UVICORN_PORT", "8000")))

# ShenYu：在代码中改配置，例如 enabled / server_lists / app_name 等
SHENYU_CONFIG.enabled = True
SHENYU_CONFIG.app_name = "python-demo"
SHENYU_CONFIG.context_path = "/python-demo"
SHENYU_CONFIG.server_lists = "http://127.0.0.1:9095"
SHENYU_CONFIG.heartbeat_interval_seconds = 10
SHENYU_CONFIG.admin_username = "admin"
SHENYU_CONFIG.admin_password = "123456"


@asynccontextmanager
async def lifespan(app: FastAPI):
    """
    - 启动阶段 ``register`` 若抛错：记录日志后向上抛出，**不会**执行 ``yield``，应用不完成启动；此时未成功注册，**不需要** ``shutdown``。
    - 启动成功并 ``yield`` 之后：无论之后是正常停服还是其它原因结束 lifespan，都会在 ``finally`` 里调用 ``shutdown``（离线上报）。
    - ``shutdown`` 内部若再抛错：在 ``finally`` 里捕获并记日志，**避免**掩盖其它清理逻辑或影响进程退出。
    """
    try:
        SHENYU_CONFIG.register(service_port=HTTP_PORT)
    except Exception:
        logger.exception("ShenYu 启动注册失败，应用将不会完成启动")
        raise
    try:
        yield
    finally:
        try:
            SHENYU_CONFIG.shutdown()
        except Exception:
            logger.exception("ShenYu shutdown（心跳停止 / 离线上报）失败，可在 Admin 侧检查或手工下线实例")


app = FastAPI(title="Shenyu Client Demo", version="0.1.0", lifespan=lifespan)

@app.get("/")
def read_root():
    return {"message": "Hello, FastAPI"}


@app.get("/health")
def health():
    return {"status": "ok"}


if __name__ == "__main__":
    uvicorn.run("example:app", host="0.0.0.0", port=HTTP_PORT, reload=False)