当前位置： 首页 > news >正文

关于MySQL与Python后端命令交互备份

news 2025/9/20 5:53:56

关于发现自己关于数据库和后端与数据库的交互方面有些问题，因此复习一下；

突然遇到了一个需要后端数据库通信的场景，不知道怎么解决，查了下AI具体实现的流程
连接数据库——>查看所有表——>添加、查取、删除、访问 等操作实现

  • 具体代码
#!/usr/bin/env python3
import mysql.connectordef find_key():try:# 连接到MySQL数据库conn = mysql.connector.connect(host="localhost",user="root",password="",database="lanqiao")cursor = conn.cursor()# 获取所有表cursor.execute("SHOW TABLES")tables = cursor.fetchall()print("找到的表:")for table in tables:print(table[0])# 检查每个表的结构for table in tables:table_name = table[0]print(f"\n检查表: {table_name}")# 查看表结构cursor.execute(f"DESCRIBE {table_name}")columns = cursor.fetchall()print("表结构:")for column in columns:print(f"  列名: {column[0]}, 类型: {column[1]}")# 查看表内容（前几行）try:cursor.execute(f"SELECT * FROM {table_name} LIMIT 5")rows = cursor.fetchall()print("表内容:")for i, row in enumerate(rows):print(f"  第{i+1}行: {row}")# 检查是否有看起来像密钥的值for value in row:if value and (isinstance(value, str) and ('key' in str(value).lower() or len(str(value)) > 10)):print(f"  发现可能密钥: {value}")return valueexcept Exception as e:print(f"  查询表内容时出错: {e}")continuecursor.close()conn.close()return Noneexcept Exception as e:print(f"错误: {e}")return Nonedef save_to_file(key_value):# 将密钥保存到key.txt文件with open('key.txt', 'w') as f:f.write(str(key_value))print(f"密钥已保存到 key.txt: {key_value}")if __name__ == "__main__":key = find_key()if key:save_to_file(key)print("任务完成！请点击左下角的【提交检测】按钮。")else:print("未找到明确的密钥，尝试其他搜索方法...")# 如果没有找到，尝试更详细的搜索try:conn = mysql.connector.connect(host="localhost",user="root",password="",database="lanqiao")cursor = conn.cursor()# 检查所有表中的所有数据cursor.execute("SHOW TABLES")tables = cursor.fetchall()for table in tables:table_name = table[0]try:cursor.execute(f"SELECT * FROM {table_name}")rows = cursor.fetchall()for row in rows:for value in row:if value and isinstance(value, str) and len(value) > 8:# 可能是密钥print(f"在表 {table_name} 中找到可能密钥: {value}")with open('key.txt', 'w') as f:f.write(str(value))print("已保存到 key.txt")exit(0)except:continuecursor.close()conn.close()except Exception as e:print(f"详细搜索时出错: {e}")# 创建默认测试文件with open('key.txt', 'w') as f:f.write("default_test_key_12345")print("已创建默认测试文件")

关于Mysql语法这方面也记录下

MySQL 基础命令大全

1. 连接和基本操作

# 连接MySQL服务器
mysql -u root -p
mysql -u root -p database_name
mysql -h hostname -u username -p# 查看当前数据库
SELECT DATABASE();# 查看MySQL版本
SELECT VERSION();# 显示所有数据库
SHOW DATABASES;# 使用特定数据库
USE database_name;# 显示当前用户
SELECT USER();# 退出MySQL
EXIT;
QUIT;
\q

2. 表操作命令

-- 显示所有表
SHOW TABLES;-- 显示表结构
DESCRIBE table_name;
DESC table_name;
SHOW COLUMNS FROM table_name;-- 显示表创建语句
SHOW CREATE TABLE table_name;-- 创建新表
CREATE TABLE users (id INT AUTO_INCREMENT PRIMARY KEY,username VARCHAR(50) NOT NULL UNIQUE,email VARCHAR(100),created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);-- 删除表
DROP TABLE table_name;-- 重命名表
RENAME TABLE old_name TO new_name;-- 修改表结构
ALTER TABLE table_name ADD COLUMN new_column VARCHAR(100);
ALTER TABLE table_name DROP COLUMN column_name;
ALTER TABLE table_name MODIFY COLUMN column_name NEW_DATA_TYPE;

