通过REST API强制执行外键约束的最佳实践是什么？

Question

我正在将待办事项列表WebApp作为一个爱好项目，使用烧瓶作为后端和一个PostgreSQL数据库来存储数据。数据库模型如下：

models.py.py

class Group(db.Model):
    __tablename__ = "groups"
    group_id = db.Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    name = db.Column(db.String(20), unique=True)


class Collection(db.Model):
    __tablename__ = "collections"
    collection_id = db.Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    name = db.Column(db.String(20), unique=True)
    group_name = db.Column(db.String(20), db.ForeignKey("groups.name"), nullable=True)

    def to_dict(self):
        return {
            "collection_id": self.collection_id,
            "name": self.name,
            "group_name": self.group_name,
        }


class Task(db.Model):
    __tablename__ = "tasks"
    task_id = db.Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    text = db.Column(db.String(200))
    completed = db.Column(db.Boolean, default=False)
    collection_name = db.Column(
        db.String(20), db.ForeignKey("collections.name"), nullable=False
    )

    def to_dict(self):
        return {
            "task_id": self.task_id,
            "text": self.text,
            "completed": self.completed,
            "collection_name": self.collection_name,
        }

在制作用于将任务添加到数据库中的RET RET API逻辑时，我不确定是否：

• 我应该检查Collection> Collection_Name column在尝试将数据插入数据库之前，已经存在于Collections表中。
• 我应该尝试插入行，并捕获sqlalchemy.exc.integrityerror如果发生的话。

我在第一个解决方案中看到的问题是，我需要查询collections对有效collection_name的表格当然，如果这是一个很好的练习表现。

我在第二个解决方案中看到的问题是sqlalchemy.exc.IntegrityError异常非常模糊，在更复杂的表格，带有几个外国钥匙，我需要解析例外的信息才能知道哪个信息外国钥匙被侵犯了。

目前，我实现了第二个解决方案，因为我有一个非常简单的表和一个外键约束。

在以下内容中，您可以看到与数据库交谈的API调用和service.py的代码。

controler.py

@tasks_api.route(COMMON_API_ENDPOINT + "/tasks", methods=["POST"])
def add_task():
    request_body = request.get_json()

    # Check for missing fields in the call
    mandatory_fields = set(["text", "collection_name"])
    try:
        missing_fields = mandatory_fields - set(request_body.keys())
        assert len(missing_fields) == 0
    except AssertionError:
        return (
            jsonify(
                {
                    "error": "The following mandatory fields are missing: "
                    + str(missing_fields)
                }
            ),
            400,
        )

    # Try to call the add task service function
    try:
        task = TaskService.add_task(
            text=request_body["text"], collection_name=request_body["collection_name"]
        )
    except CollectionNotFoundError as e:
        return jsonify({"error_message": str(e)}), 400
    else:
        return (
            jsonify(
                {
                    "result": "A new task was created successfully.",
                    "description": task.to_dict(),
                }
            ),
            201,
        )

service.py

def add_task(text: str, collection_name: str) -> Task:
    try:
        with get_session() as session:
            task = Task(text=text, collection_name=collection_name)
            session.add(task)
            return task
    except sqlalchemy.exc.IntegrityError:
        raise CollectionNotFoundError(
            "Foreign key violation: There is no collection with the name "
            + collection_name
        )

在撰写这篇文章时，我想知道这是否是两个解决方案不是最好的问题。我也接受其他建议。

谢谢 ！