fastapi是一个很优秀的框架，但是缺少一个合适的orm，官方代码里面使用的是SQLALchemy
# SQLALchemy
pip install sqlalchemy
# 连接数据库
from sqlalchemy import create_engine
# 创建数据库引擎
engine = create_engine('sqlite:///my_database.db', echo=True)
# 在内存中创建数据库
engine = create_engine('sqlite:///:memory:', echo=True)
# echo=True 参数用于在终端输出 SQL 查询语句
# 定义表结构
from sqlalchemy import Table, Column, Integer, String, MetaData
metadata = MetaData()
# 创建一个数据表
users = Table('users', metadata,
Column('id', Integer, primary_key=True),
Column('name', String),
Column('age', Integer)
)
metadata.create_all(engine)
# SQLModel
SQLModel (opens new window) 是一个用于 Python 的数据库交互库，它结合了 SQLAlchemy 的强大数据库操作能力和 Pydantic 的数据验证与序列化功能，旨在提供简洁、类型安全且高效的数据库操作体验。
# 快速上手示例
- 定义模型
from sqlmodel import Field, SQLModel, create_engine, Session, select
# 这个 Hero 类既是 SQLAlchemy 的数据库模型（table=True），也是 Pydantic 的数据验证模型
class Hero(SQLModel, table=True):
id: int | None = Field(default=None, primary_key=True)
name: str
secret_name: str
age: int | None = None
- 写入数据库
engine = create_engine("sqlite:///database.db")
SQLModel.metadata.create_all(engine)
hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
hero_2 = Hero(name="Spider-Boy", secret_name="Pedro Parqueador")
hero_3 = Hero(name="Rusty-Man", secret_name="Tommy Sharp", age=48)
with Session(engine) as session:
session.add(hero_1)
session.add(hero_2)
session.add(hero_3)
session.commit()
- 查询数据库
with Session(engine) as session:
statement = select(Hero).where(Hero.name == "Spider-Boy")
hero = session.exec(statement).first()
print(hero)
# 输出: Hero(id=2, name='Spider-Boy', secret_name='Pedro Parqueador', age=None)
- 与 FastAPI 集成
from fastapi import FastAPI
from sqlmodel import select
app = FastAPI()
@app.get("/heroes/{name}")
def get_hero(name: str):
with Session(engine) as session:
statement = select(Hero).where(Hero.name == name)
hero = session.exec(statement).first()
return hero
# Tortoise
Tortoise ORM (opens new window) 是受 Django 启发的易于使用的异步 ORM
pip install tortoise-orm
# Tortoise ORM 支持的数据库 settings.py
- PostgreSQL >= 9.4（使用asyncpg）
from tortoise import Tortoise
# 配置 PostgreSQL 数据库
config = {
'db_url': 'postgres://user:password@localhost/dbname',
'modules': {
'tortoise.backends.postgres': {
'sslmode': 'disable' # 可选的，根据你的 PostgreSQL 配置调整
}
},
'generate_schemas': True # 自动创建表结构
}
# 初始化 Tortoise ORM
Tortoise.init(**config)
# 创建表
Tortoise.generate_schemas()
- SQLite（使用aiosqlite）
from tortoise import Tortoise
# 配置 SQLite 数据库
config = {
# 使用内存中的 SQLite 数据库，也可以指定文件路径
'db_url': 'sqlite://:memory:',
'modules': {
'tortoise.backends.sqlite': {
'_fk': True, # 开启外键支持
}
},
'generate_schemas': True # 自动创建表结构
}
# 初始化 Tortoise ORM
Tortoise.init(**config)
# 创建表
Tortoise.generate_schemas()
- MySQL/MariaDB（使用aiomysql或使用asyncmy）
from tortoise import Tortoise
# 配置 MySQL 数据库
TORTOISE_ORM = {
"connections": {
"default": {
# "engine": "tortoise.backends.asyncpg", # 数据库引擎 PostgresQL
"engine": "tortoise.backends.mysql", # 数据库引擎 Mysql or Mariadb
"credentials": {
"host": "127.0.0.1", # 地址
"port": "3306", # 端口
"user": "root", # 用户名
"password": "root", # 密码
"database": "fastapi", # 数据库名称（需要提前创建数据库）
"minsize": 1, # 最少连接
"maxsize": 5, # 最大连接
"charset": "utf8mb4", # 编码
"echo": True # 是否反馈SQL语句
}
}
},
"apps": {
"models": {
# Aerich 需要在 TORTOISE_ORM 配置里注册其模型，否则无法找到默认连接
"models": ["models", "aerich.models"],
"default_connection": "default"
}
},
"use_tz": False,
"timezone": "Asia/Shanghai"
}
# 以下的可以在项目启动中配置
# 初始化 Tortoise ORM
# Tortoise.init(**TORTOISE_ORM)
# 创建表
# Tortoise.generate_schemas()
- Oracle
from tortoise import Tortoise
config = {
'db_url': 'oracle://username:password@host:port/service_name',
'modules': {
'tortoise.backends.oracle': {
# 这里可以配置其他 Oracle 特定的设置，如使用钱包等
}
},
'generate_schemas': True # 如果需要自动创建表结构，设置为 True
}
Tortoise.init(**config)
# 创建模型文件 models.py
from tortoise import fields, Model
class Classes(Model):
name = fields.CharField(max_length=255, description="班级名称")
class Teacher(Model):
id = fields.IntField(pk=True)
name = fields.CharField(max_length=255, description="老师名称")
tno = fields.IntField(description="帐号")
pwd = fields.CharField(max_length=255, description="密码")
class Course(Model):
id = fields.IntField(pk=True)
name = fields.CharField(max_length=255, description="课程名称")
# 多对一
teacher = fields.ForeignKeyField("models.Teacher", related_name="courses")
class Student(Model):
id = fields.IntField(pk=True)
name = fields.CharField(max_length=255, description="学生名称")
sno = fields.IntField(description="帐号")
pwd = fields.CharField(max_length=255, description="密码")
# 一对多
classes = fields.ForeignKeyField("models.Classes", related_name="students")
# 多对多
courses = fields.ManyToManyField("models.Course", related_name="students")
# aerich 迁移工具
aerich是一种ORM迁移工具，需要结合tortoise异步orm框架使用
pip install aerich
- 初始化配置，只需要使用一次
aerich init -t settings.TORTOISE_ORM # TORTOISE_ORM 配置的位置
# 初始化完会在当前目录生成一个文件：pyproject.toml和一个文件夹：migrations
# pyproject.toml：保存配置文件路径，低版本可能是aerich.ini
# migrations：存放迁移文件
- 初始化数据库，一般情况下只用一次
aerich init-db
# 此时数据库中就有相应的表格
# 如果TORTOISE_ORM配置文件中的models改了名，则执行这条命令时需要增加--app参数，来指定你修改的名字
- 更新模型并进行迁移
# 修改model类，重新生成迁移文件,比如添加一个字段
aerich migrate [--name] (标记修改操作) # aerich migrate
# 注意，此时sql并没有执行，数据库中表中没有更新
- 重新执行迁移，写入数据库
aerich upgrade
- 回到上一个版本
aerich downgrade
- 查看历史迁移记录
aerich history
# 项目启动
import uvicorn
from fastapi import FastAPI
from tortoise.contrib.fastapi import register_tortoise
import orm
from settings import TORTOISE_ORM
app = FastAPI()
app.include_router(orm.api_student, prefix='/orm', tags=['orm'])
# 该方法会在fastapi启动时触发，内部通过传递进去的app对象，监听服务启动和终止事件
# 当检测到启动事件时，会初始化Tortoise对象，如果generate_schemas为True则还会进行数据库迁移
# 当检测到终止事件时，会关闭连接
register_tortoise(
app,
config=TORTOISE_ORM,
# generate_schemas=True, # 如果数据库为空，则自动生成对应表单，生产环境不要开
# add_exception_handlers=True, # 生产环境不要开，会泄露调试信息
)
if __name__ == '__main__':
uvicorn.run('quick:app', host='127.0.0.1', port=8020, reload=True, workers=1)
# 增删改查
from typing import List
from fastapi import APIRouter
from pydantic import BaseModel
from models import Student, Course
api_student = APIRouter()
class StudentModel(BaseModel):
name: str
sno: int
pwd: str
phone: str
classes_id: int
courses: List[int]
@api_student.get("/students")
async def getAllStudent():
students = await Student.all().values("name", "classes_id")
for student in students:
print(student)
return students
@api_student.get("/student")
async def getStudent():
student = await Student.filter(name__icontains='三').values("name", "classes_id")
# student = await Student.get(name__icontains='一') # 不存在会报错
student = await Student.get_or_none(name__icontains='一') # 不存在返回None
return student
@api_student.post("/student")
async def addStudent(stu: StudentModel):
# 方式一
# student = Student(
# name=stu.name,
# sno=stu.sno,
# pwd=stu.pwd,
# phone=stu.phone,
# classes_id=stu.classes_id
# )
# await student.save()
# 方式二
student = await Student.create(
name=stu.name,
sno=stu.sno,
pwd=stu.pwd,
phone=stu.phone,
classes_id=stu.classes_id
)
# 添加多对多关系
courses = await Course.filter(id__in=stu.courses).all()
await student.courses.add(*courses)
return student # 返回的是新增的对象
@api_student.put("/student/{id}")
async def updateStudent(id: int, stu: StudentModel):
# 方式一
# student = await Student.get(id=id)
# student.name = stu.name
# student.sno = stu.sno
# student.pwd = stu.pwd
# student.phone = stu.phone
# student.classes_id = stu.classes_id
# await student.save()
# 方式二
student = await Student.filter(id=id).update(
name=stu.name,
sno=stu.sno,
pwd=stu.pwd,
phone=stu.phone,
classes_id=stu.classes_id
)
# 更新多对多关系
courses = await Course.filter(id__in=stu.courses).all()
await student.courses.clear()
await student.courses.add(*courses)
return student
@api_student.delete("/student/{id}")
async def deleteStudent(id: int):
# 方式一
# student = await Student.get(id=id)
# await student.delete()
# return student # 返回的是删除的对象
# 方式二
student = await Student.filter(id=id).delete()
return student # 返回的是删除的行数
# 这两种方式都可以删除多对多关系
# 过滤条件
- 比较运算符
# 获取id等于1的数据
students = await Student.filter(id=1).all()
# 获取id不等于1的数据
students = await Student.filter(id__not=1).all()
# 获取id大于1的数据
students = await Student.filter(id__gt=1).all()
# 获取id大于等于1的数据
students = await Student.filter(id__gte=1).all()
# 获取id小于5的数据
students = await Student.filter(id__lt=1).all()
# 获取id小于等于5的数据
students = await Student.filter(id__lte=1).all()
- 成员运算符
# 获取姓名 在 指定列表中的数据
names = ['赵德柱', '李铁柱']
students = await Student.filter(name__in=names).all()
for student in students:
print(student.id, student.name)
# 获取姓名 不在 指定列表中的数据
names = ['赵德柱', '李铁柱']
students = await Student.filter(name__nin=names).all()
for student in students:
print(student.id, student.name)
- 模糊查询
# Tortoise ORM 不直接支持SQL中的LIKE模糊查询，
# 但可以使用`icontains`、`istartswith`、`iendswith`等操作符进行模糊查询。
# 学号包含200
student = await Student.filter(sno__icontains='200')
# 学号是200开头
student = await Student.filter(sno__istartswith='200')
# 学号是200结尾
student = await Student.filter(sno__iendswith='200')
# exclude() 方法用于排除满足条件的数据，返回不满足条件的数据集
# 获取名字不是'赵德柱'的所有数据
students = await Student.exclude(name='赵德柱').all()
- 查询在指定范围之间
students = await Student.filter(sno__range=[2001, 2003]).all()
for student in students:
print(student.id, student.name)
- 是否为空
# 查询学生姓名为空的数据
students = await Student.filter(name__isnull=True).all()
- 正则表达式匹配
# 查询名字匹配正则表达式的数据
pattern = r'^赵.*' # 以 赵 开头的名字
students = await Student.filter(name__regex=pattern).all()
# 查询名字不区分大小写匹配正则表达式的用户
pattern = r'^a.*' # 以 a（iregex 不区分大小写）开头的名字
students = await Student.filter(name__iregex=pattern).all()
- 分页查询
# 按id升序获取所有数据
students = await Student.all().order_by("id")
# 按id降序获取所有数据
students = await Student.all().order_by("id")
# 获取前5个用户
first_five_sutdents = await Student.all().limit(5)
# 跳过前5个用户，再获取5个用户
next_five_students = await Student.all().offset(5).limit(5)
← 基本介绍