3. 数据查询命令 (DQL)

-- 基本查询
SELECT * FROM table_name;
SELECT column1, column2 FROM table_name;
SELECT DISTINCT column_name FROM table_name;-- 条件查询
SELECT * FROM table_name WHERE condition;
SELECT * FROM users WHERE age > 18;
SELECT * FROM products WHERE price BETWEEN 10 AND 100;
SELECT * FROM users WHERE name LIKE 'J%';
SELECT * FROM users WHERE email IS NOT NULL;-- 排序和限制
SELECT * FROM table_name ORDER BY column_name ASC/DESC;
SELECT * FROM users ORDER BY created_at DESC LIMIT 10;
SELECT * FROM products ORDER BY price ASC LIMIT 5 OFFSET 10;-- 聚合函数
SELECT COUNT(*) FROM table_name;
SELECT AVG(price) FROM products;
SELECT SUM(sales) FROM orders;
SELECT MAX(age), MIN(age) FROM users;
SELECT category, COUNT(*) FROM products GROUP BY category;-- 连接查询
SELECT users.name, orders.amount 
FROM users 
INNER JOIN orders ON users.id = orders.user_id;SELECT users.name, orders.amount 
FROM users 
LEFT JOIN orders ON users.id = orders.user_id;

4. 数据操作命令 (DML)

-- 插入数据
INSERT INTO table_name (column1, column2) VALUES (value1, value2);
INSERT INTO users (username, email) VALUES ('john', 'john@example.com');-- 更新数据
UPDATE table_name SET column1 = value1 WHERE condition;
UPDATE users SET email = 'new@email.com' WHERE id = 1;-- 删除数据
DELETE FROM table_name WHERE condition;
DELETE FROM users WHERE id = 5;-- 清空表
TRUNCATE TABLE table_name;

5. 信息模式查询 (INFORMATION_SCHEMA)

-- 查看所有数据库信息
SELECT * FROM INFORMATION_SCHEMA.SCHEMATA;-- 查看所有表信息
SELECT TABLE_NAME, TABLE_TYPE, ENGINE 
FROM INFORMATION_SCHEMA.TABLES 
WHERE TABLE_SCHEMA = 'database_name';-- 查看列信息
SELECT COLUMN_NAME, DATA_TYPE, IS_NULLABLE, COLUMN_DEFAULT
FROM INFORMATION_SCHEMA.COLUMNS 
WHERE TABLE_SCHEMA = 'database_name' AND TABLE_NAME = 'table_name';-- 查看索引信息
SELECT INDEX_NAME, COLUMN_NAME, NON_UNIQUE
FROM INFORMATION_SCHEMA.STATISTICS
WHERE TABLE_SCHEMA = 'database_name' AND TABLE_NAME = 'table_name';

6. 用户和权限管理

-- 创建用户
CREATE USER 'username'@'localhost' IDENTIFIED BY 'password';-- 授予权限
GRANT SELECT, INSERT ON database_name.* TO 'username'@'localhost';
GRANT ALL PRIVILEGES ON database_name.* TO 'username'@'localhost';-- 撤销权限
REVOKE INSERT ON database_name.* FROM 'username'@'localhost';-- 查看用户权限
SHOW GRANTS FOR 'username'@'localhost';-- 删除用户
DROP USER 'username'@'localhost';

7. 备份和恢复

# 备份整个数据库
mysqldump -u root -p database_name > backup.sql# 备份特定表
mysqldump -u root -p database_name table1 table2 > backup.sql# 恢复数据库
mysql -u root -p database_name < backup.sql# 只备份结构
mysqldump -u root -p --no-data database_name > structure.sql# 只备份数据
mysqldump -u root -p --no-create-info database_name > data.sql

8. 实用技巧和函数

