如何通过Python Eve程序化创建DOMAIN端点及Mongo集合并持久化

我来一步步帮你解决这个问题，先搞定当前的动态创建端点，再扩展到Schema持久化和启动自动加载的需求。

一、修复动态创建集合的端点 #

你的当前代码有几个关键问题：

1. 路由默认是GET请求，而提交Schema应该用POST方法
2. 路由函数不能直接接收schema参数，需要从请求体里获取JSON数据
3. Eve的DOMAIN配置需要符合规范（不是直接放schema.def，而是要包含schema字段和必要的资源配置）
4. 修改DOMAIN后需要重新加载EVE的配置才能让新集合生效

修正后的代码示例 #

from flask import request
import json
from eve import Eve

app = Eve()

@app.route('/gen', methods=['POST'])
def generate_collection():
    try:
        # 获取请求中的JSON数据
        payload = request.get_json()
        if not payload or not isinstance(payload, list) or len(payload) == 0:
            return {"error": "Invalid payload: expected a non-empty list of schema definitions"}, 400
        
        schema_data = payload[0]
        # 验证必要字段
        if 'name' not in schema_data or 'def' not in schema_data:
            return {"error": "Missing required fields: 'name' and 'def' are mandatory"}, 400
        
        coll_name = schema_data['name']
        # 解析Schema定义：如果提交的是字符串则转成字典，否则直接使用
        schema_def = schema_data['def']
        if isinstance(schema_def, str):
            try:
                schema_def = json.loads(schema_def)
            except json.JSONDecodeError:
                return {"error": "Invalid JSON string in 'def' field"}, 400
        
        # 检查集合是否已存在
        if coll_name in app.config['DOMAIN']:
            return {"error": f"Collection '{coll_name}' already exists"}, 409
        
        # 配置EVE资源（符合EVE的DOMAIN规范）
        app.config['DOMAIN'][coll_name] = {
            'item_title': coll_name,
            'schema': schema_def,
            # 开放必要的HTTP方法
            'resource_methods': ['GET', 'POST'],
            'item_methods': ['GET', 'PUT', 'DELETE']
        }
        
        # 重新加载EVE配置，让新集合生效
        app.reload()
        
        return {"message": f"Collection '{coll_name}' created successfully"}, 201
    
    except Exception as e:
        return {"error": str(e)}, 500

if __name__ == '__main__':
    app.run(debug=True)

修正后的Curl命令 #

注意：之前的Curl有两个问题：def字段不需要转成字符串，提交数据时不需要用数组（EVE默认POST单条数据）

1. 创建集合的命令：

curl -H "Content-Type: application/json" -d '[{"name":"test", "def": {"age": "integer"}}]' http://localhost:5000/gen

这里用"integer"而不是"int"，因为EVE的Schema类型遵循MongoEngine的规范。

2. 向新集合提交数据的命令：

curl -H "Content-Type: application/json" -d '{"age":5}' http://localhost:5000/test

二、持久化Schema到Mongo并启动自动加载 #

接下来实现把Schema保存到Mongo，让应用重启后自动加载这些动态创建的集合。

1. 保存Schema到Mongo #

在上面的generate_collection函数末尾，添加代码把Schema数据存入Mongo的schemas集合：

# 获取EVE的Mongo驱动连接
from eve.io.mongo import Mongo
mongo_driver = Mongo(app)
db = mongo_driver.db

# 保存Schema到schemas集合（自动去重，避免重复保存）
db.schemas.update_one(
    {"name": coll_name},
    {"$set": schema_data},
    upsert=True
)

2. 启动时自动加载Schema #

在EVE应用初始化后，读取schemas集合里的所有数据，自动添加到DOMAIN并加载：

app = Eve()

# 启动时加载保存的Schemas
def load_saved_schemas():
    try:
        mongo_driver = Mongo(app)
        db = mongo_driver.db
        
        # 查询所有保存的Schema
        saved_schemas = db.schemas.find()
        for schema in saved_schemas:
            coll_name = schema['name']
            schema_def = schema['def']
            # 跳过已存在的集合
            if coll_name in app.config['DOMAIN']:
                continue
            # 添加到DOMAIN
            app.config['DOMAIN'][coll_name] = {
                'item_title': coll_name,
                'schema': schema_def,
                'resource_methods': ['GET', 'POST'],
                'item_methods': ['GET', 'PUT', 'DELETE']
            }
        
        # 重新加载配置
        app.reload()
        print(f"Loaded {db.schemas.count_documents({})} saved schemas")
    except Exception as e:
        print(f"Failed to load saved schemas: {str(e)}")

# 在应用启动前加载Schema
with app.app_context():
    load_saved_schemas()

# 然后是路由定义...

额外注意事项 #

• 权限控制：这个/gen端点非常敏感，建议添加认证（比如EVE的AUTH_USERNAME_FIELD和对应的权限校验），避免未授权用户创建集合。
• Schema验证：可以在保存前验证提交的Schema是否符合EVE的规范，比如检查字段类型是否合法，避免无效Schema导致EVE报错。
• 批量创建：如果需要支持批量创建多个集合，可以修改代码遍历payload列表里的所有Schema数据。

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