Python 微服务实战:一套 10 服务的 ERP 系统是怎么搭出来的?

class AuthVerifyMiddleware(BaseHTTPMiddleware):    async def dispatch(self, request, call_next):        path = request.url.path        # 白名单直接放行（登录/验证码/刷新Token）        for white_path in config.whitelist_paths:            if path.startswith(white_path):                return await call_next(request)        # 验证 Token        token = request.headers.get("Authorization", "")        payload = JwtUtil.verify_token(token)        if not payload:            return ResponseUtil.unauthorized(msg="token无效或已过期")        # 用户信息注入 request.state        request.state.user = payload        return await call_next(request)

class RequestLimiterMiddleware(BaseHTTPMiddleware):    async def dispatch(self, request, call_next):        # 从 URL 提取服务名        # /api/system-service/user/list → system-service        service_name = parts[1]        # 从 Consul KV 动态读取限流规则        consul_key = f"config/gateway/limit/{service_name}"        limit_rule = ConsulUtil.get_config_kv(consul_key)  # "60/minute"        # 按 IP + 服务名限流        cache_key = f"{client_ip}:{service_name}"        if len(REQUEST_RECORD[cache_key]) >= max_count:            return ResponseUtil.too_many_requests()

@router.api_route("/{service_name}/{path:path}", methods=["GET","POST","PUT","DELETE"])async def proxy_request(request, service_name, path):    # 从 Consul 获取服务实例    service_map = ConsulUtil.get_all_services_with_instances()    instances = service_map[service_name]    # 构建目标 URL    instance = instances[0]    target_url = f"http://{instance['address']}:{instance['port']}/{path}"    # 流式转发（支持大文件/SSE）    resp = await client.send(req, stream=True)    return StreamingResponse(resp.aiter_raw(), ...)

@asynccontextmanagerasync def lifespan(app: FastAPI):    # 启动：注册到 Consul    ConsulUtil.register_service(        name="system-service",        address=None,  # 自动获取本地IP        port=8001,        service_id="system-service-001"    )    yield    # 关停：从 Consul 注销    ConsulUtil.deregister_service("system-service", "system-service-001")

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmakerasync_engine = create_async_engine(    url="mysql+asyncmy://user:pass@host:3306/xeerp?charset=utf8mb4",    pool_size=50,           # 连接池大小    max_overflow=10,        # 允许溢出连接数    pool_recycle=3600,      # 连接回收周期（防止 MySQL 8小时断开）    pool_pre_ping=True,     # 每次使用前健康检查)AsyncSessionLocal = async_sessionmaker(    bind=async_engine,    expire_on_commit=False,  # 关键：提交后对象不失效)

class Base(AsyncAttrs, DeclarativeBase):    """所有 Model 的基类"""    passclass SysUser(Base):    __tablename__ = 'sys_user'    user_id = Column(BigInteger, primary_key=True)    user_name = Column(String(64), nullable=False)    dept_id = Column(BigInteger)    status = Column(CHAR(1), default='0')    create_time = Column(DateTime, default=datetime.now)

@CheckPerm("system:user:add")  # 装饰器声明所需权限async def add_user(request, user):    ...# 装饰器实现def CheckPerm(perm: str):    def decorator(func):        async def wrapper(*args, **kwargs):            user = kwargs.get("user")            # 超级管理员直接通过            if "*" in user.perms:                return await func(*args, **kwargs)            # 检查权限标识            if perm not in user.perms:                raise ServiceException(f"无权限：{perm}", 403)            return await func(*args, **kwargs)        return wrapper    return decorator

class DataPermission:    @staticmethod    def get_filter(model, user):        scope = user.data_scope        # 1=全部 2=本部门及子级 3=本部门 4=本人 5=自定义        if scope == 1: return True        if scope == 4: return model.create_user_id == user.user_id        if scope == 3: return model.create_dept_id == user.dept_id        if scope == 5: return model.create_dept_id.in_(user.custom_dept_ids)

class BaseClient:    """所有服务客户端的基类"""    def __init__(self, base_url: str, timeout: int = 30):        self.base_url = base_url        self.timeout = aiohttp.ClientTimeout(total=timeout)    async def _request(self, method, path, params=None, data=None):        session = await self.get_session()        url = f"{self.base_url}/{path.lstrip('/')}"        async with session.request(method, url, params=params, json=data) as resp:            return resp.status, await resp.json()    async def get(self, path, **kwargs):        return await self._request("GET", path, **kwargs)    async def post(self, path, **kwargs):        return await self._request("POST", path, **kwargs)

class StockClient(BaseClient):    def __init__(self):        super().__init__(base_url="http://127.0.0.1:8003")    async def get_inventory(self, product_id: int):        return await self.get(f"/api/stock/inventory/{product_id}")

class ResponseUtil:    @classmethod    def success(cls, msg="操作成功", data=None, rows=None, total=None):        return JSONResponse(content={            "code": 200,            "msg": msg,            "success": True,            "time": datetime.now().isoformat(),            "data": data,            "rows": rows,            "total": total        })    @classmethod    def fail(cls, msg="操作失败", code=500): ...    @classmethod    def unauthorized(cls, msg="请先登录"): ...    @classmethod    def too_many_requests(cls, msg="请求频繁"): ...

# 业务异常 → 400app.add_exception_handler(ServiceException, biz_exception_handler)# HTTP 异常 → 对应状态码app.add_exception_handler(HTTPException, http_exception_handler)# 参数校验 → 400app.add_exception_handler(RequestValidationError, validation_exception_handler)# 兜底异常 → 500app.add_exception_handler(Exception, global_exception_handler)

from apscheduler.schedulers.asyncio import AsyncIOSchedulerfrom apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore# 任务存储到数据库（重启不丢失）job_stores = {    'default': SQLAlchemyJobStore(engine=sync_engine)}scheduler = AsyncIOScheduler(    jobstores=job_stores,    executors={'default': AsyncIOExecutor()},)

# 自定义 Cron 支持 6/7 位class MyCronTrigger(CronTrigger):    @classmethod    def from_crontab(cls, expr, timezone=None):        values = expr.split()        # 5位=标准 / 6位=带秒 / 7位=带秒+年        if len(values) == 5:            # 秒 分 时 日 月 周            ...

# 服务启动时，一次性加载高频数据到 Redisawait RedisUtil.init_sys_dict(redis)    # 字典预热await RedisUtil.init_sys_config(redis)  # 参数预热await RedisUtil.init_sys_user(redis)    # 在线用户恢复

service/├── api/v1/             # Controller 层：路由定义 + 参数校验│   ├── user_controller.py│   └── dept_controller.py├── service/            # Service 层：业务编排 + 事务│   ├── user_service.py│   └── dept_service.py├── dao/                # DAO 层：数据库操作封装│   ├── user_dao.py│   └── dept_dao.py├── models/             # Model 层：ORM 实体├── schemas/            # Schema 层：Pydantic 校验模型├── clients/            # Client 层：跨服务调用├── core/               # 本服务私有配置└── middlewares/        # 本服务中间件

app = FastAPI(json_serializer=orjson.dumps)

if os.name == "posix":    LOOP = "uvloop"       # 替换默认 asyncio 事件循环    HTTP = "httptools"    # 替换默认 HTTP 解析器

# 50 个常驻连接 + 10 个溢出连接pool_size=50, max_overflow=10# 连接回收（防 MySQL wait_timeout 断开）pool_recycle=3600# 使用前探活（避免拿到死连接）pool_pre_ping=True