-- 字符串函数
SELECT CONCAT(first_name, ' ', last_name) AS full_name FROM users;
SELECT UPPER(name), LOWER(email) FROM users;
SELECT SUBSTRING(description, 1, 100) FROM products;
SELECT LENGTH(username) FROM users;-- 日期函数
SELECT NOW(), CURDATE(), CURTIME();
SELECT DATE_ADD(created_at, INTERVAL 7 DAY) FROM orders;
SELECT DATEDIFF(NOW(), created_at) FROM users;-- 条件函数
SELECT name, CASE WHEN age < 18 THEN 'Minor' WHEN age BETWEEN 18 AND 65 THEN 'Adult' ELSE 'Senior' END AS age_group
FROM users;-- 数学函数
SELECT ROUND(price, 2), CEIL(price), FLOOR(price) FROM products;
SELECT RAND();  -- 随机数-- 查看进程和性能
SHOW PROCESSLIST;
SHOW STATUS LIKE 'Threads_connected';
SHOW VARIABLES LIKE 'max_connections';

9. Python MySQL 连接示例

import mysql.connector
from mysql.connector import Errordef connect_to_mysql():try:connection = mysql.connector.connect(host='localhost',database='lanqiao',user='root',password='')if connection.is_connected():print("成功连接到MySQL数据库")# 执行查询cursor = connection.cursor()cursor.execute("SELECT DATABASE()")db_name = cursor.fetchone()print(f"当前数据库: {db_name[0]}")# 获取所有表cursor.execute("SHOW TABLES")tables = cursor.fetchall()print("所有表:", [table[0] for table in tables])except Error as e:print(f"连接错误: {e}")finally:if connection.is_connected():cursor.close()connection.close()print("MySQL连接已关闭")# 使用上下文管理器的方式
def safe_query():try:with mysql.connector.connect(host='localhost',database='lanqiao',user='root',password='') as connection:with connection.cursor() as cursor:cursor.execute("SELECT * FROM your_table LIMIT 5")results = cursor.fetchall()for row in results:print(row)except Error as e:print(f"查询错误: {e}")if __name__ == "__main__":connect_to_mysql()

10. 故障排除和调试

-- 查看错误日志位置
SHOW VARIABLES LIKE 'log_error';-- 查看慢查询日志
SHOW VARIABLES LIKE 'slow_query_log';-- 查看当前连接数
SHOW STATUS LIKE 'Threads_connected';-- 查看表状态
SHOW TABLE STATUS LIKE 'table_name';-- 分析查询性能
EXPLAIN SELECT * FROM users WHERE age > 18;

11. 安全最佳实践

-- 使用预处理语句防止SQL注入
PREPARE stmt FROM 'SELECT * FROM users WHERE id = ?';
SET @id = 1;
EXECUTE stmt USING @id;-- 定期更改密码
ALTER USER 'username'@'localhost' IDENTIFIED BY 'new_password';-- 限制用户权限（最小权限原则）
GRANT SELECT ON database_name.table_name TO 'username'@'localhost';

常用的应该基本就这些，刚接触这种场景，等下次复现操作有问题的化在增改吧；

看到您正在系统性地复习数据库和后端交互知识，这是非常好的学习方式！我来为您整理一个更完整、更结构化的MySQL与Python交互指南。

🗃️ 关于MySQL与Python后端交互

1. 数据库连接管理（后期查阅用）

import mysql.connector
from mysql.connector import Error, pooling
import contextlib
from typing import List, Dict, Any, Optionalclass DatabaseManager:def __init__(self, config: dict):self.config = configself.connection_pool = Nonedef create_connection_pool(self, pool_size: int = 5):"""创建数据库连接池"""try:self.connection_pool = pooling.MySQLConnectionPool(pool_name="mysql_pool",pool_size=pool_size,**self.config)print("数据库连接池创建成功")except Error as e:print(f"连接池创建失败: {e}")raise@contextlib.contextmanagerdef get_connection(self):"""获取数据库连接（上下文管理器）"""if not self.connection_pool:self.create_connection_pool()conn = self.connection_pool.get_connection()try:yield connfinally:conn.close()@contextlib.contextmanagerdef get_cursor(self, dictionary: bool = True):"""获取游标（上下文管理器）"""with self.get_connection() as conn:cursor = conn.cursor(dictionary=dictionary)try:yield cursorconn.commit()  # 自动提交事务except Exception as e:conn.rollback()  # 出错时回滚raise efinally:cursor.close()# 配置示例
db_config = {'host': 'localhost','database': 'lanqiao','user': 'root','password': '','autocommit': True,'charset': 'utf8mb4','pool_reset_session': True
}db_manager = DatabaseManager(db_config)

