Flet开发网站，数据库模块SQLAlchemy使用“十要领”

SQLAlchemy 是 Python 中最流行的 ORM (对象关系映射) 工具之一，它提供了高效、灵活的数据库访问方式。以下是 SQLAlchemy 的全面使用指南。

要领一：安装与基本配置

首先安装 SQLAlchemy：

```bash

pip install sqlalchemy

```

### 1.1 创建引擎 (Engine)

```python

from sqlalchemy import create_engine

# SQLite 示例

engine = create_engine('sqlite:///example.db', echo=True)

# PostgreSQL 示例

# engine = create_engine('postgresql://user:password@localhost/mydatabase')

# MySQL 示例

# engine = create_engine('mysql+pymysql://user:password@localhost/mydatabase')

```

`echo=True` 会输出 SQL 语句，便于调试。

要领二：声明式基类与模型定义

SQLAlchemy 有两种主要使用方式：

- **声明式** (Declarative) - 更现代、更常用

- **经典式** (Classical) - 更接近 SQL

推荐使用声明式：

```python

from sqlalchemy.ext.declarative import declarative_base

from sqlalchemy import Column, Integer, String, DateTime, ForeignKey

from sqlalchemy.orm import relationship

from datetime import datetime

Base = declarative_base()

class User(Base):

__tablename__ = 'users'

id = Column(Integer, primary_key=True)

username = Column(String(50), unique=True, nullable=False)

email = Column(String(100), unique=True)

created_at = Column(DateTime, default=datetime.utcnow)

# 一对多关系

posts = relationship("Post", back_populates="author")

def __repr__(self):

return f"<User(username='{self.username}', email='{self.email}')>"

class Post(Base):

__tablename__ = 'posts'

id = Column(Integer, primary_key=True)

title = Column(String(100), nullable=False)

content = Column(String(1000))

user_id = Column(Integer, ForeignKey('users.id'))

# 多对一关系

author = relationship("User", back_populates="posts")

def __repr__(self):

return f"<Post(title='{self.title}', author='{self.author.username}')>"

```

要领三：创建表

```python

Base.metadata.create_all(engine)

```

这会创建所有定义的表。如果表已存在，不会重复创建。

要领四：会话管理

SQLAlchemy 使用 Session 来管理数据库操作：

```python

from sqlalchemy.orm import sessionmaker

Session = sessionmaker(bind=engine)

session = Session()

```

要领五：CRUD 操作

### 5.1 创建 (Create)

```python

# 创建单个对象

new_user = User(username='john_doe', email='john@example.com')

session.add(new_user)

session.commit()

# 批量创建

session.add_all([

User(username='alice', email='alice@example.com'),

User(username='bob', email='bob@example.com')

])

session.commit()

```

### 5.2 查询 (Read)

#### 基本查询

```python

# 获取所有用户

users = session.query(User).all()

# 获取第一个用户

first_user = session.query(User).first()

# 按主键获取

user = session.query(User).get(1) # 获取id=1的用户

```

#### 过滤查询

```python

from sqlalchemy import or_

# WHERE 条件

john = session.query(User).filter(User.username == 'john_doe').first()

# 多个条件

users = session.query(User).filter(

User.email.like('%@example.com'),

or_(User.username == 'john_doe', User.username == 'alice')

).all()

# 排序

users = session.query(User).order_by(User.created_at.desc()).all()

# 限制

users = session.query(User).limit(10).offset(5).all() # 跳过5条，取10条

```

### 5.3 更新 (Update)

```python

user = session.query(User).filter_by(username='john_doe').first()

if user:

user.email = 'new_email@example.com'

session.commit()

```

### 5.4 删除 (Delete)

```python

user = session.query(User).filter_by(username='john_doe').first()

if user:

session.delete(user)

session.commit()

```

要领六：关系操作

### 一对多关系

```python

# 创建带关系的对象

user = User(username='author1', email='author1@example.com')

post1 = Post(title='First Post', content='Hello World!', author=user)

post2 = Post(title='Second Post', content='SQLAlchemy is awesome!', author=user)

session.add_all([user, post1, post2])

session.commit()

# 查询用户的所有文章

user = session.query(User).filter_by(username='author1').first()

print(user.posts) # 输出 [<Post(title='First Post')>, <Post(title='Second Post')>]

# 查询文章的作者

post = session.query(Post).filter_by(title='First Post').first()

print(post.author.username) # 输出 'author1'

```

### 多对多关系

```python

from sqlalchemy import Table

# 关联表

post_tags = Table('post_tags', Base.metadata,

Column('post_id', Integer, ForeignKey('posts.id')),

Column('tag_id', Integer, ForeignKey('tags.id'))

)

class Tag(Base):

__tablename__ = 'tags'

id = Column(Integer, primary_key=True)

name = Column(String(50), unique=True)

posts = relationship("Post", secondary=post_tags, back_populates="tags")

# 修改Post类添加关系

class Post(Base):

# ... 其他字段 ...

tags = relationship("Tag", secondary=post_tags, back_populates="posts")

# 使用示例

post = session.query(Post).get(1)

tag1 = Tag(name='python')

tag2 = Tag(name='webdev')

post.tags.extend([tag1, tag2])

session.commit()

```

要领七：高级查询

### 连接查询

```python

# 内连接

results = session.query(User, Post).join(Post).filter(User.username == 'author1').all()

# 左外连接

from sqlalchemy import outerjoin

results = session.query(User, Post).outerjoin(Post).all()

```

### 聚合函数

```python

from sqlalchemy import func

# 计数

count = session.query(func.count(User.id)).scalar()

# 分组统计

from sqlalchemy import desc

result = session.query(

User.username,

func.count(Post.id).label('post_count')

).outerjoin(Post).group_by(User.username).order_by(desc('post_count')).all()

```

要领八：事务管理

```python

try:

user1 = User(username='user1', email='user1@example.com')

session.add(user1)

post = Post(title='Transaction Test', content='Testing transaction')

session.add(post)

# 这里会抛出异常，因为user_id是必须的

session.commit()

except:

session.rollback()

print("Transaction failed, rolled back")

finally:

session.close()

```

要领九：异步支持 (SQLAlchemy 1.4+)

```python

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession

from sqlalchemy.orm import sessionmaker

async def main():

# 异步引擎

engine = create_async_engine(

"postgresql+asyncpg://user:password@localhost/dbname"

)

AsyncSessionLocal = sessionmaker(

expire_on_commit=False,

class_=AsyncSession,

bind=engine

)

async with AsyncSessionLocal() as session:

# 异步查询

result = await session.execute(

select(User).where(User.username == 'john_doe')

)

user = result.scalars().first()

print(user)

# 运行异步代码

import asyncio

asyncio.run(main())

```

要领十：最佳实践

1. **会话生命周期管理**：确保会话及时关闭

2. **批量操作**：大量数据时使用 `bulk_insert_mappings`/`bulk_update_mappings`

3. **避免 N+1 查询**：使用 `joinedload` 或 `selectinload` 预加载关联数据

4. **使用 Alembic 进行数据库迁移**：管理数据库模式变更

5. **连接池配置**：生产环境中合理配置连接池

```python

from sqlalchemy.orm import joinedload

# 避免N+1查询的示例

users_with_posts = session.query(User).options(joinedload(User.posts)).all()

for user in users_with_posts:

print(user.username)

for post in user.posts: # 不会产生额外查询

print(f" - {post.title}")

```

（汇报完毕，收藏+点赞）