掌握FastAPI SQL 数据库操作，轻松玩转数据存储

2023-09-03 19:09:23

FastAPI 与 SQL 数据库：联手打造强大 web 应用

一、强强联手：FastAPI 与 SQL 数据库

FastAPI 是一款性能卓越、灵活多变的 web 应用框架，深受开发者的喜爱。SQL 数据库，如 MySQL、PostgreSQL 等，提供强大的数据存储能力，是数据管理的不二之选。将二者结合，可打造出功能强劲且可靠稳定的 web 应用。

二、ORM 工具：通往数据库操作的桥梁

对象关系映射（ORM）工具是实现 FastAPI 与 SQL 数据库完美结合的关键。ORM 工具将对象映射到数据库表，简化数据交互。FastAPI 支持多种 ORM 工具，其中 SQLAlchemy 和 Tortoise ORM 广受推崇。

三、SQLAlchemy：助力数据库操作的利器

SQLAlchemy 是 FastAPI 中最受欢迎的 ORM 工具之一。它功能强大，支持 MySQL、PostgreSQL、SQLite 等多种数据库。以下是使用 SQLAlchemy 进行数据库操作的步骤：

1. 安装 SQLAlchemy

pip install sqlalchemy

2. 创建 SQLAlchemy 引擎

from sqlalchemy import create_engine

engine = create_engine("mysql+pymysql://username:password@host:port/database_name")

3. 创建数据模型

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    username = Column(String(255), unique=True)
    password = Column(String(255))

4. 创建数据库会话

from sqlalchemy.orm import sessionmaker

Session = sessionmaker(bind=engine)
session = Session()

5. 操作数据库

# 添加用户
new_user = User(username='test_user', password='test_password')
session.add(new_user)

# 提交更改
session.commit()

# 查询用户
user = session.query(User).filter_by(username='test_user').first()

6. 提交更改

session.commit()

7. 关闭数据库会话

session.close()

四、Tortoise ORM：快速搭建数据库应用的秘密武器

如果你更青睐异步编程，Tortoise ORM 是一个理想的选择。它是一款轻量级的 ORM 工具，能够轻松构建异步数据库应用。Tortoise ORM 的特点包括：

异步编程：支持非阻塞 I/O，提升应用性能
简单易用：语法简洁，学习成本低
高性能：采用非阻塞 I/O 和协程，确保应用高性能

五、案例实战：体验 FastAPI 数据库操作的魅力

为了加深对 FastAPI 与 SQL 数据库操作的理解，我们准备了一个案例实战。我们将使用 SQLAlchemy 与 MySQL 数据库进行交互，完成用户注册和登录功能。

1. 创建 FastAPI 应用

from fastapi import FastAPI, Request, HTTPException

app = FastAPI()

2. 定义注册路由

@app.post("/register")
async def register(request: Request):
    data = await request.json()
    username = data["username"]
    password = data["password"]

    # 验证用户是否存在
    user = session.query(User).filter_by(username=username).first()
    if user:
        raise HTTPException(status_code=400, detail="用户已存在")

    # 创建新用户
    new_user = User(username=username, password=password)
    session.add(new_user)
    session.commit()

    return {"detail": "注册成功"}

3. 定义登录路由

@app.post("/login")
async def login(request: Request):
    data = await request.json()
    username = data["username"]
    password = data["password"]

    # 验证用户是否存在
    user = session.query(User).filter_by(username=username).first()
    if not user:
        raise HTTPException(status_code=400, detail="用户不存在")

    # 验证密码
    if user.password != password:
        raise HTTPException(status_code=400, detail="密码错误")

    # 生成 token
    token = jwt.encode({"username": username}, "secret_key")

    return {"detail": "登录成功", "token": token}