2. 完整的CRUD操作类

class CRUDOperations:def __init__(self, db_manager: DatabaseManager):self.db = db_managerdef execute_query(self, query: str, params: tuple = None) -> List[Dict]:"""执行查询并返回结果"""try:with self.db.get_cursor(dictionary=True) as cursor:cursor.execute(query, params)return cursor.fetchall()except Error as e:print(f"查询执行失败: {e}")return []def execute_command(self, query: str, params: tuple = None) -> int:"""执行写操作并返回影响行数"""try:with self.db.get_cursor() as cursor:cursor.execute(query, params)return cursor.rowcountexcept Error as e:print(f"命令执行失败: {e}")return 0# 表操作def get_all_tables(self) -> List[str]:"""获取所有表名"""result = self.execute_query("SHOW TABLES")return [list(table.values())[0] for table in result]def get_table_structure(self, table_name: str) -> List[Dict]:"""获取表结构"""return self.execute_query(f"DESCRIBE {table_name}")def get_table_data(self, table_name: str, limit: int = 10) -> List[Dict]:"""获取表数据"""return self.execute_query(f"SELECT * FROM {table_name} LIMIT %s", (limit,))# 数据操作def insert_data(self, table_name: str, data: Dict) -> int:"""插入数据"""columns = ', '.join(data.keys())placeholders = ', '.join(['%s'] * len(data))values = tuple(data.values())query = f"INSERT INTO {table_name} ({columns}) VALUES ({placeholders})"return self.execute_command(query, values)def update_data(self, table_name: str, data: Dict, condition: str, condition_params: tuple) -> int:"""更新数据"""set_clause = ', '.join([f"{key} = %s" for key in data.keys()])values = tuple(data.values()) + condition_paramsquery = f"UPDATE {table_name} SET {set_clause} WHERE {condition}"return self.execute_command(query, values)def delete_data(self, table_name: str, condition: str, params: tuple) -> int:"""删除数据"""query = f"DELETE FROM {table_name} WHERE {condition}"return self.execute_command(query, params)def search_data(self, table_name: str, search_column: str, search_value: str) -> List[Dict]:"""搜索数据"""query = f"SELECT * FROM {table_name} WHERE {search_column} LIKE %s"return self.execute_query(query, (f"%{search_value}%",))# 使用示例
crud = CRUDOperations(db_manager)

3. 高级查询和事务管理

class AdvancedOperations:def __init__(self, db_manager: DatabaseManager):self.db = db_managerdef transactional_operation(self, operations: List[tuple]) -> bool:"""执行事务操作"""try:with self.db.get_connection() as conn:with conn.cursor() as cursor:# 开始事务conn.start_transaction()for query, params in operations:cursor.execute(query, params)# 提交事务conn.commit()return Trueexcept Error as e:print(f"事务操作失败: {e}")return Falsedef batch_insert(self, table_name: str, data_list: List[Dict]) -> int:"""批量插入数据"""if not data_list:return 0columns = ', '.join(data_list[0].keys())placeholders = ', '.join(['%s'] * len(data_list[0]))values = [tuple(data.values()) for data in data_list]query = f"INSERT INTO {table_name} ({columns}) VALUES ({placeholders})"try:with self.db.get_cursor() as cursor:cursor.executemany(query, values)return cursor.rowcountexcept Error as e:print(f"批量插入失败: {e}")return 0def get_database_info(self) -> Dict:"""获取数据库详细信息"""info = {}# 获取数据库大小size_query = """SELECT table_schema, ROUND(SUM(data_length + index_length) / 1024 / 1024, 2) as size_mbFROM information_schema.tables WHERE table_schema = %sGROUP BY table_schema"""info['size'] = self.db.execute_query(size_query, (db_config['database'],))# 获取表统计信息tables_query = """SELECT table_name, table_rows, avg_row_length, data_length, index_lengthFROM information_schema.tables WHERE table_schema = %s"""info['tables'] = self.db.execute_query(tables_query, (db_config['database'],))return info

