MySQL接入Codex进行数据管理的全面指南

操作系统：Windows 10+/macOS 10.15+/Linux (Ubuntu 20.04+)
MySQL 版本：5.7+ 或 8.0+（推荐 8.0.29+）
Node.js：14.0+（用于 Codex CLI）
Python：3.8+（可选，用于 API 调用）
内存：8GB+（推荐 16GB）

# Ubuntu/Debian
sudo apt update
sudo apt install mysql-server
# CentOS/RHEL
sudo yum install mysql-server
# macOS (Homebrew)
brew install mysql
# Windows
# 下载安装包：https://dev.mysql.com/downloads/installer/

# 安装 Codex CLI
npm install -g @openai/codex
# 登录 OpenAI 账户
codex login
# 验证安装
codex --version

# 启动交互模式
codex
# 在交互模式下输入自然语言指令
# 示例：生成 MySQL 连接代码

CREATE DATABASE codex_demo;
USE codex_demo;

CREATE TABLE users (
    id INT AUTO_INCREMENT PRIMARY KEY,
    username VARCHAR(50) NOT NULL UNIQUE,
    email VARCHAR(100) NOT NULL,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

CREATE TABLE posts (
    id INT AUTO_INCREMENT PRIMARY KEY,
    user_id INT,
    title VARCHAR(200) NOT NULL,
    content TEXT,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    FOREIGN KEY (user_id) REFERENCES users(id)
);

[database]
host = "localhost"
port = 3306
database = "codex_demo"
username = "root"
password = "your_password"
charset = "utf8mb4"
[codex]
api_key = "your_openai_api_key"
model = "codex-002"
temperature = 0.3
max_tokens = 1000

生成 Python Flask 应用的用户表 CRUD 操作代码，
表名为 users，字段包括 id（INT 主键）、username（VARCHAR 50）、email（VARCHAR 100），
使用 PyMySQL 直接执行 SQL。

import pymysql
from pymysql.cursors import DictCursor
class UserDAO:
    def __init__(self, config):
        self.config = config
    def get_connection(self):
        return pymysql.connect(
            host=self.config['host'],
            user=self.config['user'],
            password=self.config['password'],
            database=self.config['database'],
            charset='utf8mb4',
            cursorclass=DictCursor
        )
    def create_user(self, username, email):
        sql = "INSERT INTO users (username, email) VALUES (%s, %s)"
        try:
            conn = self.get_connection()
            with conn.cursor() as cursor:
                cursor.execute(sql, (username, email))
                conn.commit()
                return cursor.lastrowid
        finally:
            conn.close()
    def get_user(self, user_id):
        sql = "SELECT * FROM users WHERE id = %s"
        try:
            conn = self.get_connection()
            with conn.cursor() as cursor:
                cursor.execute(sql, (user_id,))
                return cursor.fetchone()
        finally:
            conn.close()
    def update_user(self, user_id, username=None, email=None):
        updates = []
        params = []
        if username:
            updates.append("username = %s")
            params.append(username)
        if email:
            updates.append("email = %s")
            params.append(email)
        if not updates:
            return False
        params.append(user_id)
        sql = f"UPDATE users SET {', '.join(updates)} WHERE id = %s"
        try:
            conn = self.get_connection()
            with conn.cursor() as cursor:
                cursor.execute(sql, params)
                conn.commit()
                return cursor.rowcount > 0
        finally:
            conn.close()
    def delete_user(self, user_id):
        sql = "DELETE FROM users WHERE id = %s"
        try:
            conn = self.get_connection()
            with conn.cursor() as cursor:
                cursor.execute(sql, (user_id,))
                conn.commit()
                return cursor.rowcount > 0
        finally:
            conn.close()

生成 MySQL WITH AS 语法示例，包含主查询和两个 CTE 子查询，
使用员工表和部门表，查询各部门薪资高于平均的员工信息。

WITH 
    dept_avg_salary AS (
        SELECT 
            department_id,
            AVG(salary) as avg_salary
        FROM employees
        GROUP BY department_id
    ),
    high_earners AS (
        SELECT 
            e.id,
            e.name,
            e.salary,
            e.department_id,
            d.department_name,
            das.avg_salary
        FROM employees e
        INNER JOIN departments d ON e.department_id = d.id
        INNER JOIN dept_avg_salary das ON e.department_id = das.department_id
        WHERE e.salary > das.avg_salary
    )
SELECT 
    department_name,
    COUNT(*) as high_earner_count,
    AVG(salary) as avg_high_salary,
    MAX(salary) as max_salary
FROM high_earners
GROUP BY department_name
ORDER BY high_earner_count DESC;

生成 MySQL 存储过程，实现批量插入用户数据并返回成功数量，
包含错误处理和事务管理。

DELIMITER $$

CREATE PROCEDURE batch_insert_users(
    IN user_data JSON,
    OUT success_count INT,
    OUT error_message VARCHAR(500)
)
BEGIN
    DECLARE done INT DEFAULT FALSE;
    DECLARE current_username VARCHAR(50);
    DECLARE current_email VARCHAR(100);
    DECLARE current_index INT DEFAULT 0;
    DECLARE total_count INT;
    DECLARE exit handler for sqlexception
    BEGIN
        ROLLBACK;
        SET success_count = 0;
        SET error_message = CONCAT('Error at index ', current_index, ': ', 
                                  IFNULL(@error_msg, 'Unknown error'));
    END;
    
    SET success_count = 0;
    SET error_message = '';
    SET total_count = JSON_LENGTH(user_data);
    
    START TRANSACTION;
    
    WHILE current_index < total_count DO
        SET current_username = JSON_UNQUOTE(
            JSON_EXTRACT(user_data, CONCAT('$[', current_index, '].username'))
        );
        SET current_email = JSON_UNQUOTE(
            JSON_EXTRACT(user_data, CONCAT('$[', current_index, '].email'))
        );
        
        IF current_username IS NOT NULL AND current_email IS NOT NULL THEN
            INSERT INTO users (username, email) 
            VALUES (current_username, current_email);
            
            IF ROW_COUNT() > 0 THEN
                SET success_count = success_count + 1;
            END IF;
        END IF;
        
        SET current_index = current_index + 1;
    END WHILE;
    
    COMMIT;
END$$

DELIMITER ;

-- 创建专用用户
CREATE USER 'codex_app'@'localhost' IDENTIFIED BY 'StrongPassword123!';

-- 授予最小权限
GRANT SELECT, INSERT, UPDATE, DELETE ON codex_demo.* TO 'codex_app'@'localhost';

-- 限制连接数
ALTER USER 'codex_app'@'localhost' WITH MAX_USER_CONNECTIONS 10;

-- 刷新权限
FLUSH PRIVILEGES;

# 推荐：使用环境变量
import os
from dotenv import load_dotenv
load_dotenv()
CODex_API_KEY = os.getenv('CODex_API_KEY')
DATABASE_PASSWORD = os.getenv('DB_PASSWORD')
# .env 文件（不要提交到版本控制）
CODex_API_KEY=sk-xxxxxxxxxxxxxxxxxxxxxxxx
DB_PASSWORD=your_secure_password

# ✅ 正确做法：参数化查询
def safe_query(user_id):
    sql = "SELECT * FROM users WHERE id = %s"
    cursor.execute(sql, (user_id,))
# ❌ 错误做法：字符串拼接
def unsafe_query(user_id):
    sql = f"SELECT * FROM users WHERE id = {user_id}"  # SQL 注入风险
    cursor.execute(sql)

-- Codex 生成的索引建议
CREATE INDEX idx_users_email ON users(email);
CREATE INDEX idx_posts_user_id ON posts(user_id);
CREATE INDEX idx_posts_created_at ON posts(created_at DESC);

-- 复合索引
CREATE INDEX idx_users_username_email ON users(username, email);

-- 启用慢查询日志
SET GLOBAL slow_query_log = 'ON';
SET GLOBAL long_query_time = 2;
SET GLOBAL slow_query_log_file = '/var/log/mysql/mysql-slow.log';

-- 分析慢查询
SELECT 
    query,
    avg_timer_wait,
    max_timer_wait,
    count_star
FROM performance_schema.events_statements_summary_by_digest
ORDER BY avg_timer_wait DESC
LIMIT 10;

# HikariCP 连接池配置（Java）
HikariConfig config = new HikariConfig();
config.setJdbcUrl("jdbc:mysql://localhost:3306/codex_demo");
config.setUsername("codex_app");
config.setPassword("password");
config.setMaximumPoolSize(20);
config.setMinimumIdle(5);
config.setConnectionTimeout(30000);
config.setIdleTimeout(600000);
config.setMaxLifetime(1800000);

# Python 连接池
from DBUtils.PooledDB import PooledDB
import pymysql

pool = PooledDB(
    creator=pymysql,
    maxconnections=20,
    mincached=5,
    maxcached=10,
    blocking=True,
    host='localhost',
    user='codex_app',
    password='password',
    database='codex_demo',
    charset='utf8mb4'
)

-- 1. 创建数据库
CREATE DATABASE student_management;
USE student_management;

-- 2. 创建表结构
CREATE TABLE students (
    id INT AUTO_INCREMENT PRIMARY KEY,
    student_id VARCHAR(20) UNIQUE NOT NULL,
    name VARCHAR(50) NOT NULL,
    gender ENUM('M', 'F') NOT NULL,
    birth_date DATE,
    class VARCHAR(20),
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    INDEX idx_student_id (student_id),
    INDEX idx_class (class)
);

CREATE TABLE courses (
    id INT AUTO_INCREMENT PRIMARY KEY,
    course_code VARCHAR(20) UNIQUE NOT NULL,
    course_name VARCHAR(100) NOT NULL,
    credits INT DEFAULT 3,
    teacher VARCHAR(50),
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    INDEX idx_course_code (course_code)
);

CREATE TABLE scores (
    id INT AUTO_INCREMENT PRIMARY KEY,
    student_id VARCHAR(20) NOT NULL,
    course_code VARCHAR(20) NOT NULL,
    score DECIMAL(5,2),
    semester VARCHAR(20),
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    FOREIGN KEY (student_id) REFERENCES students(student_id) ON DELETE CASCADE,
    FOREIGN KEY (course_code) REFERENCES courses(course_code) ON DELETE CASCADE,
    UNIQUE KEY uk_student_course (student_id, course_code, semester),
    INDEX idx_score (score),
    INDEX idx_semester (semester)
);

-- 1. 添加缺失索引
CREATE INDEX idx_orders_user_id_created_at ON orders(user_id, created_at DESC);
CREATE INDEX idx_orders_status ON orders(status);
CREATE INDEX idx_order_items_order_id ON order_items(order_id);

-- 2. 分区表优化（按时间分区）
ALTER TABLE orders 
PARTITION BY RANGE (YEAR(created_at) * 100 + MONTH(created_at)) (
    PARTITION p202501 VALUES LESS THAN (202502),
    PARTITION p202502 VALUES LESS THAN (202503),
    PARTITION p202503 VALUES LESS THAN (202504),
    PARTITION pmax VALUES LESS THAN MAXVALUE
);

-- 3. 读写分离配置建议
-- 主库：写操作
-- 从库：读操作（SELECT）
-- 使用中间件如 MyCat 或 ShardingSphere

Ctrl+Shift+P  - 打开命令面板
Ctrl+Alt+C    - 激活 Codex 聊天窗口
Ctrl+Enter    - 发送消息到 Codex
Alt+Up/Down   - 切换历史消息

SHOW DATABASES;           -- 查看所有数据库
USE database_name;        -- 选择数据库
SHOW TABLES;              -- 查看表
DESC table_name;          -- 查看表结构
SHOW INDEX FROM table_name; -- 查看索引
EXPLAIN SELECT ...        -- 分析查询执行计划

# 检查网络连接
ping api.openai.com

# 增加超时时间
export CODex_TIMEOUT=120

# 重启 Codex 服务
codex restart

-- 修改认证插件
ALTER USER 'root'@'localhost' IDENTIFIED WITH mysql_native_password BY 'password';

-- 刷新权限
FLUSH PRIVILEGES;

# Python 连接时指定认证插件
connection = pymysql.connect(
    host='localhost',
    user='root',
    password='password',
    database='test',
    charset='utf8mb4',
    cursorclass=pymysql.cursors.DictCursor,
    ssl_disabled=True  # 禁用 SSL（仅开发环境）
)