shenyu-client-python 1.0.0

基于shenyu 2.7.0.2的python客户端

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)