SQLAlchemy 教程：从入门到精通

SQLAlchemy 教程：从入门到精通

SQLAlchemy 是一个功能强大的 Python SQL 工具包和对象关系映射器 (ORM)，它为应用程序开发人员提供了 SQL 的全部功能和灵活性。本教程将引导你从 SQLAlchemy 的基础知识逐步深入到高级用法，最终达到精通的水平。

第一部分：入门基础

1. 安装 SQLAlchemy:

使用 pip 安装 SQLAlchemy:

bash
pip install sqlalchemy

1. 连接数据库:

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')
```

1. 创建表:

使用 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)
```

1. 插入数据:

使用 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'))

1. 查询数据:

使用 select 语句查询数据：

```python
from sqlalchemy import select

with engine.connect() as conn:
result = conn.execute(select(users))
for row in result:
print(row)
```

第二部分：核心概念 - ORM

1. 声明式映射:

使用 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})>"

```

1. 会话管理:

使用 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)
```

1. 关系映射:

定义表之间的关系，例如一对多、多对多等：

```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()
```

第三部分：高级用法

1. 复杂查询:

使用 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()
```

1. 事务管理:

使用 session.begin() 和 session.rollback() 管理事务：

python
try:
session.begin()
# 执行数据库操作
session.commit()
except Exception:
session.rollback()
raise

1. 事件监听:

使用 event.listen 监听数据库事件：

```python
from sqlalchemy import event

@event.listens_for(User, 'before_insert')
def before_insert_listener(mapper, connection, target):
print("Before insert:", target)
```

1. Core 和 ORM 的结合:

在 ORM 中使用 Core 的功能：

```python
from sqlalchemy.sql import text

result = session.execute(text("SELECT * FROM users WHERE name=:name"), {"name": "jack"})
```

1. 异步操作 (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())

```

第四部分：最佳实践

1. 使用连接池:

配置连接池以提高性能。

1. 合理使用 ORM 和 Core:

根据实际情况选择使用 ORM 或 Core。

1. 编写单元测试:

确保代码的正确性。

总结:

本教程涵盖了 SQLAlchemy 的核心概念和常用用法，从基本的数据库连接和表操作到高级的 ORM 映射和异步操作。 通过学习和实践这些知识，你可以充分利用 SQLAlchemy 的强大功能，构建高效、可靠的数据库应用程序。 记住，持续学习和探索 SQLAlchemy 的更多高级特性，例如 Alembic 数据库迁移工具，可以进一步提升你的技能。 希望本教程能够帮助你成为 SQLAlchemy 专家!