4. 安全性和错误处理

class SecureDatabaseOperations:def __init__(self, db_manager: DatabaseManager):self.db = db_managerdef safe_query(self, query: str, params: tuple = None) -> List[Dict]:"""安全的查询执行，防止SQL注入"""try:# 验证查询类型（只允许SELECT）if not query.strip().upper().startswith('SELECT'):raise ValueError("只允许执行SELECT查询")with self.db.get_cursor(dictionary=True) as cursor:cursor.execute(query, params or ())return cursor.fetchall()except (Error, ValueError) as e:print(f"安全查询失败: {e}")return []def parameterized_query(self, table_name: str, filters: Dict) -> List[Dict]:"""参数化查询构建"""if not filters:return self.db.execute_query(f"SELECT * FROM {table_name}")where_clauses = []params = []for column, value in filters.items():if isinstance(value, (list, tuple)):placeholders = ', '.join(['%s'] * len(value))where_clauses.append(f"{column} IN ({placeholders})")params.extend(value)else:where_clauses.append(f"{column} = %s")params.append(value)where_clause = " AND ".join(where_clauses)query = f"SELECT * FROM {table_name} WHERE {where_clause}"return self.db.execute_query(query, tuple(params))def validate_table_name(self, table_name: str) -> bool:"""验证表名安全性"""import re# 只允许字母、数字和下划线return bool(re.match(r'^[a-zA-Z0-9_]+$', table_name))

5. 完整的应用示例

def find_secret_key():"""查找密钥的完整解决方案"""try:# 初始化数据库管理器db_manager = DatabaseManager(db_config)crud = CRUDOperations(db_manager)# 获取所有表tables = crud.get_all_tables()print(f"找到 {len(tables)} 个表: {tables}")# 搜索包含key字段的表key_tables = []for table in tables:structure = crud.get_table_structure(table)for column in structure:if 'key' in column['Field'].lower():key_tables.append((table, column['Field']))break# 从包含key字段的表中查找密钥for table, column in key_tables:print(f"检查表 {table} 的列 {column}")data = crud.execute_query(f"SELECT `{column}` FROM {table} LIMIT 10")for row in data:value = row[column]if value and isinstance(value, str) and len(value) > 10:print(f"找到可能密钥: {value}")return value# 如果没有找到，搜索所有表的所有数据print("未在key列中找到密钥，开始全面搜索...")for table in tables:print(f"全面搜索表: {table}")data = crud.get_table_data(table, limit=50)for row in data:for key, value in row.items():if (value and isinstance(value, str) and (len(value) > 15 or 'key' in str(value).lower())):print(f"在表 {table} 的列 {key} 中找到可能密钥: {value}")return valuereturn Noneexcept Exception as e:print(f"查找密钥时发生错误: {e}")return Nonedef main():"""主函数"""key = find_secret_key()if key:# 保存到文件with open('key.txt', 'w', encoding='utf-8') as f:f.write(str(key))print(f"密钥已保存到 key.txt: {key}")else:print("未找到密钥")# 创建测试文件with open('key.txt', 'w', encoding='utf-8') as f:f.write("default_test_key_not_found")print("已创建测试文件")if __name__ == "__main__":main()

6. 调试和监控工具

