Repository: ErikBjare/gptme
File: gptme/server/api_v2.py (Python)

Code before:

    # Notify all sessions that a new message was added
    SessionManager.add_event(
        conversation_id,
        {
            "type": "message_added",
            "message": {
                "role": msg.role,
                "content": msg.content,
                "timestamp": msg.timestamp.isoformat(),
            },
        },
    )

    return flask.jsonify({"status": "ok"})


@v2_api.route("/api/v2/conversations/<string:conversation_id>", methods=["DELETE"])
def api_conversation_delete(conversation_id: str):
    """Delete a conversation."""

    # Validate conversation_id to prevent path traversal
    if "/" in conversation_id or ".." in conversation_id or "\\" in conversation_id:
        return flask.jsonify({"error": "Invalid conversation_id"}), 400

    logdir = get_logs_dir() / conversation_id
    if not logdir.exists():
        return flask.jsonify(
            {"error": f"Conversation not found: {conversation_id}"}
        ), 404

    try:
        shutil.rmtree(logdir)
    except OSError as e:
        logger.error(f"Error deleting conversation {conversation_id}: {e}")
        return flask.jsonify({"error": f"Could not delete conversation: {e}"}), 500

    SessionManager.remove_all_sessions_for_conversation(conversation_id)

    return flask.jsonify({"status": "ok"})


@v2_api.route("/api/v2/conversations/<string:conversation_id>/events")
def api_conversation_events(conversation_id: str):
    """Subscribe to conversation events."""
    session_id = request.args.get("session_id")
    if not session_id:
        # load chat config
        logdir = get_logs_dir() / conversation_id
        chat_config = ChatConfig.load_or_create(logdir, ChatConfig())


Review comment (suggestion):
Defensive stuff:

```suggestion
    logdir = get_logs_dir() / conversation_id
    assert logdir.parent == get_logs_dir()
```

Code after:

    # Notify all sessions that a new message was added
    SessionManager.add_event(
        conversation_id,
        {
            "type": "message_added",
            "message": {
                "role": msg.role,
                "content": msg.content,
                "timestamp": msg.timestamp.isoformat(),
            },
        },
    )

    return flask.jsonify({"status": "ok"})


@v2_api.route("/api/v2/conversations/<string:conversation_id>", methods=["DELETE"])
def api_conversation_delete(conversation_id: str):
    """Delete a conversation."""

    # Validate conversation_id to prevent path traversal
    if "/" in conversation_id or ".." in conversation_id or "\\" in conversation_id:
        return flask.jsonify({"error": "Invalid conversation_id"}), 400

    logdir = get_logs_dir() / conversation_id
    assert logdir.parent == get_logs_dir()
    if not logdir.exists():
        return flask.jsonify(
            {"error": f"Conversation not found: {conversation_id}"}
        ), 404

    try:
        shutil.rmtree(logdir)
    except OSError as e:
        logger.error(f"Error deleting conversation {conversation_id}: {e}")
        return flask.jsonify({"error": f"Could not delete conversation: {e}"}), 500

    SessionManager.remove_all_sessions_for_conversation(conversation_id)

    return flask.jsonify({"status": "ok"})


@v2_api.route("/api/v2/conversations/<string:conversation_id>/events")
def api_conversation_events(conversation_id: str):
    """Subscribe to conversation events."""
    session_id = request.args.get("session_id")
    if not session_id:
        # load chat config
        logdir = get_logs_dir() / conversation_id
        chat_config = ChatConfig.load_or_create(logdir, ChatConfig())