多Flask应用共享AWS RDS时，如何建立User与Table的一对多关联

第一步：修复基础代码错误 #

首先修正second_application.py中的低级错误：将class Table(db.model)改为class Table(db.Model)（注意Model首字母大写），否则模型无法正确初始化。

第二步：实现跨应用的一对多关联 #

由于两个应用共享同一AWS RDS数据库，只需保证模型的关联定义一致即可，以下提供两种实现方式：

方式1：在各自应用中同步模型定义 #

修改first_application.py的User模型

在User类中添加关联字段，建立与Table的一对多关系：

class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(80), nullable=False)
    password = db.Column(db.String(6), unique=True, nullable=False)
    role = db.Column(
        db.Enum(RoleEnum),
        default=RoleEnum.waiter,
        nullable=False
    )
    shift = db.Column(
        db.Enum(ShiftEnum),
        default=ShiftEnum.night,
        nullable=False
    )
    started_job = db.Column(db.DateTime, default=datetime.utcnow)
    # 添加一对多关联：一个用户可负责多个餐桌
    tables = db.relationship('Table', backref='assigned_waiter', lazy=True)

    def __repr__(self):
        return '<User %r>' % self.name

修改second_application.py的Table模型

修正类名并确保外键逻辑正确：

class Table(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    number = db.Column(db.String(80), nullable=False)
    chairs = db.Column(db.Integer)
    status = db.Column(
        db.Enum(TableStatusEnum),
        default=TableStatusEnum.free_table,
        nullable=False
    )
    # 外键关联user表的id字段，允许为空表示餐桌未分配服务员
    waiter_id = db.Column(db.Integer, db.ForeignKey('user.id'), nullable=True)

这里backref='assigned_waiter'会自动在Table模型中生成assigned_waiter属性，用于反向查询对应的User实例；User模型中的tables属性则可直接获取该用户负责的所有餐桌。

方式2：抽离共享模型到公共模块（推荐） #

为避免重复代码和模型定义不一致，建议将共享的模型、枚举类抽成独立公共模块，两个应用统一从该模块导入。

1. 创建公共模型模块

新建shared/models.py文件，统一存放所有共享模型：

import enum
from datetime import datetime
from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()

class RoleEnum(enum.Enum):
    waiter = 'waiter'
    manager = 'manager'

class ShiftEnum(enum.Enum):
    night = 'night'
    day = 'day'

class TableStatusEnum(enum.Enum):
    reserved = 'Reserved'
    free_table = 'Free table'
    preperation = 'Preperation'
    occupied = 'Occupied'

class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(80), nullable=False)
    password = db.Column(db.String(6), unique=True, nullable=False)
    role = db.Column(
        db.Enum(RoleEnum),
        default=RoleEnum.waiter,
        nullable=False
    )
    shift = db.Column(
        db.Enum(ShiftEnum),
        default=ShiftEnum.night,
        nullable=False
    )
    started_job = db.Column(db.DateTime, default=datetime.utcnow)
    tables = db.relationship('Table', backref='assigned_waiter', lazy=True)

    def __repr__(self):
        return '<User %r>' % self.name

class Table(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    number = db.Column(db.String(80), nullable=False)
    chairs = db.Column(db.Integer)
    status = db.Column(
        db.Enum(TableStatusEnum),
        default=TableStatusEnum.free_table,
        nullable=False
    )
    waiter_id = db.Column(db.Integer, db.ForeignKey('user.id'), nullable=True)

2. 改造第一个应用（first_application.py）

导入公共模型并初始化：

import os
from datetime import datetime, timezone

from flask import Flask, jsonify, request
from flask_jwt_extended import create_access_token, get_jwt_identity, jwt_required, JWTManager
from flask_cors import CORS, cross_origin
from dotenv import load_dotenv
from shared.models import db, User, RoleEnum, ShiftEnum

load_dotenv()
application = Flask(__name__)
CORS(application, support_credentials=True)

# 从环境变量读取配置，避免硬编码
application.config['SECRET_KEY'] = os.getenv('SECRET_KEY', 'same_secret')
application.config['SQLALCHEMY_DATABASE_URI'] = os.getenv('DATABASE_URI', 'same_rds_db_uri')

# 初始化db和JWT
db.init_app(application)
jwt = JWTManager(application)

@application.route("/login", methods=["POST"])
def login():
    password = request.json.get("password", None)
    user = User.query.filter_by(password=password).first_or_404()
    access_token = create_access_token(identity=user.name)
    return jsonify(access_token=access_token)

@application.route("/start-job", methods=["POST"])
@jwt_required()
def start_job():
    current_user = get_jwt_identity()
    user = User.query.filter_by(name=current_user).first()
    if not user:
        return jsonify({"message": "用户不存在"}), 404
    user.started_job = datetime.now(timezone.utc)
    db.session.commit()  # 必须提交会话才能保存修改
    return jsonify({"message": "已开始工作"}), 201

with application.app_context():
    db.create_all()

if __name__ == "__main__":
    application.run(debug=True, port=5000)

3. 改造第二个应用（second_application.py）

同样导入公共模型：

import os
from flask import Flask, jsonify, request
from flask_cors import CORS
from dotenv import load_dotenv
from shared.models import db, Table, TableStatusEnum

load_dotenv()
application = Flask(__name__)
CORS(application, support_credentials=True)

application.config['SECRET_KEY'] = os.getenv('SECRET_KEY', 'same_secret')
application.config['SQLALCHEMY_DATABASE_URI'] = os.getenv('DATABASE_URI', 'same_rds_db_uri')

db.init_app(application)

# 示例路由：获取指定服务员负责的所有餐桌
@application.route("/waiter/<int:waiter_id>/tables", methods=["GET"])
def get_waiter_tables(waiter_id):
    tables = Table.query.filter_by(waiter_id=waiter_id).all()
    table_list = [
        {
            "id": t.id,
            "number": t.number,
            "chairs": t.chairs,
            "status": t.status.value
        } for t in tables
    ]
    return jsonify(table_list)

if __name__ == "__main__":
    application.run(debug=True, port=5001)  # 使用不同端口避免冲突

关键注意事项 #

• 配置统一管理：不要硬编码SECRET_KEY和数据库URI，用.env文件存储，两个应用读取相同环境变量，确保配置一致。
• 数据库会话提交：所有修改数据库的操作（如start_job中的更新）必须调用db.session.commit()才能生效。
• 数据库迁移：使用Flask-Migrate管理数据库结构变更，只在一个应用中执行迁移命令，避免多应用同时修改结构导致冲突。
• 端口冲突：本地开发时两个应用需使用不同端口启动，避免端口占用。

内容的提问来源于stack exchange，提问作者Mike D Hovhannisyan