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

SQLAlchemy 库

news 2025/9/25 20:19:50

SQLAlchemy 是 Python 中最著名、功能最强大的 ORM（对象关系映射）库和 SQL 工具包。它提供了完整的企业级持久化模式，旨在高效、高性能地访问数据库。

1. 核心作用概述

主要功能

# SQLAlchemy 的三大核心组件
core_components = {"ORM (对象关系映射)": "将Python类映射到数据库表","Core (SQL表达式语言)": "提供SQL构造器，可编写数据库无关的SQL","Engine (数据库引擎)": "处理数据库连接和方言适配"
}

2. ORM：对象关系映射

传统SQL vs ORM方式

# 传统方式 - 直接写SQL
import sqlite3
conn = sqlite3.connect('test.db')
cursor = conn.cursor()
cursor.execute("INSERT INTO users (name, age) VALUES (?, ?)", ("张三", 25))
conn.commit()# ORM方式 - 操作Python对象
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmakerBase = declarative_base()class User(Base):__tablename__ = 'users'id = Column(Integer, primary_key=True)name = Column(String(50))age = Column(Integer)# 操作对象而非SQL
new_user = User(name="张三", age=25)
session.add(new_user)
session.commit()

3. 核心组件详解

a) 声明式基类

from sqlalchemy import create_engine, Column, Integer, String, DateTime
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from datetime import datetimeBase = declarative_base()class Product(Base):__tablename__ = 'products'id = Column(Integer, primary_key=True, autoincrement=True)name = Column(String(100), nullable=False)price = Column(Integer, default=0)created_at = Column(DateTime, default=datetime.now)def __repr__(self):return f"<Product(id={self.id}, name='{self.name}', price={self.price})>"

b) 数据库引擎和会话

# 创建数据库引擎（支持多种数据库）
engine = create_engine('sqlite:///mydatabase.db', echo=True)  # echo=True显示SQL# 创建所有表
Base.metadata.create_all(engine)# 创建会话工厂
Session = sessionmaker(bind=engine)
session = Session()

4. 基本CRUD操作

创建（Create）

# 添加单个对象
new_product = Product(name="笔记本电脑", price=5999)
session.add(new_product)# 添加多个对象
products = [Product(name="鼠标", price=99),Product(name="键盘", price=199),Product(name="显示器", price=1299)
]
session.add_all(products)
session.commit()  # 提交事务

查询（Read）

# 查询所有
all_products = session.query(Product).all()# 条件查询
expensive_products = session.query(Product).filter(Product.price > 500).all()# 获取单个对象
mouse = session.query(Product).filter(Product.name == "鼠标").first()# 复杂查询
from sqlalchemy import and_, or_
result = session.query(Product).filter(and_(Product.price >= 100, Product.price <= 1000)
).order_by(Product.price.desc()).all()

更新（Update）

# 更新对象
product = session.query(Product).filter_by(name="鼠标").first()
if product:product.price = 79  # 直接修改属性session.commit()    # 提交更改

删除（Delete）

# 删除对象
product = session.query(Product).filter_by(name="键盘").first()
if product:session.delete(product)session.commit()

5. 高级功能

a) 关系映射

from sqlalchemy import ForeignKey
from sqlalchemy.orm import relationshipclass Order(Base):__tablename__ = 'orders'id = Column(Integer, primary_key=True)user_id = Column(Integer, ForeignKey('users.id'))# 定义关系user = relationship("User", back_populates="orders")class User(Base):__tablename__ = 'users'id = Column(Integer, primary_key=True)name = Column(String(50))# 反向关系orders = relationship("Order", back_populates="user")# 使用关系
user = session.query(User).first()
for order in user.orders:  # 直接访问关联的订单print(order.id)

b) 数据库迁移（Alembic）

# 使用Alembic进行数据库版本控制
# 生成迁移脚本
# alembic revision --autogenerate -m "添加新字段"# 应用迁移
# alembic upgrade head

c) 连接池管理

# 配置连接池
engine = create_engine('postgresql://user:pass@localhost/dbname',pool_size=10,           # 连接池大小max_overflow=20,        # 最大溢出连接数pool_timeout=30,        # 超时时间pool_recycle=3600       # 连接回收时间
)

6. 多数据库支持

SQLAlchemy 支持几乎所有主流数据库：

# 不同数据库的连接字符串示例
databases = {"SQLite": "sqlite:///database.db","PostgreSQL": "postgresql://user:password@localhost/dbname","MySQL": "mysql+pymysql://user:password@localhost/dbname","Oracle": "oracle+cx_oracle://user:password@localhost:1521/dbname","SQL Server": "mssql+pyodbc://user:password@dsn_name"
}# 代码无需修改，只需更改连接字符串
engine = create_engine(databases["PostgreSQL"])

7. 在实际项目中的应用

Web应用中的使用

from flask import Flask
from flask_sqlalchemy import SQLAlchemyapp = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///app.db'
db = SQLAlchemy(app)class User(db.Model):id = db.Column(db.Integer, primary_key=True)username = db.Column(db.String(80), unique=True, nullable=False)email = db.Column(db.String(120), unique=True, nullable=False)def __repr__(self):return f'<User {self.username}>'# 在视图函数中使用
@app.route('/users')
def list_users():users = User.query.all()return {'users': [user.username for user in users]}

数据分析和批处理

# 批量插入数据
def bulk_insert_products(products_data):products = [Product(**data) for data in products_data]session.bulk_save_objects(products)session.commit()# 分页查询
def get_products_page(page=1, per_page=20):return session.query(Product).order_by(Product.id).offset((page-1)*per_page).limit(per_page).all()

8. 优势总结

优势	说明
数据库无关性	一套代码支持多种数据库
类型安全	Python类型与数据库类型映射
自动SQL生成	避免SQL注入，提高安全性
连接池管理	高效的数据库连接管理
事务支持	完整的事务管理机制
丰富的查询API	链式调用，表达力强
生态系统完善	有Alembic等配套工具