4、基础API框架搭建:Flask项目结构设计、蓝图(Blueprint)模块化、配置管理、日志系统集成。
好,咱们直接进入正题。这一章聊的是API框架的「地基」怎么打。你想想看,一个运动控制API,少说几十个接口,多则上百。如果全塞在一个文件里,那维护起来简直是噩梦。我早年接手过一个项目,一个app.py文件三千多行,改个bug得翻半天——嗯,那滋味,谁经历谁知道。
4.1 项目结构设计:别让代码长成「毛线球」
我个人习惯,项目一开始就要把目录结构定死。结构清晰了,后面加功能才不会打架。下面是我常用的一个模板,专门针对运动控制API场景:
motion_control_api/
├── app/
│ ├── __init__.py # 应用工厂
│ ├── config.py # 配置管理
│ ├── extensions.py # 扩展初始化
│ ├── blueprints/
│ │ ├── __init__.py
│ │ ├── axis/ # 轴控制模块
│ │ │ ├── __init__.py
│ │ │ ├── routes.py # 路由定义
│ │ │ └── schemas.py # 请求/响应模型
│ │ ├── io/ # IO控制模块
│ │ │ ├── __init__.py
│ │ │ ├── routes.py
│ │ │ └── schemas.py
│ │ └── system/ # 系统管理模块
│ │ ├── __init__.py
│ │ ├── routes.py
│ │ └── schemas.py
│ ├── services/ # 业务逻辑层
│ │ ├── axis_service.py
│ │ └── io_service.py
│ ├── models/ # 数据模型
│ │ └── device.py
│ └── utils/ # 工具函数
│ ├── logger.py # 日志配置
│ └── response.py # 统一响应格式
├── logs/ # 日志文件目录
├── tests/ # 单元测试
├── requirements.txt
└── run.py # 启动入口
这个结构有几个好处:按功能模块切分,轴归轴,IO归IO,系统归系统。每个模块内部再分路由、数据校验、业务逻辑。我在项目中遇到过,如果一开始不分层,后面加一个「回零」功能,可能要改三个文件,还容易漏掉。
核心原则:「高内聚,低耦合」。每个蓝图只关心自己的事,别跨模块引用路由。服务层(services)负责真正的运动控制逻辑,路由层只做参数校验和响应封装。
4.2 蓝图(Blueprint)模块化:给API「分房间」
蓝图是什么?说白了,就是给Flask应用划分「功能区」。每个蓝图可以有自己的路由、错误处理器、模板。对于运动控制API,我通常按物理设备或功能域来划分蓝图。
举个例子,轴控制蓝图的定义:
# app/blueprints/axis/__init__.py
from flask import Blueprint
axis_bp = Blueprint('axis', __name__, url_prefix='/api/v1/axis')
from . import routes # 导入路由模块
# app/blueprints/axis/routes.py
from . import axis_bp
from flask import request, jsonify
@axis_bp.route('/move', methods=['POST'])
def move_axis():
"""轴运动控制接口"""
data = request.get_json()
axis_id = data.get('axis_id')
position = data.get('position')
speed = data.get('speed', 100)
# 这里调用服务层
# result = axis_service.move_to(axis_id, position, speed)
return jsonify({'status': 'ok', 'message': f'Axis {axis_id} moving to {position}'})
@axis_bp.route('/status', methods=['GET'])
def axis_status():
"""获取轴状态"""
axis_id = request.args.get('axis_id')
# 实际项目中会查询硬件状态
return jsonify({'axis_id': axis_id, 'status': 'idle', 'position': 0})
然后在应用工厂里注册蓝图:
# app/__init__.py
from flask import Flask
from .blueprints.axis import axis_bp
from .blueprints.io import io_bp
from .blueprints.system import system_bp
def create_app():
app = Flask(__name__)
# 注册蓝图
app.register_blueprint(axis_bp)
app.register_blueprint(io_bp)
app.register_blueprint(system_bp)
return app
我的小技巧:每个蓝图都加上 url_prefix,这样路由天然带版本号。比如 /api/v1/axis/move,以后升级到v2,直接新建一个蓝图,旧接口还能继续用。我曾经因为没加版本前缀,升级时被迫停服,那叫一个狼狈。
4.3 配置管理:别把密码写死在代码里
配置管理这件事,看着简单,但坑不少。我见过有人把数据库密码直接写在config.py里,然后上传到GitHub——嗯,后果你懂的。
我推荐用「分层配置」模式:
| 配置层级 | 来源 | 适用场景 |
|---|---|---|
| 默认配置 | 代码内硬编码 | 通用参数,如日志格式 |
| 环境变量 | 操作系统环境 | 敏感信息,如密码、密钥 |
| 配置文件 | config.yaml / .env | 环境差异,如IP、端口 |
| 运行时参数 | 启动命令行 | 临时调试,如日志级别 |
具体实现:
# app/config.py
import os
from dotenv import load_dotenv
load_dotenv() # 加载 .env 文件
class Config:
"""基础配置"""
SECRET_KEY = os.getenv('SECRET_KEY', 'dev-secret-key')
DEBUG = False
LOG_LEVEL = os.getenv('LOG_LEVEL', 'INFO')
# 运动控制相关
MOTION_TIMEOUT = int(os.getenv('MOTION_TIMEOUT', '5000')) # 毫秒
AXIS_COUNT = int(os.getenv('AXIS_COUNT', '4'))
DEFAULT_SPEED = int(os.getenv('DEFAULT_SPEED', '100'))
# 日志配置
LOG_DIR = os.getenv('LOG_DIR', './logs')
LOG_MAX_BYTES = 10 * 1024 * 1024 # 10MB轮转
class DevelopmentConfig(Config):
DEBUG = True
LOG_LEVEL = 'DEBUG'
class ProductionConfig(Config):
DEBUG = False
LOG_LEVEL = 'WARNING'
注意:.env文件永远不要提交到Git仓库!在.gitignore里加上 .env。我习惯在项目里放一个 .env.example,把需要的变量名列出来,值留空,方便新同事快速上手。
4.4 日志系统集成:出了事得有据可查
运动控制API最怕什么?怕设备跑飞了,你连日志都找不到。我经历过一次,客户说轴撞了限位,我查了半天日志,发现根本没记录——因为日志级别设成了ERROR,而报警是WARNING级别。从那以后,我对日志系统格外上心。
一个靠谱的日志系统,至少要做到三点:
- 分级记录:DEBUG、INFO、WARNING、ERROR,一个不能少
- 自动轮转:日志文件不能无限增长,否则磁盘会爆
- 上下文关联:每条日志要能追溯到是哪个请求、哪个设备
看代码:
# app/utils/logger.py
import logging
import os
from logging.handlers import RotatingFileHandler
from flask import request, g
def setup_logger(app):
"""配置日志系统"""
log_dir = app.config.get('LOG_DIR', './logs')
os.makedirs(log_dir, exist_ok=True)
# 根日志记录器
logger = logging.getLogger('motion_api')
logger.setLevel(app.config.get('LOG_LEVEL', 'INFO'))
# 文件处理器 - 带轮转
file_handler = RotatingFileHandler(
os.path.join(log_dir, 'api.log'),
maxBytes=app.config.get('LOG_MAX_BYTES', 10*1024*1024),
backupCount=5,
encoding='utf-8'
)
file_handler.setLevel(logging.DEBUG)
# 控制台处理器
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.INFO)
# 格式化器 - 带上请求ID
formatter = logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - '
'[%(request_id)s] - %(message)s'
)
file_handler.setFormatter(formatter)
console_handler.setFormatter(formatter)
logger.addHandler(file_handler)
logger.addHandler(console_handler)
# 在请求开始时注入request_id
@app.before_request
def inject_request_id():
g.request_id = request.headers.get('X-Request-ID', 'N/A')
return logger
然后在路由里这样用:
# 在路由中使用
from flask import current_app
@axis_bp.route('/move', methods=['POST'])
def move_axis():
logger = current_app.logger
data = request.get_json()
axis_id = data.get('axis_id')
logger.info(f'收到轴运动请求: axis={axis_id}, target={data.get("position")}')
try:
# 执行运动控制
# ...
logger.info(f'轴 {axis_id} 运动完成')
return jsonify({'status': 'ok'})
except Exception as e:
logger.error(f'轴 {axis_id} 运动失败: {str(e)}', exc_info=True)
return jsonify({'status': 'error', 'message': str(e)}), 500
避坑指南:我曾经在生产环境忘记加 exc_info=True,结果异常日志只有一句话「Something went wrong」,根本没法排查。记住,记录异常时一定要带上堆栈信息,否则等于没记。
4.5 知识体系总览
下面这张图,把这一章的核心逻辑串起来了。你可以把它当作搭建API框架的「施工图」:
你看,整个框架就是「一个工厂,三个核心」。应用工厂负责组装,蓝图管路由,配置管参数,日志管追踪。这三件事做好了,后面加接口、改逻辑、查问题,都会顺手很多。
嗯,这一章就到这里。代码都在上面了,你可以直接拿去用。记住,框架搭得稳,后面才能跑得快。