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

Fastapi 进阶三：数据库的应用

news 2025/11/8 8:38:08

FastAPI 进阶三：数据库的深度应用

一、数据库基础与SQL概述

1.1 关系型数据库与非关系型数据库

在现代Web开发中，数据库系统主要分为两类：关系型数据库（SQL）和非关系型数据库（NoSQL）。关系型数据库通过预定义的表结构和SQL语言实现数据的规范化管理，典型代表包括MySQL、PostgreSQL和SQL Server。其核心特性包括：

ACID事务：原子性、一致性、隔离性和持久性
结构化查询语言（SQL）：标准化的数据库操作语言
主外键约束：保障数据完整性

非关系型数据库（如MongoDB、Redis）则采用灵活的数据模型，适用于大规模分布式系统，其优势在于：

水平扩展能力：通过分片实现横向扩展
动态模式：无需预定义表结构
高性能读写：适用于高并发场景

1.2 Python操作SQLite数据库

SQLite作为轻量级嵌入式数据库，常用于开发和测试环境。其Python操作流程如下：

import sqlite3# 创建数据库连接
conn = sqlite3.connect('example.db')
cursor = conn.cursor()# 创建表
cursor.execute('''CREATE TABLE IF NOT EXISTS users (id INTEGER PRIMARY KEY AUTOINCREMENT,name TEXT NOT NULL,email TEXT UNIQUE)
''')# 插入数据
cursor.execute("INSERT INTO users (name, email) VALUES (?, ?)", ('Alice', 'alice@example.com'))
conn.commit()# 查询数据
cursor.execute("SELECT * FROM users")
results = cursor.fetchall()
for row in results:print(row)# 关闭连接
conn.close()

关键操作包括：

connect()建立数据库连接
cursor()创建游标对象
execute()执行SQL语句
commit()提交事务
fetchall()获取查询结果

1.3 ORM操作数据库

对象关系映射（ORM）技术通过将数据库表映射为Python类，实现代码与数据库的解耦。其核心优势包括：

类型安全：编译时检查字段类型
链式查询：构建复杂的查询条件
自动迁移：根据模型生成数据库表

二、SQLAlchemy库深度应用

2.1 SQLAlchemy同步使用方式

SQLAlchemy作为Python最强大的ORM框架，提供声明式模型定义方式：

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))email = Column(String(100), unique=True)# 创建引擎
engine = create_engine('sqlite:///example.db', echo=True)# 创建表
Base.metadata.create_all(engine)# 创建会话
Session = sessionmaker(bind=engine)
session = Session()# 添加数据
new_user = User(name='Bob', email='bob@example.com')
session.add(new_user)
session.commit()# 查询数据
users = session.query(User).filter(User.name == 'Bob').all()
for user in users:print(user.id, user.name)

2.2 SQLAlchemy异步使用方式

异步模式通过asyncmy驱动实现非阻塞I/O操作：

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmakerengine = create_async_engine('mysql+asyncmy://user:password@localhost/db_name',echo=True
)AsyncSessionLocal