3.1 SQL语言概述
SQL（Structured Query Language）是用于管理关系数据库的标准语言。MySQL支持标准SQL语法，并提供了一些扩展功能。
3.1.1 SQL语言分类
class SQLLanguageCategories:
def __init__(self):
self.categories = {
"DDL (Data Definition Language)": {
"用途": "数据定义语言，用于定义数据库结构",
"主要命令": ["CREATE", "ALTER", "DROP", "TRUNCATE"],
"操作对象": ["数据库", "表", "索引", "视图", "存储过程"]
},
"DML (Data Manipulation Language)": {
"用途": "数据操作语言，用于操作数据",
"主要命令": ["INSERT", "UPDATE", "DELETE", "SELECT"],
"操作对象": ["表中的数据行"]
},
"DCL (Data Control Language)": {
"用途": "数据控制语言，用于控制访问权限",
"主要命令": ["GRANT", "REVOKE", "DENY"],
"操作对象": ["用户权限", "角色权限"]
},
"TCL (Transaction Control Language)": {
"用途": "事务控制语言，用于管理事务",
"主要命令": ["COMMIT", "ROLLBACK", "SAVEPOINT"],
"操作对象": ["事务"]
}
}
def show_categories(self):
"""显示SQL语言分类"""
print("SQL语言分类:")
print("=" * 60)
for category, info in self.categories.items():
print(f"\n{category}:")
print(f" 用途: {info['用途']}")
print(f" 主要命令: {', '.join(info['主要命令'])}")
print(f" 操作对象: {', '.join(info['操作对象'])}")
def sql_execution_order(self):
"""SQL执行顺序"""
execution_order = [
"FROM", # 确定数据源
"WHERE", # 行过滤
"GROUP BY", # 分组
"HAVING", # 分组过滤
"SELECT", # 选择列
"DISTINCT", # 去重
"ORDER BY", # 排序
"LIMIT" # 限制结果数量
]
print("\nSQL执行顺序:")
print("=" * 30)
for i, clause in enumerate(execution_order, 1):
print(f"{i}. {clause}")
# 示例查询
example_query = """
SELECT DISTINCT department, AVG(salary) as avg_salary
FROM employees
WHERE hire_date > '2020-01-01'
GROUP BY department
HAVING AVG(salary) > 50000
ORDER BY avg_salary DESC
LIMIT 5;
"""
print("\n示例查询:")
print(example_query)
# SQL语言分类演示
sql_categories = SQLLanguageCategories()
sql_categories.show_categories()
sql_categories.sql_execution_order()
3.1.2 MySQL数据类型详解
class MySQLDataTypes:
def __init__(self):
self.data_types = {
"数值类型": {
"整数类型": {
"TINYINT": {"范围": "-128 到 127", "存储": "1字节", "用途": "小整数"},
"SMALLINT": {"范围": "-32768 到 32767", "存储": "2字节", "用途": "小整数"},
"MEDIUMINT": {"范围": "-8388608 到 8388607", "存储": "3字节", "用途": "中等整数"},
"INT": {"范围": "-2147483648 到 2147483647", "存储": "4字节", "用途": "标准整数"},
"BIGINT": {"范围": "-9223372036854775808 到 9223372036854775807", "存储": "8字节", "用途": "大整数"}
},
"浮点类型": {
"FLOAT": {"精度": "单精度", "存储": "4字节", "用途": "单精度浮点数"},
"DOUBLE": {"精度": "双精度", "存储": "8字节", "用途": "双精度浮点数"},
"DECIMAL": {"精度": "精确数值", "存储": "可变", "用途": "货币、精确计算"}
}
},
"字符串类型": {
"CHAR": {"长度": "固定长度", "最大": "255字符", "用途": "固定长度字符串"},
"VARCHAR": {"长度": "可变长度", "最大": "65535字符", "用途": "可变长度字符串"},
"TEXT": {"长度": "可变长度", "最大": "65535字符", "用途": "长文本"},
"MEDIUMTEXT": {"长度": "可变长度", "最大": "16777215字符", "用途": "中等长文本"},
"LONGTEXT": {"长度": "可变长度", "最大": "4294967295字符", "用途": "超长文本"}
},
"日期时间类型": {
"DATE": {"格式": "YYYY-MM-DD", "范围": "1000-01-01 到 9999-12-31", "用途": "日期"},
"TIME": {"格式": "HH:MM:SS", "范围": "-838:59:59 到 838:59:59", "用途": "时间"},
"DATETIME": {"格式": "YYYY-MM-DD HH:MM:SS", "范围": "1000-01-01 00:00:00 到 9999-12-31 23:59:59", "用途": "日期时间"},
"TIMESTAMP": {"格式": "YYYY-MM-DD HH:MM:SS", "范围": "1970-01-01 00:00:01 到 2038-01-19 03:14:07", "用途": "时间戳"},
"YEAR": {"格式": "YYYY", "范围": "1901 到 2155", "用途": "年份"}
}
}
def show_data_types(self):
"""显示数据类型"""
print("MySQL数据类型详解:")
print("=" * 60)
for category, subcategories in self.data_types.items():
print(f"\n{category}:")
for subcat, types in subcategories.items():
print(f" {subcat}:")
for type_name, properties in types.items():
print(f" {type_name}:")
for prop, value in properties.items():
print(f" {prop}: {value}")
def data_type_examples(self):
"""数据类型使用示例"""
examples = """
-- 数据类型使用示例
CREATE TABLE data_type_examples (
-- 整数类型
id INT AUTO_INCREMENT PRIMARY KEY,
age TINYINT UNSIGNED, -- 年龄 (0-255)
population BIGINT, -- 人口数量
-- 浮点类型
price DECIMAL(10,2), -- 价格 (精确到分)
weight FLOAT, -- 重量
distance DOUBLE, -- 距离
-- 字符串类型
name VARCHAR(100) NOT NULL, -- 姓名
gender CHAR(1), -- 性别 (M/F)
description TEXT, -- 描述
-- 日期时间类型
birth_date DATE, -- 出生日期
login_time TIME, -- 登录时间
created_at DATETIME DEFAULT CURRENT_TIMESTAMP, -- 创建时间
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, -- 更新时间
birth_year YEAR, -- 出生年份
-- 布尔类型
is_active BOOLEAN DEFAULT TRUE, -- 是否激活
-- 枚举类型
status ENUM('active', 'inactive', 'pending') DEFAULT 'pending'
);
"""
print("\n数据类型使用示例:")
print("=" * 40)
print(examples)
def type_selection_guide(self):
"""数据类型选择指南"""
guide = {
"整数选择": {
"小于256": "TINYINT",
"小于65536": "SMALLINT",
"一般整数": "INT",
"超大整数": "BIGINT"
},
"字符串选择": {
"固定长度且较短": "CHAR",
"可变长度且较短": "VARCHAR",
"长文本内容": "TEXT",
"超长文本内容": "LONGTEXT"
},
"数值精度选择": {
"货币金额": "DECIMAL",
"科学计算": "DOUBLE",
"一般浮点数": "FLOAT"
},
"时间选择": {
"只需要日期": "DATE",
"只需要时间": "TIME",
"需要完整时间戳": "DATETIME",
"需要自动更新时间": "TIMESTAMP"
}
}
print("\n数据类型选择指南:")
print("=" * 40)
for category, choices in guide.items():
print(f"\n{category}:")
for scenario, recommendation in choices.items():
print(f" {scenario}: {recommendation}")
# 数据类型演示
data_types = MySQLDataTypes()
data_types.show_data_types()
data_types.data_type_examples()
data_types.type_selection_guide()
3.2 数据库操作
3.2.1 数据库管理
class DatabaseManager:
def __init__(self):
self.connection = None
def database_operations(self):
"""数据库基本操作"""
operations = {
"创建数据库": {
"基本语法": "CREATE DATABASE database_name;",
"指定字符集": "CREATE DATABASE mydb CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;",
"条件创建": "CREATE DATABASE IF NOT EXISTS mydb;"
},
"查看数据库": {
"显示所有数据库": "SHOW DATABASES;",
"显示创建语句": "SHOW CREATE DATABASE mydb;",
"查看当前数据库": "SELECT DATABASE();"
},
"使用数据库": {
"切换数据库": "USE database_name;"
},
"修改数据库": {
"修改字符集": "ALTER DATABASE mydb CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;"
},
"删除数据库": {
"删除数据库": "DROP DATABASE database_name;",
"条件删除": "DROP DATABASE IF EXISTS database_name;"
}
}
print("数据库基本操作:")
print("=" * 50)
for operation, commands in operations.items():
print(f"\n{operation}:")
for desc, cmd in commands.items():
print(f" {desc}: {cmd}")
def database_examples(self):
"""数据库操作示例"""
examples = """
-- 数据库操作完整示例
-- 1. 创建数据库
CREATE DATABASE company_db
CHARACTER SET utf8mb4
COLLATE utf8mb4_unicode_ci;
-- 2. 查看数据库
SHOW DATABASES;
-- 3. 使用数据库
USE company_db;
-- 4. 查看当前数据库
SELECT DATABASE();
-- 5. 查看数据库创建信息
SHOW CREATE DATABASE company_db;
-- 6. 修改数据库字符集
ALTER DATABASE company_db
CHARACTER SET utf8mb4
COLLATE utf8mb4_general_ci;
-- 7. 删除数据库（谨慎操作）
-- DROP DATABASE IF EXISTS old_database;
"""
print("\n数据库操作示例:")
print("=" * 40)
print(examples)
def character_set_guide(self):
"""字符集选择指南"""
character_sets = {
"utf8mb4": {
"描述": "完整的UTF-8编码，支持4字节字符（如emoji）",
"推荐度": "★★★★★",
"适用场景": "现代应用，支持国际化和emoji",
"排序规则": {
"utf8mb4_unicode_ci": "Unicode标准排序，准确但较慢",
"utf8mb4_general_ci": "简化排序规则，快速但不够准确",
"utf8mb4_bin": "二进制排序，区分大小写"
}
},
"utf8": {
"描述": "MySQL的UTF-8实现，最多3字节（不支持emoji）",
"推荐度": "★★★☆☆",
"适用场景": "旧系统兼容",
"注意": "不建议新项目使用"
},
"latin1": {
"描述": "单字节字符集，仅支持西欧语言",
"推荐度": "★☆☆☆☆",
"适用场景": "仅英文应用",
"注意": "不支持中文等多字节字符"
}
}
print("\n字符集选择指南:")
print("=" * 40)
for charset, info in character_sets.items():
print(f"\n{charset}:")
print(f" 描述: {info['描述']}")
print(f" 推荐度: {info['推荐度']}")
print(f" 适用场景: {info['适用场景']}")
if '排序规则' in info:
print(f" 排序规则:")
for rule, desc in info['排序规则'].items():
print(f" {rule}: {desc}")
if '注意' in info:
print(f" 注意: {info['注意']}")
# 数据库管理演示
db_manager = DatabaseManager()
db_manager.database_operations()
db_manager.database_examples()
db_manager.character_set_guide()
3.2.2 表结构设计
class TableDesigner:
def __init__(self):
self.constraints = {
"PRIMARY KEY": "主键约束，唯一标识表中的每一行",
"FOREIGN KEY": "外键约束，引用其他表的主键",
"UNIQUE": "唯一约束，确保列值唯一",
"NOT NULL": "非空约束，列不能为空",
"CHECK": "检查约束，确保列值满足条件",
"DEFAULT": "默认值约束，为列设置默认值"
}
def table_creation_syntax(self):
"""表创建语法"""
syntax_example = """
-- 表创建基本语法
CREATE TABLE table_name (
column1 datatype [constraints],
column2 datatype [constraints],
...
[table_constraints]
) [table_options];
-- 完整示例
CREATE TABLE employees (
-- 主键列
employee_id INT AUTO_INCREMENT PRIMARY KEY,
-- 基本信息
first_name VARCHAR(50) NOT NULL,
last_name VARCHAR(50) NOT NULL,
email VARCHAR(100) UNIQUE NOT NULL,
phone VARCHAR(20),
-- 工作信息
department_id INT,
position VARCHAR(100),
salary DECIMAL(10,2) CHECK (salary > 0),
hire_date DATE NOT NULL,
-- 状态信息
is_active BOOLEAN DEFAULT TRUE,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
-- 外键约束
FOREIGN KEY (department_id) REFERENCES departments(department_id),
-- 复合索引
INDEX idx_name (last_name, first_name),
INDEX idx_department (department_id),
INDEX idx_hire_date (hire_date)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
"""
print("表创建语法:")
print("=" * 40)
print(syntax_example)
def constraint_examples(self):
"""约束示例"""
print("\n约束类型详解:")
print("=" * 40)
for constraint, description in self.constraints.items():
print(f"{constraint}: {description}")
constraint_examples = """
-- 约束使用示例
CREATE TABLE products (
-- 主键约束
product_id INT AUTO_INCREMENT PRIMARY KEY,
-- 非空约束
product_name VARCHAR(100) NOT NULL,
-- 唯一约束
product_code VARCHAR(50) UNIQUE,
-- 检查约束
price DECIMAL(10,2) CHECK (price >= 0),
stock_quantity INT CHECK (stock_quantity >= 0),
-- 默认值约束
status ENUM('active', 'inactive') DEFAULT 'active',
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
-- 外键约束
category_id INT,
FOREIGN KEY (category_id) REFERENCES categories(category_id)
ON DELETE SET NULL
ON UPDATE CASCADE
);
"""
print("\n约束使用示例:")
print(constraint_examples)
def table_modification(self):
"""表结构修改"""
modifications = {
"添加列": {
"基本语法": "ALTER TABLE table_name ADD COLUMN column_name datatype [constraints];",
"示例": "ALTER TABLE employees ADD COLUMN middle_name VARCHAR(50) AFTER first_name;"
},
"修改列": {
"修改数据类型": "ALTER TABLE table_name MODIFY COLUMN column_name new_datatype;",
"重命名列": "ALTER TABLE table_name CHANGE old_name new_name datatype;",
"示例": "ALTER TABLE employees MODIFY COLUMN salary DECIMAL(12,2);"
},
"删除列": {
"基本语法": "ALTER TABLE table_name DROP COLUMN column_name;",
"示例": "ALTER TABLE employees DROP COLUMN middle_name;"
},
"添加约束": {
"添加主键": "ALTER TABLE table_name ADD PRIMARY KEY (column_name);",
"添加外键": "ALTER TABLE table_name ADD FOREIGN KEY (column_name) REFERENCES other_table(column_name);",
"添加唯一约束": "ALTER TABLE table_name ADD UNIQUE (column_name);"
},
"删除约束": {
"删除主键": "ALTER TABLE table_name DROP PRIMARY KEY;",
"删除外键": "ALTER TABLE table_name DROP FOREIGN KEY constraint_name;",
"删除索引": "ALTER TABLE table_name DROP INDEX index_name;"
}
}
print("\n表结构修改:")
print("=" * 40)
for operation, details in modifications.items():
print(f"\n{operation}:")
for desc, syntax in details.items():
print(f" {desc}: {syntax}")
def table_management_examples(self):
"""表管理示例"""
examples = """
-- 表管理完整示例
-- 1. 查看所有表
SHOW TABLES;
-- 2. 查看表结构
DESC employees;
-- 或者
SHOW COLUMNS FROM employees;
-- 3. 查看表创建语句
SHOW CREATE TABLE employees;
-- 4. 查看表状态
SHOW TABLE STATUS LIKE 'employees';
-- 5. 重命名表
RENAME TABLE old_table_name TO new_table_name;
-- 6. 复制表结构
CREATE TABLE employees_backup LIKE employees;
-- 7. 复制表结构和数据
CREATE TABLE employees_copy AS SELECT * FROM employees;
-- 8. 清空表数据
TRUNCATE TABLE table_name; -- 快速清空，重置AUTO_INCREMENT
-- 或者
DELETE FROM table_name; -- 逐行删除，不重置AUTO_INCREMENT
-- 9. 删除表
DROP TABLE IF EXISTS table_name;
"""
print("\n表管理示例:")
print("=" * 30)
print(examples)
# 表设计演示
table_designer = TableDesigner()
table_designer.table_creation_syntax()
table_designer.constraint_examples()
table_designer.table_modification()
table_designer.table_management_examples()
3.3 数据操作（DML）
3.3.1 数据插入（INSERT）
class DataInsertion:
def __init__(self):
self.insert_methods = {
"单行插入": "INSERT INTO table_name (columns) VALUES (values);",
"多行插入": "INSERT INTO table_name (columns) VALUES (values1), (values2), ...;",
"从查询插入": "INSERT INTO table_name (columns) SELECT columns FROM other_table WHERE condition;",
"替换插入": "REPLACE INTO table_name (columns) VALUES (values);",
"忽略重复": "INSERT IGNORE INTO table_name (columns) VALUES (values);",
"更新重复": "INSERT INTO table_name (columns) VALUES (values) ON DUPLICATE KEY UPDATE column=value;"
}
def insert_syntax_examples(self):
"""插入语法示例"""
print("数据插入语法:")
print("=" * 40)
for method, syntax in self.insert_methods.items():
print(f"{method}: {syntax}")
detailed_examples = """
-- 详细插入示例
-- 1. 基本插入
INSERT INTO employees (first_name, last_name, email, department_id, salary, hire_date)
VALUES ('John', 'Doe', 'john.doe@company.com', 1, 50000.00, '2024-01-15');
-- 2. 多行插入
INSERT INTO employees (first_name, last_name, email, department_id, salary, hire_date)
VALUES
('Jane', 'Smith', 'jane.smith@company.com', 2, 55000.00, '2024-01-16'),
('Bob', 'Johnson', 'bob.johnson@company.com', 1, 48000.00, '2024-01-17'),
('Alice', 'Brown', 'alice.brown@company.com', 3, 60000.00, '2024-01-18');
-- 3. 部分列插入（其他列使用默认值）
INSERT INTO employees (first_name, last_name, email)
VALUES ('Mike', 'Wilson', 'mike.wilson@company.com');
-- 4. 从其他表插入
INSERT INTO employees_archive (employee_id, first_name, last_name, email)
SELECT employee_id, first_name, last_name, email
FROM employees
WHERE is_active = FALSE;
-- 5. 忽略重复键错误
INSERT IGNORE INTO employees (email, first_name, last_name)
VALUES ('existing@company.com', 'Test', 'User');
-- 6. 重复键时更新
INSERT INTO employees (email, first_name, last_name, salary)
VALUES ('john.doe@company.com', 'John', 'Doe', 55000.00)
ON DUPLICATE KEY UPDATE
salary = VALUES(salary),
updated_at = CURRENT_TIMESTAMP;
-- 7. 替换插入（删除旧记录，插入新记录）
REPLACE INTO employees (employee_id, first_name, last_name, email, salary)
VALUES (1, 'John', 'Doe', 'john.doe@company.com', 60000.00);
"""
print("\n详细插入示例:")
print(detailed_examples)
def batch_insert_optimization(self):
"""批量插入优化"""
optimization_tips = {
"使用多行INSERT": {
"说明": "一次插入多行比多次单行插入效率更高",
"示例": "INSERT INTO table VALUES (1,'a'), (2,'b'), (3,'c');"
},
"禁用自动提交": {
"说明": "批量操作时禁用自动提交，手动控制事务",
"示例": "SET autocommit=0; INSERT ...; COMMIT;"
},
"调整批次大小": {
"说明": "每批插入1000-5000行，避免单次操作过大",
"建议": "根据内存和网络情况调整"
},
"使用LOAD DATA": {
"说明": "从文件批量导入数据，效率最高",
"示例": "LOAD DATA INFILE 'data.csv' INTO TABLE employees;"
},
"临时禁用索引": {
"说明": "大批量插入时可临时禁用非唯一索引",
"示例": "ALTER TABLE table DISABLE KEYS; INSERT ...; ALTER TABLE table ENABLE KEYS;"
}
}
print("\n批量插入优化技巧:")
print("=" * 40)
for tip, details in optimization_tips.items():
print(f"\n{tip}:")
print(f" 说明: {details['说明']}")
if '示例' in details:
print(f" 示例: {details['示例']}")
if '建议' in details:
print(f" 建议: {details['建议']}")
def load_data_examples(self):
"""LOAD DATA示例"""
examples = """
-- LOAD DATA批量导入示例
-- 1. 从本地文件导入
LOAD DATA LOCAL INFILE '/path/to/employees.csv'
INTO TABLE employees
FIELDS TERMINATED BY ','
LINES TERMINATED BY '\n'
IGNORE 1 ROWS -- 跳过标题行
(first_name, last_name, email, department_id, salary, hire_date);
-- 2. 指定字段分隔符和包围符
LOAD DATA LOCAL INFILE '/path/to/data.csv'
INTO TABLE employees
FIELDS TERMINATED BY ','
ENCLOSED BY '"'
LINES TERMINATED BY '\n'
(first_name, last_name, email, @salary, hire_date)
SET salary = @salary * 1.1; -- 数据转换
-- 3. 处理错误和重复
LOAD DATA LOCAL INFILE '/path/to/data.csv'
INTO TABLE employees
FIELDS TERMINATED BY ','
LINES TERMINATED BY '\n'
IGNORE 1 ROWS
REPLACE -- 或使用 IGNORE
(first_name, last_name, email, department_id, salary, hire_date);
-- 4. CSV文件示例格式
-- employees.csv:
-- first_name,last_name,email,department_id,salary,hire_date
-- John,Doe,john.doe@company.com,1,50000.00,2024-01-15
-- Jane,Smith,jane.smith@company.com,2,55000.00,2024-01-16
"""
print("\nLOAD DATA导入示例:")
print("=" * 40)
print(examples)
# 数据插入演示
data_insertion = DataInsertion()
data_insertion.insert_syntax_examples()
data_insertion.batch_insert_optimization()
data_insertion.load_data_examples()
3.3.2 数据查询（SELECT）
class DataQuery:
def __init__(self):
self.select_clauses = {
"SELECT": "指定要查询的列",
"FROM": "指定数据源表",
"WHERE": "行过滤条件",
"GROUP BY": "分组",
"HAVING": "分组后过滤",
"ORDER BY": "排序",
"LIMIT": "限制结果数量"
}
def basic_select_syntax(self):
"""基本查询语法"""
print("SELECT查询语法:")
print("=" * 40)
for clause, description in self.select_clauses.items():
print(f"{clause}: {description}")
basic_examples = """
-- 基本查询示例
-- 1. 查询所有列
SELECT * FROM employees;
-- 2. 查询指定列
SELECT first_name, last_name, email FROM employees;
-- 3. 使用别名
SELECT
first_name AS '名',
last_name AS '姓',
salary AS '工资'
FROM employees;
-- 4. 计算列
SELECT
first_name,
last_name,
salary,
salary * 12 AS annual_salary,
CONCAT(first_name, ' ', last_name) AS full_name
FROM employees;
-- 5. 去重查询
SELECT DISTINCT department_id FROM employees;
-- 6. 限制结果数量
SELECT * FROM employees LIMIT 10;
SELECT * FROM employees LIMIT 10, 5; -- 跳过10行，取5行
"""
print("\n基本查询示例:")
print(basic_examples)
def where_clause_examples(self):
"""WHERE子句示例"""
where_examples = """
-- WHERE条件查询示例
-- 1. 比较运算符
SELECT * FROM employees WHERE salary > 50000;
SELECT * FROM employees WHERE department_id = 1;
SELECT * FROM employees WHERE hire_date >= '2024-01-01';
-- 2. 逻辑运算符
SELECT * FROM employees
WHERE salary > 50000 AND department_id = 1;
SELECT * FROM employees
WHERE department_id = 1 OR department_id = 2;
SELECT * FROM employees
WHERE NOT (salary < 40000);
-- 3. 范围查询
SELECT * FROM employees
WHERE salary BETWEEN 40000 AND 60000;
SELECT * FROM employees
WHERE department_id IN (1, 2, 3);
-- 4. 模糊查询
SELECT * FROM employees
WHERE first_name LIKE 'J%'; -- 以J开头
SELECT * FROM employees
WHERE email LIKE '%@company.com'; -- 以@company.com结尾
SELECT * FROM employees
WHERE first_name LIKE '_ohn'; -- 第一个字符任意，后面是ohn
-- 5. 空值查询
SELECT * FROM employees WHERE phone IS NULL;
SELECT * FROM employees WHERE phone IS NOT NULL;
-- 6. 正则表达式
SELECT * FROM employees
WHERE email REGEXP '^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$';
"""
print("\nWHERE条件查询:")
print("=" * 30)
print(where_examples)
def aggregate_functions(self):
"""聚合函数"""
functions = {
"COUNT()": "计数",
"SUM()": "求和",
"AVG()": "平均值",
"MAX()": "最大值",
"MIN()": "最小值",
"GROUP_CONCAT()": "字符串连接",
"STDDEV()": "标准差",
"VARIANCE()": "方差"
}
print("\n聚合函数:")
print("=" * 30)
for func, desc in functions.items():
print(f"{func}: {desc}")
aggregate_examples = """
-- 聚合函数示例
-- 1. 基本聚合
SELECT COUNT(*) AS total_employees FROM employees;
SELECT COUNT(phone) AS employees_with_phone FROM employees; -- 不计算NULL值
SELECT SUM(salary) AS total_salary FROM employees;
SELECT AVG(salary) AS average_salary FROM employees;
SELECT MAX(salary) AS highest_salary FROM employees;
SELECT MIN(hire_date) AS earliest_hire_date FROM employees;
-- 2. 分组聚合
SELECT
department_id,
COUNT(*) AS employee_count,
AVG(salary) AS avg_salary,
MAX(salary) AS max_salary,
MIN(salary) AS min_salary
FROM employees
GROUP BY department_id;
-- 3. 多列分组
SELECT
department_id,
YEAR(hire_date) AS hire_year,
COUNT(*) AS employee_count
FROM employees
GROUP BY department_id, YEAR(hire_date)
ORDER BY department_id, hire_year;
-- 4. HAVING子句（分组后过滤）
SELECT
department_id,
COUNT(*) AS employee_count,
AVG(salary) AS avg_salary
FROM employees
GROUP BY department_id
HAVING COUNT(*) > 5 AND AVG(salary) > 50000;
-- 5. 字符串聚合
SELECT
department_id,
GROUP_CONCAT(CONCAT(first_name, ' ', last_name) SEPARATOR ', ') AS employee_names
FROM employees
GROUP BY department_id;
"""
print("\n聚合函数示例:")
print(aggregate_examples)
def join_operations(self):
"""连接操作"""
join_types = {
"INNER JOIN": "内连接，返回两表中匹配的记录",
"LEFT JOIN": "左连接，返回左表所有记录和右表匹配记录",
"RIGHT JOIN": "右连接，返回右表所有记录和左表匹配记录",
"FULL OUTER JOIN": "全外连接，返回两表所有记录（MySQL不直接支持）",
"CROSS JOIN": "交叉连接，返回两表的笛卡尔积",
"SELF JOIN": "自连接，表与自身连接"
}
print("\n连接操作类型:")
print("=" * 30)
for join_type, description in join_types.items():
print(f"{join_type}: {description}")
join_examples = """
-- 连接操作示例
-- 假设有以下表结构：
-- employees: employee_id, first_name, last_name, department_id
-- departments: department_id, department_name, manager_id
-- positions: position_id, position_name, department_id
-- 1. 内连接
SELECT
e.first_name,
e.last_name,
d.department_name
FROM employees e
INNER JOIN departments d ON e.department_id = d.department_id;
-- 2. 左连接
SELECT
e.first_name,
e.last_name,
d.department_name
FROM employees e
LEFT JOIN departments d ON e.department_id = d.department_id;
-- 3. 右连接
SELECT
e.first_name,
e.last_name,
d.department_name
FROM employees e
RIGHT JOIN departments d ON e.department_id = d.department_id;
-- 4. 多表连接
SELECT
e.first_name,
e.last_name,
d.department_name,
p.position_name
FROM employees e
INNER JOIN departments d ON e.department_id = d.department_id
INNER JOIN positions p ON e.position_id = p.position_id;
-- 5. 自连接（查找同部门同事）
SELECT
e1.first_name AS employee,
e2.first_name AS colleague
FROM employees e1
INNER JOIN employees e2 ON e1.department_id = e2.department_id
WHERE e1.employee_id != e2.employee_id;
-- 6. 使用USING简化连接条件
SELECT
e.first_name,
e.last_name,
d.department_name
FROM employees e
INNER JOIN departments d USING (department_id);
"""
print("\n连接操作示例:")
print(join_examples)
def subquery_examples(self):
"""子查询示例"""
subquery_examples = """
-- 子查询示例
-- 1. 标量子查询（返回单个值）
SELECT
first_name,
last_name,
salary,
(SELECT AVG(salary) FROM employees) AS avg_salary
FROM employees
WHERE salary > (SELECT AVG(salary) FROM employees);
-- 2. 列子查询（返回一列多行）
SELECT first_name, last_name
FROM employees
WHERE department_id IN (
SELECT department_id
FROM departments
WHERE department_name IN ('IT', 'Sales')
);
-- 3. 行子查询（返回一行多列）
SELECT first_name, last_name
FROM employees
WHERE (department_id, salary) = (
SELECT department_id, MAX(salary)
FROM employees
GROUP BY department_id
LIMIT 1
);
-- 4. 表子查询（返回多行多列）
SELECT *
FROM (
SELECT
department_id,
AVG(salary) AS avg_salary,
COUNT(*) AS employee_count
FROM employees
GROUP BY department_id
) AS dept_stats
WHERE avg_salary > 50000;
-- 5. 相关子查询
SELECT
e1.first_name,
e1.last_name,
e1.salary
FROM employees e1
WHERE e1.salary > (
SELECT AVG(e2.salary)
FROM employees e2
WHERE e2.department_id = e1.department_id
);
-- 6. EXISTS子查询
SELECT first_name, last_name
FROM employees e
WHERE EXISTS (
SELECT 1
FROM departments d
WHERE d.department_id = e.department_id
AND d.department_name = 'IT'
);
-- 7. NOT EXISTS子查询
SELECT first_name, last_name
FROM employees e
WHERE NOT EXISTS (
SELECT 1
FROM employee_projects ep
WHERE ep.employee_id = e.employee_id
);
"""
print("\n子查询示例:")
print("=" * 30)
print(subquery_examples)
# 数据查询演示
data_query = DataQuery()
data_query.basic_select_syntax()
data_query.where_clause_examples()
data_query.aggregate_functions()
data_query.join_operations()
data_query.subquery_examples()
3.3.3 数据更新和删除
class DataModification:
def __init__(self):
self.update_types = {
"单表更新": "UPDATE table SET column=value WHERE condition;",
"多表更新": "UPDATE table1 JOIN table2 SET table1.column=value WHERE condition;",
"条件更新": "UPDATE table SET column=CASE WHEN condition THEN value END;"
}
def update_operations(self):
"""更新操作"""
print("数据更新操作:")
print("=" * 40)
for update_type, syntax in self.update_types.items():
print(f"{update_type}: {syntax}")
update_examples = """
-- 更新操作示例
-- 1. 基本更新
UPDATE employees
SET salary = 55000
WHERE employee_id = 1;
-- 2. 多列更新
UPDATE employees
SET
salary = salary * 1.1,
updated_at = CURRENT_TIMESTAMP
WHERE department_id = 1;
-- 3. 基于计算的更新
UPDATE employees
SET
salary = CASE
WHEN YEAR(hire_date) < 2020 THEN salary * 1.15
WHEN YEAR(hire_date) < 2022 THEN salary * 1.10
ELSE salary * 1.05
END
WHERE is_active = TRUE;
-- 4. 基于子查询的更新
UPDATE employees
SET salary = (
SELECT AVG(salary) * 1.1
FROM (
SELECT salary
FROM employees e2
WHERE e2.department_id = employees.department_id
) AS dept_avg
)
WHERE performance_rating = 'Excellent';
-- 5. 多表连接更新
UPDATE employees e
INNER JOIN departments d ON e.department_id = d.department_id
SET e.salary = e.salary * 1.1
WHERE d.department_name = 'IT';
-- 6. 使用LIMIT限制更新行数
UPDATE employees
SET salary = salary * 1.05
WHERE department_id = 1
ORDER BY hire_date
LIMIT 10;
-- 7. 安全更新（避免全表更新）
-- 启用安全模式
SET SQL_SAFE_UPDATES = 1;
-- 这将阻止没有WHERE子句或LIMIT的UPDATE
UPDATE employees SET salary = salary * 1.1; -- 会报错
-- 正确的做法
UPDATE employees
SET salary = salary * 1.1
WHERE employee_id > 0; -- 明确的WHERE条件
"""
print("\n更新操作示例:")
print(update_examples)
def delete_operations(self):
"""删除操作"""
delete_types = {
"DELETE": "逐行删除，可以回滚，触发器会执行",
"TRUNCATE": "快速清空表，不可回滚，重置AUTO_INCREMENT",
"DROP": "删除整个表结构和数据"
}
print("\n删除操作类型:")
print("=" * 40)
for delete_type, description in delete_types.items():
print(f"{delete_type}: {description}")
delete_examples = """
-- 删除操作示例
-- 1. 基本删除
DELETE FROM employees WHERE employee_id = 1;
-- 2. 条件删除
DELETE FROM employees
WHERE is_active = FALSE
AND hire_date < '2020-01-01';
-- 3. 基于子查询的删除
DELETE FROM employees
WHERE department_id IN (
SELECT department_id
FROM departments
WHERE department_name = 'Discontinued'
);
-- 4. 多表删除
DELETE e
FROM employees e
INNER JOIN departments d ON e.department_id = d.department_id
WHERE d.department_name = 'Temp';
-- 5. 使用LIMIT限制删除行数
DELETE FROM employees
WHERE is_active = FALSE
ORDER BY hire_date
LIMIT 100;
-- 6. 软删除（推荐做法）
-- 不实际删除数据，而是标记为删除状态
UPDATE employees
SET
is_deleted = TRUE,
deleted_at = CURRENT_TIMESTAMP
WHERE employee_id = 1;
-- 查询时过滤已删除的记录
SELECT * FROM employees WHERE is_deleted = FALSE;
-- 7. 清空表
TRUNCATE TABLE temp_table; -- 快速清空
-- 或
DELETE FROM temp_table; -- 逐行删除
-- 8. 删除重复记录
-- 保留ID最小的记录，删除其他重复记录
DELETE e1 FROM employees e1
INNER JOIN employees e2
WHERE e1.email = e2.email
AND e1.employee_id > e2.employee_id;
-- 9. 安全删除
SET SQL_SAFE_UPDATES = 1;
-- 这将阻止没有WHERE子句的DELETE
DELETE FROM employees; -- 会报错
-- 正确的做法
DELETE FROM employees WHERE employee_id > 0;
"""
print("\n删除操作示例:")
print(delete_examples)
def transaction_examples(self):
"""事务示例"""
transaction_examples = """
-- 事务操作示例
-- 1. 基本事务
START TRANSACTION;
UPDATE accounts SET balance = balance - 1000 WHERE account_id = 1;
UPDATE accounts SET balance = balance + 1000 WHERE account_id = 2;
-- 检查操作是否成功
SELECT balance FROM accounts WHERE account_id IN (1, 2);
-- 提交事务
COMMIT;
-- 2. 事务回滚
START TRANSACTION;
UPDATE employees SET salary = salary * 2;
-- 发现错误，回滚事务
ROLLBACK;
-- 3. 保存点
START TRANSACTION;
UPDATE employees SET salary = salary * 1.1 WHERE department_id = 1;
SAVEPOINT sp1;
UPDATE employees SET salary = salary * 1.2 WHERE department_id = 2;
SAVEPOINT sp2;
UPDATE employees SET salary = salary * 1.3 WHERE department_id = 3;
-- 回滚到保存点sp2
ROLLBACK TO sp2;
-- 提交事务
COMMIT;
-- 4. 自动提交控制
-- 查看当前自动提交状态
SELECT @@autocommit;
-- 禁用自动提交
SET autocommit = 0;
-- 执行多个操作
INSERT INTO employees (first_name, last_name) VALUES ('Test', 'User1');
INSERT INTO employees (first_name, last_name) VALUES ('Test', 'User2');
-- 手动提交
COMMIT;
-- 恢复自动提交
SET autocommit = 1;
-- 5. 事务隔离级别
-- 查看当前隔离级别
SELECT @@transaction_isolation;
-- 设置隔离级别
SET SESSION TRANSACTION ISOLATION LEVEL READ COMMITTED;
-- 隔离级别说明：
-- READ UNCOMMITTED: 可以读取未提交的数据（脏读）
-- READ COMMITTED: 只能读取已提交的数据（不可重复读）
-- REPEATABLE READ: 可重复读（MySQL默认级别）
-- SERIALIZABLE: 串行化，最高隔离级别
"""
print("\n事务操作示例:")
print("=" * 30)
print(transaction_examples)
def data_modification_best_practices(self):
"""数据修改最佳实践"""
best_practices = {
"安全性": [
"始终使用WHERE子句，避免全表更新/删除",
"启用SQL_SAFE_UPDATES模式",
"在生产环境操作前先在测试环境验证",
"重要操作前先备份数据"
],
"性能": [
"批量操作使用事务包装",
"大批量删除使用LIMIT分批处理",
"更新操作尽量使用索引列作为条件",
"避免在WHERE子句中使用函数"
],
"数据完整性": [
"使用事务确保数据一致性",
"重要数据使用软删除而非物理删除",
"更新前验证数据约束",
"使用外键约束维护引用完整性"
],
"可维护性": [
"记录重要的数据变更操作",
"使用有意义的变量名和注释",
"建立数据变更审计机制",
"定期清理历史数据"
]
}
print("\n数据修改最佳实践:")
print("=" * 40)
for category, practices in best_practices.items():
print(f"\n{category}:")
for practice in practices:
print(f" - {practice}")
# 数据修改演示
data_modification = DataModification()
data_modification.update_operations()
data_modification.delete_operations()
data_modification.transaction_examples()
data_modification.data_modification_best_practices()
3.4 总结
本章详细介绍了MySQL的基本操作与SQL语法，包括：
核心内容
- SQL语言分类：DDL、DML、DCL、TCL四大类别
- 数据类型：数值、字符串、日期时间类型的选择和使用
- 数据库操作：创建、修改、删除数据库和表
- 数据操作：INSERT、SELECT、UPDATE、DELETE的各种用法
关键技能
查询技能：
- 基本查询和条件过滤
- 聚合函数和分组统计
- 多表连接和子查询
- 复杂查询的优化
数据修改：
- 安全的更新和删除操作
- 事务控制和数据一致性
- 批量操作的优化技巧
最佳实践：
- 使用适当的数据类型
- 编写高效的查询语句
- 保证数据安全和完整性
- 遵循SQL编码规范
实用技巧
性能优化：
- 合理使用索引
- 避免全表扫描
- 优化JOIN操作
- 使用EXPLAIN分析查询
安全考虑：
- 启用SQL_SAFE_UPDATES
- 使用参数化查询防止SQL注入
- 定期备份重要数据
- 建立完善的权限控制
下一章我们将学习MySQL的索引机制，包括索引的类型、创建方法、优化策略等内容，这对提升数据库查询性能至关重要。
