SQLAlchemy 教程:从入门到精通
SQLAlchemy 教程:从入门到精通
SQLAlchemy 是一个功能强大的 Python SQL 工具包和对象关系映射器 (ORM),它为应用程序开发人员提供了 SQL 的全部功能和灵活性。本教程将引导你从 SQLAlchemy 的基础知识逐步深入到高级用法,最终达到精通的水平。
第一部分:入门基础
- 安装 SQLAlchemy:
使用 pip 安装 SQLAlchemy:
bash
pip install sqlalchemy
- 连接数据库:
SQLAlchemy 支持多种数据库,例如 PostgreSQL, MySQL, SQLite 等。连接数据库的核心是 create_engine
函数:
```python
from sqlalchemy import create_engine
# 连接 SQLite 内存数据库
engine = create_engine('sqlite:///:memory:')
# 连接 PostgreSQL 数据库
# engine = create_engine('postgresql://user:password@host:port/database')
# 连接 MySQL 数据库
# engine = create_engine('mysql+pymysql://user:password@host:port/database')
```
- 创建表:
使用 SQLAlchemy 的 Table
对象和 MetaData
对象来定义数据库表:
```python
from sqlalchemy import Table, Column, Integer, String, MetaData
metadata = MetaData()
users = Table('users', metadata,
Column('id', Integer, primary_key=True),
Column('name', String),
Column('fullname', String),
)
# 创建表
metadata.create_all(engine)
```
- 插入数据:
使用 Connection
对象和 execute
方法插入数据:
python
with engine.connect() as conn:
conn.execute(users.insert().values(name='jack', fullname='Jack Jones'))
conn.execute(users.insert().values(name='wendy', fullname='Wendy Williams'))
- 查询数据:
使用 select
语句查询数据:
```python
from sqlalchemy import select
with engine.connect() as conn:
result = conn.execute(select(users))
for row in result:
print(row)
```
第二部分:核心概念 - ORM
- 声明式映射:
使用 declarative_base
创建基类,并定义映射类:
```python
from sqlalchemy.orm import declarative_base, Session
Base = declarative_base()
class User(Base):
tablename = 'users'
id = Column(Integer, primary_key=True)
name = Column(String)
fullname = Column(String)
def __repr__(self):
return f"<User(name={self.name}, fullname={self.fullname})>"
```
- 会话管理:
使用 Session
对象管理数据库连接和事务:
```python
session = Session(engine)
# 添加新用户
new_user = User(name='ed', fullname='Ed Jones')
session.add(new_user)
session.commit()
# 查询用户
users = session.query(User).all()
for user in users:
print(user)
```
- 关系映射:
定义表之间的关系,例如一对多、多对多等:
```python
from sqlalchemy.orm import relationship
class Address(Base):
tablename = 'addresses'
id = Column(Integer, primary_key=True)
email_address = Column(String, nullable=False)
user_id = Column(Integer, ForeignKey('users.id'))
user = relationship("User", back_populates="addresses")
User.addresses = relationship("Address", order_by=Address.id, back_populates="user")
# 创建关联数据
jack = session.query(User).filter_by(name='jack').one()
jack.addresses = [Address(email_address='[email protected]')]
session.commit()
```
第三部分:高级用法
- 复杂查询:
使用 filter
, order_by
, join
等方法进行复杂查询:
```python
users = session.query(User).filter(User.name.like('%j%')).order_by(User.id).all()
users_with_addresses = session.query(User).join(Address).all()
```
- 事务管理:
使用 session.begin()
和 session.rollback()
管理事务:
python
try:
session.begin()
# 执行数据库操作
session.commit()
except Exception:
session.rollback()
raise
- 事件监听:
使用 event.listen
监听数据库事件:
```python
from sqlalchemy import event
@event.listens_for(User, 'before_insert')
def before_insert_listener(mapper, connection, target):
print("Before insert:", target)
```
- Core 和 ORM 的结合:
在 ORM 中使用 Core 的功能:
```python
from sqlalchemy.sql import text
result = session.execute(text("SELECT * FROM users WHERE name=:name"), {"name": "jack"})
```
- 异步操作 (Async SQLAlchemy):
对于现代的异步框架,可以使用 Async SQLAlchemy 进行异步数据库操作:
```python
import asyncio
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
async_engine = create_async_engine("postgresql+asyncpg://user:password@host/database")
async def run():
async with AsyncSession(async_engine) as session:
result = await session.execute(select(User))
users = result.scalars().all()
print(users)
asyncio.run(run())
```
第四部分:最佳实践
- 使用连接池:
配置连接池以提高性能。
- 合理使用 ORM 和 Core:
根据实际情况选择使用 ORM 或 Core。
- 编写单元测试:
确保代码的正确性。
总结:
本教程涵盖了 SQLAlchemy 的核心概念和常用用法,从基本的数据库连接和表操作到高级的 ORM 映射和异步操作。 通过学习和实践这些知识,你可以充分利用 SQLAlchemy 的强大功能,构建高效、可靠的数据库应用程序。 记住,持续学习和探索 SQLAlchemy 的更多高级特性,例如 Alembic 数据库迁移工具,可以进一步提升你的技能。 希望本教程能够帮助你成为 SQLAlchemy 专家!