马云是做网站的,短视频seo营销系统,自己网站做电子签章有效么,四川省江油市建设局网站在 Python 中操作 MySQL 数据库主要通过 pymysql 或 mysql-connector-python 库实现。以下是完整的技术指南#xff0c;包含连接管理、CRUD 操作和最佳实践#xff1a; 一、环境准备
1. 安装驱动库
pip install pymysql # 推荐#xff08;纯Python实现#xff0…在 Python 中操作 MySQL 数据库主要通过 pymysql 或 mysql-connector-python 库实现。以下是完整的技术指南包含连接管理、CRUD 操作和最佳实践 一、环境准备
1. 安装驱动库
pip install pymysql # 推荐纯Python实现
# 或
pip install mysql-connector-python # Oracle官方驱动2. 数据库准备
CREATE DATABASE testdb;
USE testdb;CREATE TABLE users (id INT AUTO_INCREMENT PRIMARY KEY,username VARCHAR(50) UNIQUE NOT NULL,email VARCHAR(100) NOT NULL,created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);二、连接管理
1. 基础连接
import pymysql# 建立连接
conn pymysql.connect(hostlocalhost,userroot,passwordyour_password,databasetestdb,charsetutf8mb4,cursorclasspymysql.cursors.DictCursor # 返回字典格式结果
)2. 上下文管理器推荐
with pymysql.connect(hostlocalhost, userroot, passwordyour_pwd, databasetestdb) as conn:with conn.cursor() as cursor:# 执行SQL操作pass# 连接在此处自动提交/回滚取决于autocommit设置三、CRUD 操作
1. 查询数据
def get_user(username):try:with conn.cursor() as cursor:sql SELECT * FROM users WHERE username %scursor.execute(sql, (username,))result cursor.fetchone() # 获取单条记录return resultexcept pymysql.MySQLError as e:print(f数据库错误: {e})return Noneuser get_user(john_doe)
print(user) # 输出: {id: 1, username: john_doe, ...}2. 插入数据
def create_user(username, email):try:with conn.cursor() as cursor:sql INSERT INTO users (username, email) VALUES (%s, %s)cursor.execute(sql, (username, email))conn.commit() # 显式提交事务return cursor.lastrowid # 返回自增IDexcept pymysql.IntegrityError:print(用户名已存在)conn.rollback()return Nonenew_id create_user(jane_smith, janeexample.com)3. 更新数据
def update_email(user_id, new_email):try:with conn.cursor() as cursor:sql UPDATE users SET email %s WHERE id %saffected_rows cursor.execute(sql, (new_email, user_id))conn.commit()return affected_rows 0except pymysql.MySQLError:conn.rollback()return False4. 批量操作
def batch_insert(users):try:with conn.cursor() as cursor:sql INSERT INTO users (username, email) VALUES (%s, %s)cursor.executemany(sql, users) # 批量执行conn.commit()return cursor.rowcountexcept pymysql.MySQLError:conn.rollback()return 0batch_insert([(user1, u1test.com),(user2, u2test.com)
])四、高级技巧
1. 连接池管理
from dbutils.pooled_db import PooledDBpool PooledDB(creatorpymysql,maxconnections10,hostlocalhost,userroot,passwordyour_pwd,databasetestdb
)# 使用连接池获取连接
conn pool.connection()2. 存储过程调用
with conn.cursor() as cursor:cursor.callproc(get_user_stats, (1,)) # 调用存储过程result cursor.fetchall()3. 事务控制
try:with conn.cursor() as cursor:# 执行多个操作cursor.execute(UPDATE account SET balance balance - 100 WHERE id 1)cursor.execute(UPDATE account SET balance balance 100 WHERE id 2)conn.commit() # 显式提交事务
except:conn.rollback() # 回滚所有操作五、安全实践
1. 防止SQL注入
永远使用参数化查询不要拼接字符串# 正确方式
cursor.execute(SELECT * FROM users WHERE id %s, (user_id,))# 危险方式禁用
cursor.execute(fSELECT * FROM users WHERE id {user_id})2. 敏感信息处理
使用环境变量存储密码import os
password os.getenv(DB_PASSWORD)3. 连接超时设置
conn pymysql.connect(connect_timeout5, # 连接超时5秒read_timeout10 # 读取超时10秒
)六、性能优化
1. 查询优化 添加索引 ALTER TABLE users ADD INDEX idx_email (email);使用EXPLAIN分析查询 with conn.cursor() as cursor:cursor.execute(EXPLAIN SELECT * FROM users WHERE username %s, (john,))print(cursor.fetchall())2. 结果集处理 分页查询 sql SELECT * FROM users LIMIT %s OFFSET %s
cursor.execute(sql, (page_size, (page-1)*page_size))流式读取大数据量 with conn.cursor(pymysql.cursors.SSCursor) as cursor: # 使用服务器端游标cursor.execute(SELECT * FROM large_table)for row in cursor:process(row)七、完整示例
import pymysql
from contextlib import contextmanagercontextmanager
def database_connection():conn pymysql.connect(hostlocalhost,userroot,passwordyour_pwd,databasetestdb,charsetutf8mb4)try:yield connfinally:conn.close()def main():with database_connection() as conn:with conn.cursor() as cursor:# 创建用户cursor.execute(INSERT INTO users (username, email)VALUES (%s, %s), (new_user, userexample.com))user_id cursor.lastrowid# 查询用户cursor.execute(SELECT * FROM users WHERE id %s, (user_id,))print(cursor.fetchone())# 更新记录cursor.execute(UPDATE users SET email %s WHERE id %s, (new_emailexample.com, user_id))conn.commit()if __name__ __main__:main()八、故障排查
1. 常见错误码
1045: 访问被拒绝检查用户名/密码2003: 无法连接检查主机/端口1062: 唯一键冲突1146: 表不存在
2. 日志记录
import logging
logging.basicConfig(levellogging.DEBUG,format%(asctime)s - %(levelname)s - %(message)s
)# 在连接参数中添加
conn pymysql.connect(..., cursorclasspymysql.cursors.SSDictCursor,client_flagpymysql.client.CLIENT.MULTI_STATEMENTS)通过遵循这些实践可以构建安全、高效的数据库交互应用。对于复杂场景建议结合ORM框架如SQLAlchemy进行抽象层开发。