class DatabaseMonitor:def __init__(self, db_manager: DatabaseManager):self.db = db_managerdef monitor_performance(self):"""监控数据库性能"""queries = [("连接数", "SHOW STATUS LIKE 'Threads_connected'"),("查询缓存", "SHOW STATUS LIKE 'Qcache%'"),("InnoDB状态", "SHOW ENGINE INNODB STATUS"),("慢查询", "SHOW VARIABLES LIKE 'slow_query_log'")]for name, query in queries:try:result = self.db.execute_query(query)print(f"{name}: {result}")except Error as e:print(f"获取 {name} 信息失败: {e}")def explain_query(self, query: str, params: tuple = None):"""分析查询性能"""explain_query = f"EXPLAIN {query}"result = self.db.execute_query(explain_query, params)print("查询分析结果:")for row in result:print(row)

7. 配置和最佳实践

# config.py - 配置文件
import os
from dotenv import load_dotenvload_dotenv()  # 加载环境变量class Config:# 数据库配置DB_HOST = os.getenv('DB_HOST', 'localhost')DB_PORT = int(os.getenv('DB_PORT', 3306))DB_NAME = os.getenv('DB_NAME', 'lanqiao')DB_USER = os.getenv('DB_USER', 'root')DB_PASSWORD = os.getenv('DB_PASSWORD', '')# 连接池配置POOL_SIZE = int(os.getenv('POOL_SIZE', 5))POOL_RECYCLE = int(os.getenv('POOL_RECYCLE', 3600))@classmethoddef get_db_config(cls):return {'host': cls.DB_HOST,'port': cls.DB_PORT,'database': cls.DB_NAME,'user': cls.DB_USER,'password': cls.DB_PASSWORD,'autocommit': True,'charset': 'utf8mb4','pool_size': cls.POOL_SIZE,'pool_recycle': cls.POOL_RECYCLE}# .env 文件示例
# DB_HOST=localhost
# DB_PORT=3306
# DB_NAME=lanqiao
# DB_USER=root
# DB_PASSWORD=
# POOL_SIZE=5
# POOL_RECYCLE=3600
  1. 连接管理：连接池、上下文管理器
  2. CRUD操作：完整的增删改查实现
  3. 事务处理：保证数据一致性
  4. 安全性：防止SQL注入、参数化查询
  5. 错误处理：完善的异常处理机制
  6. 性能监控：查询分析、性能监控
  7. 配置管理：环境变量配置
http://www.dtcms.com/a/390543.html

相关文章：

  • 大模型上下文工程实践- 上下文管理策略
  • 资产测绘工具-Nmap
  • 智能体环境配置测试
  • 如何将非结构化文档智能解析高质量数据，并按照阅读顺序还原版面？
  • 第八章 惊喜11 认知觉醒
  • 嵌入式系统学习Day37（ARM）
  • 02-Media-10-video_decoder.py 对H.264或H.265格式视频解码并在液晶屏或外接HDMI显示屏上进行显示的示例程序
  • Go面试题及详细答案120题（61-80）
  • 第二部分：VTK核心类详解（第35章：vtkDataSetAttributes数据集属性类）
  • 智能文献分析系统：让AI成为学术研究助手
  • MATLAB基于AHP-熵权法-TOPSIS的学习能力评价研究
  • Ubuntu 部署 PostgreSQL 数据库（附shell脚本一键部署↓）
  • 《数据驱动下的双样本推断：均值与比例的硬核技术实践与方法论思考》
  • Git设置单个仓库用户名和邮箱的方法
  • MongoDB Integer
  • 深度学习第二章 线性代数简介
  • HTB precious
  • 【前后端与数据库交互】从零构建 Python + Vue + MongoDB 网站
  • 对比django，flask，opencv三大
  • 【6/20】MongoDB 入门：连接数据库，实现数据存储与查询
  • 【笔记】Docker使用
  • k8s自定义CNI插件实现指南
  • 使用Docker部署Kubernetes（K8s）详解
  • 【Docker】网络
  • 磁共振成像原理（理论）8：射频回波 (RF Echoes)-三脉冲回波（1）
  • 华为云 ELB：智慧负载均衡，让您的应用永葆流畅体验
  • 【实时Linux实战系列】PM QoS 与 C/P-State 管理：功耗与时延的平衡
  • github修改repo名称
  • 使用 C# 操作 Excel 工作表：添加、删除、复制、移动、重命名
  • Python 高效实现 Excel 转 PDF： 不依赖Office