如何用Python Flask基于用户输入创建对应数据库表的访问路由

方案1：直接实现动态路由匹配动态生成的表 #

如果你需要保留动态建表的逻辑，可以通过Flask的动态路径参数结合严格的输入校验实现需求，代码示例如下：

import re
# 提前维护用户创建的表白名单，生产环境建议存在数据库中并关联用户权限
USER_CREATED_TABLES = set()

# 动态访问路由
@app.route('/<string:table_name>', methods=['GET'])
def get_table_data(table_name):
    # 严格校验表名格式，仅允许字母、数字、下划线，长度不超过50
    if not re.fullmatch(r'^[a-zA-Z0-9_]{1,50}$', table_name):
        return jsonify({"error": "非法表名"}), 400
    # 校验表为合法的用户生成表，避免用户访问系统内置表
    if table_name not in USER_CREATED_TABLES:
        return jsonify({"error": "表不存在"}), 404
    
    # 动态构造当前表的SQLAlchemy模型
    class DynamicTable(db.Model):
        __tablename__ = table_name
        __table_args__ = {'extend_existing': True} # 避免重复定义模型报错
        ID = db.Column(db.Integer, primary_key=True)
        city = db.Column(db.String(100))
        country = db.Column(db.String(100))
    
    # 动态构造序列化器
    class DynamicSchema(ma.Schema):
        class Meta:
            fields = ('ID', 'city', 'country')
    
    # 执行查询返回结果
    data = DynamicTable.query.all()
    return DynamicSchema(many=True).jsonify(data)

# 原有的建表接口需要同步更新白名单
@app.route('/create_table', methods=['POST'])
def create_table():
    table_name = request.json['table_name']
    # 你的原有建表逻辑
    # ...
    # 建表完成后加入白名单
    USER_CREATED_TABLES.add(table_name)
    return jsonify({"msg": "创建成功"})

本方案风险提示 #

• 表名校验不严会导致SQL注入、越权访问系统表的安全漏洞
• 每次请求动态构造模型会带来额外性能开销，生产环境建议对模型做缓存
• 动态表数量过多会大幅提升数据库备份、迁移的维护成本

方案2：更安全的替代实现（推荐） #

完全放弃动态建表的设计，通过单表加标识字段的方式实现相同业务需求，从架构层面规避安全风险，代码示例如下：

# 修改原有模型，新增table_ident字段区分不同用户的数据集
class Destinations(db.Model):
    ID = db.Column(db.Integer, primary_key=True)
    table_ident = db.Column(db.String(50), nullable=False, index=True)
    city = db.Column(db.String(100))
    country = db.Column(db.String(100))

    def __init__(self, table_ident, city, country):
        self.table_ident = table_ident
        self.city = city
        self.country = country

# 序列化器同步新增字段
class DestinationsSchema(ma.Schema):
    class Meta:
        fields = ('ID', 'table_ident', 'city', 'country')

destination_schema = DestinationsSchema()
destinations_schema = DestinationsSchema(many=True)

# 带标识的新增数据路由
@app.route('/<string:table_ident>/destinations', methods=['POST'])
def add_dynamic_destination(table_ident):
    # 可自行添加权限校验，判断当前用户是否有权限操作该标识的数据集
    city = request.json['city']
    country = request.json['country']
    new_dest = Destinations(table_ident=table_ident, city=city, country=country)
    db.session.add(new_dest)
    db.session.commit()
    return destination_schema.jsonify(new_dest)

# 带标识的查询路由
@app.route('/<string:table_ident>/destinations', methods=['GET'])
def get_dynamic_destinations(table_ident):
    all_dest = Destinations.query.filter_by(table_ident=table_ident).all()
    result = destinations_schema.dump(all_dest)
    return jsonify(result)

本方案优势 #

• 无动态表、动态模型相关的安全风险，所有SQL均为预编译逻辑
• 数据库结构稳定，维护成本极低
• 可轻松扩展权限控制，关联用户ID后实现仅允许用户访问自己创建的数据集
• 性能远高于动态建表方案

内容的提问来源于stack exchange，提问作者Amit Goft