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

import mysql.connector 
 
# 創(chuàng)建連接 
conn = mysql.connector.connect(
 host="localhost",
 user="your_username",
 password="your_password",
 database="your_database" 
) 
 
# 創(chuàng)建游標 
cursor = conn.cursor() 
 
# 執(zhí)行SQL查詢 
cursor.execute("SELECT * FROM your_table") 
 
# 獲取結果 
results = cursor.fetchall() 
for row in results: 
print(row) 
 
# 關閉連接 
cursor.close() 
conn.close() 

import pymysql 
 
# 創(chuàng)建連接 
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:
         # 執(zhí)行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()
 
 # 執(zhí)行多個SQL操作
 cursor.execute(sql1, val1)
 cursor.execute(sql2, val2)
 
 # 提交事務 conn.commit() 
except Exception as e:
 # 發(fā)生錯誤時回滾
 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 
 
# 創(chuàng)建引擎 
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) 
 
# 創(chuàng)建表 
Base.metadata.create_all(engine) 
 
# 創(chuàng)建會話 
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 )