Python远程控制MySQL的完整指南

pip install mysql-connector-python 
# 或 
pip install pymysql 

import mysql.connector 
 
# 创建连接 
conn = mysql.connector.connect(
 host="localhost",
 user="your_username",
 password="your_password",
 database="your_database" 
) 
 
# 创建游标 
cursor = conn.cursor() 
 
# 执行SQL查询 
cursor.execute("SELECT * FROM your_table") 
 
# 获取结果 
results = cursor.fetchall() 
for row in results: 
print(row) 
 
# 关闭连接 
cursor.close() 
conn.close() 

import pymysql 
 
# 创建连接 
conn = pymysql.connect(
 host='localhost',
 user='your_username',
 password='your_password',
 db='your_database',
 charset='utf8mb4',
 cursorclass=pymysql.cursors.DictCursor 
) 
 
# 使用上下文管理器自动管理连接 
with conn:
     with conn.cursor() as cursor:
         # 执行SQL查询
         sql = "SELECT * FROM your_table"
         cursor.execute(sql)
 
         # 获取结果
         results = cursor.fetchall()
         for row in results:
             print(row) 

cursor.execute(""" 
CREATE TABLE IF NOT EXISTS users (
     id INT AUTO_INCREMENT PRIMARY KEY,
     name VARCHAR(255) NOT NULL,
     email VARCHAR(255) NOT NULL UNIQUE,
     created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
 ) 
""") 

# 单条插入 
sql = "INSERT INTO users (name, email) VALUES (%s, %s)" 
val = ("John Doe", "john@example.com") 
cursor.execute(sql, val) 
 
# 多条插入 
sql = "INSERT INTO users (name, email) VALUES (%s, %s)" 
val = [
     ("Jane Smith", "jane@example.com"),
     ("Bob Johnson", "bob@example.com") 
] 
cursor.executemany(sql, val) 
 
# 提交事务 
conn.commit() 

# 查询所有记录 
cursor.execute("SELECT * FROM users") 
rows = cursor.fetchall() 
 
# 查询单条记录 
cursor.execute("SELECT * FROM users WHERE id = %s", (1,)) 
row = cursor.fetchone() 
 
# 带条件的查询 
cursor.execute("SELECT name, email FROM users WHERE name LIKE %s", ("%John%",)) 
rows = cursor.fetchall() 

sql = "UPDATE users SET name = %s WHERE id = %s" 
val = ("John Smith", 1) 
cursor.execute(sql, val) 
conn.commit() 

sql = "DELETE FROM users WHERE id = %s" 
val = (1,) 
cursor.execute(sql, val) 
conn.commit() 

try:
 # 开始事务
 conn.start_transaction()
 
 # 执行多个SQL操作
 cursor.execute(sql1, val1)
 cursor.execute(sql2, val2)
 
 # 提交事务 conn.commit() 
except Exception as e:
 # 发生错误时回滚
 conn.rollback()
 print(f"Transaction failed: {e}") 

from sqlalchemy import create_engine, Column, Integer, String 
from sqlalchemy.ext.declarative import declarative_base 
from sqlalchemy.orm import sessionmaker 
 
# 创建引擎 
engine = create_engine('mysql+pymysql://user:password@localhost/dbname') 
 
# 声明基类 
Base = declarative_base() 
 
# 定义模型 
class User(Base):
 __tablename__ = 'users' 
 
 id = Column(Integer, primary_key=True)
 name = Column(String(255))
 email = Column(String(255), unique=True) 
 
# 创建表 
Base.metadata.create_all(engine) 
 
# 创建会话 
Session = sessionmaker(bind=engine) 
session = Session() 
 
# 添加新用户 
new_user = User(name='Alice', email='alice@example.com') 
session.add(new_user) 
session.commit() 
 
# 查询用户 
users = session.query(User).filter_by(name='Alice').all() 
for user in users:
 print(user.name, user.email) 

try:
 conn = mysql.connector.connect(**config)
 cursor = conn.cursor()
 cursor.execute("SELECT * FROM non_existent_table") 
except mysql.connector.Error as err:
 print(f"Error: {err}") 
finally: if 'conn' in locals() and conn.is_connected():
 cursor.close()
 conn.close()

# SQLAlchemy（支持多种数据库）
pip install sqlalchemy
 
# DBUtils（通用连接池）
pip install dbutils
 
# Psycopg2（PostgreSQL专用）
pip install psycopg2-binary

from sqlalchemy import create_engine
 
# 带连接池的配置（连接池大小5-10）
engine = create_engine(
    "mysql+pymysql://user:pass@host/db",
    pool_size=5,
    max_overflow=5,
    pool_recycle=3600
)

from dbutils.pooled_db import PooledDB
import pymysql
 
pool = PooledDB(
    creator=pymysql,
    maxconnections=10,
    host='localhost',
    user='root',
    database='test'
)
conn = pool.connection()  # 获取连接

# PostgreSQL设置隔离级别 
import psycopg2 
conn = psycopg2.connect(dsn) 
conn.set_isolation_level(
 psycopg2.extensions.ISOLATION_LEVEL_REPEATABLE_READ )