技术栈Python Flask Vue 3 TypeScript MySQL Redis --- ## 前言 大家好今天想和大家分享一下开发一个完整企业级 SaaS 平台的全过程。从需求分析、技术选型、架构设计到代码实现全流程的踩坑经验和心得体会都在这里了。 先上一张项目功能全景图 | 模块 | 功能说明 | |---|---| | AI 智能助手 | AI 对话、智能报表生成、项目 AI 分析、任务智能拆解 | | 项目管理 | 项目全生命周期管理、任务跟踪、进度监控、风险预警 | | 客户管理CRM | 客户档案、商机跟进、销售订单、回款管理 | | 进销存 | 商品管理、库存管理、采购管理、仓库管理、库存盘点 | | 物料管理BOM | 物料档案、BOM 结构、生产工单、领料管理 | | 模具管理 | 模具档案、出入库、生产计划 | | 财务管理 | 账户管理、收支流水、付款单、报销单、资金调拨 | | 考勤管理 | 员工考勤记录、考勤报表 | | ⚙️ 系统设置 | 企业信息、用户管理、角色权限、部门管理、数据字典 | --- ## 一、技术选型与架构设计 ### 1.1 为什么选 Flask Vue 3 选择技术栈时我考虑了以下几个因素 | 维度 | 选择 | 理由 | |---|---|---| | 后端框架 | **Flask** | 轻量、灵活、插件生态丰富足够承载中小 SaaS 的业务量 | | 前端框架 | **Vue 3 TypeScript** | API 简洁学习成本低Composition API 非常适合大型项目 | | 数据库 | **MySQL** | 成熟稳定关系型数据适合企业业务 | | 缓存 | **Redis** | 验证码、Session、速率限制 | | ORM | **SQLAlchemy** | Python 生态最成熟的 ORM支持复杂查询 | | 认证 | **JWT Session Cookie** | 无状态 HttpOnly 安全存储 | ### 1.2 整体架构图 ┌──────────────────────────────────────────────────────┐ │ 前端 Vue 3 │ │ 路由管理 │ 状态管理(Pinia) │ API 请求封装 │ └────────────────────┬─────────────────────────────────┘ │ HTTPS CORS ┌────────────────────▼─────────────────────────────────┐ │ Flask Web 服务器 │ │ ┌─────────┐ ┌──────────┐ ┌──────────┐ ┌────────┐ │ │ │ 认证 │ │ 权限校验 │ │ 多租户 │ │ 限流 │ │ │ └────┬────┘ └────┬─────┘ └────┬─────┘ └────┬───┘ │ │ │ │ │ │ │ │ ┌────▼────────────▼──────────────▼──────────────▼──┐ │ │ │ 业务 API 层 (Blueprint) │ │ │ │ /api/v1/auth | /api/v1/crm | /api/v1/inv... │ │ │ └──────────────────────┬───────────────────────────┘ │ │ │ │ │ ┌──────────────────────▼───────────────────────────┐ │ │ │ Service 业务服务层 │ │ │ │ CRUD、业务规则、数据校验、事务管理 │ │ │ └──────────────────────┬───────────────────────────┘ │ │ │ │ │ ┌──────────────────────▼───────────────────────────┐ │ │ │ SQLAlchemy ORM 层 │ │ │ │ 多租户表模型、关联查询、事务 │ │ │ └──────────────────────┬───────────────────────────┘ │ └─────────────────────────┼──────────────────────────────┘ │ ┌───────────┴────────────┐ ▼ ▼ MySQL 数据库 Redis 缓存 (多租户隔离: tenant_id) (验证码、限流、会话) ### 1.3 目录结构 pear-admin-flask/ ├── applications/ │ ├── __init__.py # Flask 应用工厂 │ ├── config.py # 配置管理pydantic-settings │ ├── api/v1/ # API 路由各业务模块 │ ├── models/ # 数据库模型ORM │ ├── services/ # 业务服务层 │ ├── common/ # 公共工具 │ │ ├── tenant_utils.py # 多租户核心逻辑 ⭐ │ │ ├── operation_log.py # 操作日志 │ │ └── utils/ # 验证码、文件上传等 │ └── extensions/ # 插件初始化 │ ├── init_sqlalchemy.py # 数据库连接 │ ├── init_jwt.py # JWT 认证 │ └── init_limiter.py # 速率限制 ├── frontend/ │ ├── src/ │ │ ├── api/ # API 调用封装 │ │ ├── views/ # 页面组件各业务模块 │ │ ├── router/ # Vue Router │ │ └── store/ # Pinia 状态管理 │ └── package.json └── docs/ # 使用说明书 --- ## 二、核心难点多租户架构设计 ### 2.1 租户模型设计 SaaS 平台的核心设计就是**多租户数据隔离**。我采用了最简单且实用的方案**共享数据库 Schema 内的 tenant_id 字段隔离**。 python # applications/models/tenant.py class TenantTenant(db.Model): 租户表 __tablename__ tenant_tenant id db.Column(db.Integer, primary_keyTrue, autoincrementTrue) code db.Column(db.String(32), uniqueTrue, nullableFalse, comment租户编码) name db.Column(db.String(100), nullableFalse, comment企业名称) plan_id db.Column(db.Integer, db.ForeignKey(tenant_plan.id), comment当前套餐) status db.Column(db.SmallInteger, default1, comment状态:0停用1正常2过期) # ... 其他字段 class TenantPlan(db.Model): 套餐表 - 定义功能权限和资源限制 __tablename__ tenant_plan # features 字段存储 JSON描述套餐支持的模块和数量限制 features db.Column(db.JSON, comment功能权限JSON) class TenantSubscription(db.Model): 订阅记录表 - 跟踪租户付费历史 tenant_id db.Column(db.Integer, db.ForeignKey(tenant_tenant.id), nullableFalse) plan_id db.Column(db.Integer, db.ForeignKey(tenant_plan.id), nullableFalse) # ... 起止日期、金额、状态等 ### 2.2 所有业务表都加入 tenant_id python # applications/models/crm.py - 客户表 class Customer(db.Model): __tablename__ crm_customer id db.Column(db.BigInteger(), primary_keyTrue, autoincrementTrue) tenant_id db.Column(db.BigInteger, default0, indexTrue, comment租户ID) # ⭐ name db.Column(db.String(100), nullableFalse, comment客户名称) # ... 其他字段 __table_args__ ( db.Index(ix_crm_customer_tenant_owner, tenant_id, owner_user_id), ) **关键设计要点** 1. **tenant_id 必须加索引**所有查询都会带上 WHERE tenant_id ? 2. **创建时自动注入 tenant_id**通过 TenantMixin 统一处理 3. **查询时自动过滤**通过 tenant_query() 工具方法避免遗漏 ### 2.3 多租户上下文管理 这是最核心的部分——**确保每个请求都在正确的租户上下文中执行**。 python # applications/common/tenant_utils.py def get_current_tenant_id(): 从 JWT token 或请求头获取当前租户 ID from flask import g # 优先从 g 中获取已通过中间件解析 if hasattr(g, tenant_id) and g.tenant_id: return g.tenant_id # 从 JWT payload 中解析 auth_header request.headers.get(Authorization, ) if auth_header.startswith(Bearer ): token auth_header[7:] payload decode_token(token) if payload: return payload.get(tenant_id) return None def tenant_query(model): 自动注入 tenant_id 过滤条件的查询辅助函数 from applications.extensions import db tenant_id get_current_tenant_id() query db.session.query(model).filter_by(is_deleted0) if tenant_id: query query.filter(model.tenant_id tenant_id) return query # 装饰器确保 API 只能访问本租户数据 def tenant_required(f): wraps(f) def decorated_function(*args, **kwargs): tenant_id get_current_tenant_id() if not tenant_id: return api_error(401, 未指定租户或登录已过期) return f(*args, **kwargs) return decorated_function 然后在每个业务 API 中这样使用 python # applications/api/v1/crm.py bp.route(/customer, methods[GET]) jwt_required() tenant_required def customer_list(): 客户列表 - 自动按当前租户过滤 # 直接用 tenant_query()不用手动加 tenant_id 条件 customers tenant_query(Customer).order_by(Customer.created_at.desc()).all() return api_success({list: [c.to_dict() for c in customers], total: len(customers)}) ### 2.4 请求中间件自动解析租户上下文 python # applications/__init__.py - create_app() 中注册 # 1. 请求 ID 中间件 app.before_request def set_request_id(): import uuid g.request_id uuid.uuid4().hex[:16] # 2. 多租户上下文初始化必须在路由之前 init_tenant_context(app) init_tenant_context 的核心逻辑是**在每次请求到达业务代码之前从 JWT Token 中解析出 tenant_id 并存入 flask.g**。 ### 2.5 套餐功能权限控制 SaaS 平台需要根据租户购买的套餐限制功能访问 python def plan_required(modules: str | list[str]): 套餐功能权限校验装饰器 if isinstance(modules, str): modules [modules] def decorator(f): wraps(f) def decorated_function(*args, **kwargs): tenant_id get_current_tenant_id() if not tenant_id: return api_error(401, 未登录) # 从套餐 features 中检查是否有权限 features get_plan_features(tenant_id) for mod in modules: if not features.get(mod, True): # 默认 True 表示不限制 return api_error(403, f当前套餐不支持【{mod}】模块) return f(*args, **kwargs) return decorated_function return decorator def plan_limit_check(model, message已达到套餐数量上限): 套餐资源数量限制如最多创建5个项目 def decorator(f): wraps(f) def decorated_function(*args, **kwargs): tenant_id get_current_tenant_id() features get_plan_features(tenant_id) limit_key f{model.__tablename__}_limit limit features.get(limit_key) if limit: current_count tenant_query(model).count() if current_count limit: return api_error(403, message) return f(*args, **kwargs) return decorated_function return decorator --- ## 三、认证与安全JWT Session Cookie 双轨制 ### 3.1 认证流程设计 前端登录 → 发送用户名密码验证码 ↓ 后端校验 → 生成 JWT Token内含 user_id, tenant_id, roles ↓ 返回响应 → { code: 0, data: { access_token: ..., user: {...} } } ↓ 前端存储 → Pinia store localStorage仅缓存不用于鉴权 ↓ 后续请求 → Authorization: Bearer token ### 3.2 验证码生成带容错降级 这是一个真实的生产环境问题**Redis 挂了的时候验证码也不能生成**。 解决思路**Redis 不可用时自动降级到内存存储**。 python # applications/common/captcha_store.py class CaptchaStore: 验证码存储优先 Redis失败降级到内存字典 def __init__(self): self._memory_store {} try: import redis self._redis redis.from_url(REDIS_URL, socket_connect_timeout2) self._redis.ping() # 测试连接 except Exception: self._redis None # Redis 不可用降级到内存 def save(self, key: str, code: str, ttl: int 300): if self._redis: try: self._redis.setex(fcaptcha:{key}, ttl, code) return except Exception: pass # Redis 突然挂了降级到内存 # 内存存储仅开发环境 self._memory_store[key] code def verify(self, key: str, code: str) - bool: # 类似的降级逻辑 pass ### 3.3 速率限制防止暴力破解 同样使用 Redis 存储限流计数器但也有降级机制 python # applications/extensions/init_limiter.py limiter Limiter( key_funcget_remote_address, # 按 IP 限流 default_limits[200 per minute], storage_urimemory://, # 默认内存存储 ) def init_limiter(app: Flask): 根据 Redis 可用性选择存储后端 uri _resolve_storage_uri(app) # 优先 Redis不可用降级到 memory try: limiter._storage_uri uri limiter.init_app(app) logger.info(fFlask-Limiter 已初始化存储: {uri}) except Exception as e: # Redis 初始化失败回退到内存存储 limiter._storage_uri memory:// limiter.init_app(app) ### 3.4 安全加固 python # applications/__init__.py - CSP 内容安全策略 app.after_request def add_security_headers(response): if response.content_type and text/html in response.content_type: response.headers[Content-Security-Policy] ( default-src self; script-src self unsafe-inline; style-src self unsafe-inline; img-src self data: blob:; font-src self data:; connect-src self http://127.0.0.1:5000; object-src none; base-uri self ) response.headers[X-Content-Type-Options] nosniff # 防止 MIME 嗅探 response.headers[X-Frame-Options] SAMEORIGIN # 防止点击劫持 return response # applications/config.py - Session Cookie 安全配置 SESSION_COOKIE_HTTPONLY: True, # JS 不可读取防 XSS SESSION_COOKIE_SAMESITE: Lax, # 防 CSRF SESSION_COOKIE_SECURE: not DEBUG, # 生产环境强制 HTTPS --- ## 四、配置管理从 .env 到 Python 对象 对于 SaaS 平台**配置不能硬编码**。使用 pydantic-settings 优雅处理 python # applications/config.py from pydantic import Field from pydantic_settings import BaseSettings, SettingsConfigDict class Settings(BaseSettings): 从环境变量和 .env 文件读取应用配置 model_config SettingsConfigDict( env_file.env, env_file_encodingutf-8, case_sensitiveFalse, extraignore, ) # 基础配置 FLASK_ENV: str Field(defaultdevelopment) SECRET_KEY: str Field(defaultchange-me-in-env) # 数据库 DATABASE_URL: Optional[str] Field(defaultNone) MYSQL_USERNAME: str Field(defaultroot) # ... # Redis REDIS_URL: str Field(defaultredis://localhost:6379/0) # CORS CORS_ORIGINS: str Field(defaulthttp://localhost:3000) property def sqlalchemy_database_uri(self) - str: 自动拼接 MySQL 连接串 if self.DATABASE_URL: return self.DATABASE_URL return fmysqlpymysql://{self.MYSQL_USERNAME}:{self.MYSQL_PASSWORD}{self.MYSQL_HOST}:{self.MYSQL_PORT}/{self.MYSQL_DATABASE} property def ratelimit_storage_uri(self) - str: 限流存储默认复用 Redis return self.RATELIMIT_STORAGE_URI or self.REDIS_URL def to_flask_config(self) - Dict[str, Any]: 转换为 Flask app.config 格式 return { SQLALCHEMY_DATABASE_URI: self.sqlalchemy_database_uri, # ... } lru_cache(maxsize1) # 缓存配置对象避免每次读取 .env def get_settings() - Settings: env_file Path(__file__).resolve().parents[1] / .env return Settings(_env_fileenv_file) .env 文件示例 ini # 数据库 MYSQL_USERNAMEroot MYSQL_PASSWORDyour_password MYSQL_HOST127.0.0.1 MYSQL_DATABASE # Redis REDIS_URLredis://localhost:6379/0 # 安全 SECRET_KEY生产环境请替换为强随机密钥 # 跨域 CORS_ORIGINShttp://localhost:3000,https://your-domain.com --- ## 五、前端架构Vue 3 TypeScript ### 5.1 API 请求封装 typescript // frontend/src/utils/request.ts import axios, { AxiosInstance } from axios import { useUserStore } from /store/user import router from /router const service: AxiosInstance axios.create({ baseURL: /api/v1, // 通过 Vite 代理到后端 timeout: 15000, withCredentials: true, // 允许跨域携带 cookie }) // 请求拦截器自动注入 Token service.interceptors.request.use(config { const userStore useUserStore() if (userStore.accessToken) { config.headers[Authorization] Bearer ${userStore.accessToken} } return config }) // 响应拦截器统一处理错误码 service.interceptors.response.use( response { const data response.data if (data.code 0) return data // 401: 未认证 → 跳登录 if (data.code 401) { useUserStore().logout() router.push(/auth/login) } return Promise.reject(data.message || 请求失败) }, error { return Promise.reject(error.message) } ) export default service ### 5.2 Vite 代理配置开发环境 typescript // frontend/vite.config.ts server: { proxy: { /api: { target: http://127.0.0.1:5000, changeOrigin: true, }, }, } ### 5.3 前端 SPA 与后端 Flask 的集成 生产环境打包后前端 dist/ 作为 Flask 的静态资源 python # applications/__init__.py _frontend_dist os.path.join(os.path.dirname(__file__), .., frontend, dist) # 静态资源路由 app.route(/assets/path:filename) def serve_assets(filename): return send_from_directory(os.path.join(_frontend_dist, assets), filename) # Catch-all支持 Vue Router history 模式 app.route(/path:path, methods[GET]) def spa_catch_all(path): if path.startswith(api/): abort(404) # API 404 不应返回 HTML return send_from_directory(_frontend_dist, index.html) --- ## 六、核心业务模块实现 ### 6.1 CRM 客户管理模块 数据库模型设计 python class Customer(db.Model): __tablename__ crm_customer id db.Column(db.BigInteger(), primary_keyTrue) tenant_id db.Column(db.BigInteger, default0, indexTrue) # 多租户 name db.Column(db.String(100), nullableFalse) short_name db.Column(db.String(50)) level db.Column(db.String(1), defaultC) # A/B/C/D status db.Column(db.SmallInteger, default0) # 0潜在 1跟进 2成交 3流失 in_pool db.Column(db.SmallInteger, default1) # 是否在公海池 owner_user_id db.Column(db.BigInteger) # 负责人 # ... class Contact(db.Model): # 联系人1对多客户→联系人 class Followup(db.Model): # 跟进记录 class Opportunity(db.Model): # 商机 class SalesOrder(db.Model): # 销售订单 业务 API 示例 python bp.route(/customer/int:customer_id/release, methods[POST]) jwt_required() tenant_required def customer_release(customer_id): 释放客户到公海池 tenant_id get_current_tenant_id() customer Customer.query.filter_by( idcustomer_id, tenant_idtenant_id, is_deleted0 ).first() if not customer: return api_error(404, 客户不存在) customer.in_pool 1 customer.owner_user_id None db.session.commit() return api_success({message: 已释放到公海池}) ### 6.2 AI 智能助手集成 这是产品差异化的亮点。架构设计 前端 AI 对话页面 → /api/v1/ai/chat ↓ 后端接收消息 → 调用 AI 服务商 APIOpenAI/通义/其他 ↓ 流式返回结果 → SSE (Server-Sent Events) 给前端 ↓ 记录使用量 → AIModelUsage 表用于计费/配额统计 --- ## 七、部署与运维 ### 7.1 生产环境部署 bash # 1. 克隆项目 git clone https://github.com/your-repo/jianyu.git cd jianyu # 2. 安装依赖 pip install -r requirements.txt cd frontend npm install npm run build cd .. # 3. 配置环境变量 cp .env.example .env # 编辑 .env数据库、Redis、SECRET_KEY # 4. 启动服务使用 gunicorn nginx gunicorn -w 4 -b 0.0.0.0:5000 applications:create_app() ### 7.2 Nginx 配置参考 nginx server { listen 443 ssl; server_name your-domain.com; # SSL 证书Lets Encrypt ssl_certificate /etc/letsencrypt/live/your-domain.com/fullchain.pem; ssl_certificate_key /etc/letsencrypt/live/your-domain.com/privkey.pem; # API 请求转发到 Flask location /api/ { proxy_pass http://127.0.0.1:5000; proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; } # 静态资源前端打包文件 location /assets/ { alias /path/to/frontend/dist/assets/; expires 30d; # 强缓存 } # 其他路径返回 SPA 入口 location / { try_files $uri /index.html; alias /path/to/frontend/dist/; } } --- ## 八、踩坑经验与最佳实践 ### 8.1 常见坑 | 问题 | 解决方案 | |---|---| | **验证码字体缺失** | 优先使用系统字体失败时用 PIL 默认字体兜底 | | **Redis 未启动导致全挂** | Redis 不可用时降级到内存存储仅开发环境 | | **Flask 2.3 JSON 中文转义** | app.json.ensure_ascii False | | **多租户查询遗漏 tenant_id** | 用 tenant_query() 统一封装禁止裸写 db.session.query() | | **Vue Router history 模式 404** | Flask Catch-all 路由返回 index.html | | **SQLAlchemy 事务回滚** | 批量导入时 try-except rollback确保单条失败不影响整体 | ### 8.2 开发建议 1. **配置文件绝不硬编码**全部走 .env pydantic-settings 2. **模块间解耦**每个业务模块有独立的 models、services、api 文件 3. **操作日志**所有写操作都要记录操作人、时间、内容 4. **逻辑删除**业务数据用 is_deleted0/1不真删 5. **错误兜底**每个外部依赖Redis、字体文件、AI API都要有降级方案 6. **数据库连接池**合理配置 pool_size、pool_recycleMySQL 8小时超时问题 --- ## 九、项目开源与后续计划 当前项目已实现的功能模块 - ✅ 多租户架构租户-套餐-订阅 - ✅ 用户权限管理RBAC用户-角色-权限 - ✅ AI 智能助手对话、报表、项目分析 - ✅ CRM 客户关系管理客户、联系人、商机、订单、回款 - ✅ 进销存商品、库存、采购、仓库、盘点、调拨 - ✅ BOM 物料管理 生产工单 领料管理 - ✅ 模具管理档案、出入库、生产计划 - ✅ 财务管理账户、收支、付款、报销、资金调拨 - ✅ 考勤 / 薪资 / 员工管理 - ✅ 项目管理 - ✅ 操作日志 / 登录日志审计 - ✅ 标签打印 后续计划 - 对接真实支付网关支付宝、微信支付 - SaaS 订阅管理自动化自动续费、过期通知 - 移动端小程序 - 插件市场允许第三方开发者扩展功能 --- ## 十、写在最后 从一个简单的后台管理模板起步到完整的多租户 SaaS 平台这一路走来最大的感悟是 **架构设计比写代码更重要。** 多租户隔离、配置管理、错误降级这些看似不紧急的事情如果一开始设计不当后续的维护成本会指数级增长。 **工程化思维比技术炫技更重要。** 一个能真正跑起来的 SaaS 平台需要考虑 Redis 挂了怎么办、验证码字体找不到怎么办、客户批量导入失败了怎么办——这些都是教科书不会告诉你的真实场景。 如果你也想做一个类似的项目欢迎交流讨论。也欢迎关注我的 CSDN 博客后续会分享更多实战经验。 --- **在线演示**[https://www.ztian.cc.cd] **默认测试账号**admin / Admin123 *如果本文对你有帮助欢迎点赞 、收藏 ⭐、关注 *