FastAPI 和数据库:将 Tortoise ORM 与 FastAPI 结合使用
connygpt 2024-11-08 10:33 46 浏览
FastAPI 是一个现代、快速（高性能）的 Web 框架，用于基于标准 Python 类型提示使用 Python 3.7+ 构建 API。FastAPI 的优点之一是它可以通过 ORM（对象关系映射器）与各种数据库轻松集成。Tortoise ORM 是一个轻量级且直观的 ORM，支持异步操作，与 FastAPI 完美匹配。
在本博客中，我们将介绍如何使用 FastAPI 设置 Tortoise ORM，包括配置、创建模型以及执行 CRUD（创建、读取、更新、删除）操作。
1. 设置项目
首先，让我们创建一个新的 FastAPI 项目。我们将从安装必要的依赖项开始。
pip install fastapi[all] tortoise-orm接下来，为您的项目创建一个新目录并添加以下文件：
main.py：FastAPI 应用程序的主要入口点。
models.py：在此处定义数据库模型。
schemas.py：定义请求和响应模型的 Pydantic 模式。
config.py：Tortoise ORM 的配置设置。
2. 配置 Tortoise ORM
创建一个 config.py 文件来配置 Tortoise ORM。
TORTOISE_ORM = {
"connections": {
"default": "sqlite://db.sqlite3" # Using SQLite for simplicity
},
"apps": {
"models": {
"models": ["models"],
"default_connection": "default",
}
},
}3. 定义模型
创建一个 models.py 文件来定义数据库模型。
from tortoise import fields
from tortoise.models import Model
class User(Model):
id = fields.IntField(pk=True)
username = fields.CharField(max_length=50, unique=True)
email = fields.CharField(max_length=255, unique=True)
hashed_password = fields.CharField(max_length=255)
def __str__(self):
return self.username4. 创建 Pydantic 模式
创建一个 schemas.py 文件来定义 Pydantic 模式。
from pydantic import BaseModel, EmailStr
class UserCreate(BaseModel):
username: str
email: EmailStr
password: str
class UserOut(BaseModel):
id: int
username: str
email: EmailStr
class Config:
orm_mode = True5. 使用 Tortoise ORM 设置 FastAPI
修改 main.py 文件以设置 FastAPI 并集成 Tortoise ORM。
from fastapi import FastAPI, HTTPException
from tortoise.contrib.fastapi import register_tortoise, HTTPNotFoundError
from models import User
from schemas import UserCreate, UserOut
app = FastAPI()
register_tortoise(
app,
db_url='sqlite://db.sqlite3',
modules={'models': ['models']},
generate_schemas=True,
add_exception_handlers=True,
)
@app.post('/users/', response_model=UserOut)
async def create_user(user: UserCreate):
user_obj = await User.create(
username=user.username,
email=user.email,
hashed_password=user.password # In a real application, hash the password
)
return user_obj
@app.get('/users/{user_id}', response_model=UserOut, responses={404: {"model": HTTPNotFoundError}})
async def get_user(user_id: int):
user = await User.get_or_none(id=user_id)
if not user:
raise HTTPException(status_code=404, detail="User not found")
return user
@app.put('/users/{user_id}', response_model=UserOut, responses={404: {"model": HTTPNotFoundError}})
async def update_user(user_id: int, user: UserCreate):
user_obj = await User.get_or_none(id=user_id)
if not user_obj:
raise HTTPException(status_code=404, detail="User not found")
user_obj.username = user.username
user_obj.email = user.email
user_obj.hashed_password = user.password # In a real application, hash the password
await user_obj.save()
return user_obj
@app.delete('/users/{user_id}', response_model=dict, responses={404: {"model": HTTPNotFoundError}})
async def delete_user(user_id: int):
user_obj = await User.get_or_none(id=user_id)
if not user_obj:
raise HTTPException(status_code=404, detail="User not found")
await user_obj.delete()
return {"message": "User deleted successfully"}6. 运行应用程序
要运行 FastAPI 应用程序，请使用以下命令：
uvicorn main:app --reload打开浏览器并导航到 http://127.0.0.1:8000/docs 查看自动生成的 API 文档并测试端点。
7. 使用 Tortoise ORM 和 FastAPI 的其他演示
以下是一些其他演示，用于扩展上面演示的基本 CRUD 操作。这些示例包括处理一对多和多对多关系、使用查询过滤器以及实现分页。
7.1 一对多关系
假设我们想要添加一个帖子模型，其中每个帖子都与一个用户相关联。
models.py
from tortoise import fields, models
class User(models.Model):
id = fields.IntField(pk=True)
username = fields.CharField(max_length=50, unique=True)
email = fields.CharField(max_length=255, unique=True)
hashed_password = fields.CharField(max_length=255)
posts = fields.ReverseRelation['Post']
def __str__(self):
return self.username
class Post(models.Model):
id = fields.IntField(pk=True)
title = fields.CharField(max_length=255)
content = fields.TextField()
user = fields.ForeignKeyField('models.User', related_name='posts')
def __str__(self):
return self.titleschemas.py
from pydantic import BaseModel, EmailStr
from typing import List, Optional
class PostCreate(BaseModel):
title: str
content: str
class PostOut(BaseModel):
id: int
title: str
content: str
user_id: int
class Config:
orm_mode = True
class UserCreate(BaseModel):
username: str
email: EmailStr
password: str
class UserOut(BaseModel):
id: int
username: str
email: EmailStr
posts: List[PostOut] = []
class Config:
orm_mode = Truemain.py
from fastapi import FastAPI, HTTPException
from tortoise.contrib.fastapi import register_tortoise, HTTPNotFoundError
from models import User, Post
from schemas import UserCreate, UserOut, PostCreate, PostOut
app = FastAPI()
register_tortoise(
app,
db_url='sqlite://db.sqlite3',
modules={'models': ['models']},
generate_schemas=True,
add_exception_handlers=True,
)
@app.post('/users/', response_model=UserOut)
async def create_user(user: UserCreate):
user_obj = await User.create(
username=user.username,
email=user.email,
hashed_password=user.password # In a real application, hash the password
)
return user_obj
@app.post('/users/{user_id}/posts/', response_model=PostOut)
async def create_post_for_user(user_id: int, post: PostCreate):
user = await User.get_or_none(id=user_id)
if not user:
raise HTTPException(status_code=404, detail="User not found")
post_obj = await Post.create(**post.dict(), user=user)
return post_obj
@app.get('/users/{user_id}/posts/', response_model=List[PostOut])
async def get_posts_for_user(user_id: int):
posts = await Post.filter(user_id=user_id).all()
return posts7.2 多对多关系
让我们添加一个标签模型以及 Post 和标签之间的多对多关系。
models.py
from tortoise import fields, models
class User(models.Model):
id = fields.IntField(pk=True)
username = fields.CharField(max_length=50, unique=True)
email = fields.CharField(max_length=255, unique=True)
hashed_password = fields.CharField(max_length=255)
posts = fields.ReverseRelation['Post']
def __str__(self):
return self.username
class Post(models.Model):
id = fields.IntField(pk=True)
title = fields.CharField(max_length=255)
content = fields.TextField()
user = fields.ForeignKeyField('models.User', related_name='posts')
tags = fields.ManyToManyField('models.Tag', related_name='posts')
def __str__(self):
return self.title
class Tag(models.Model):
id = fields.IntField(pk=True)
name = fields.CharField(max_length=50, unique=True)
posts = fields.ManyToManyRelation[Post]
def __str__(self):
return self.nameschemas.py
from pydantic import BaseModel, EmailStr
from typing import List, Optional
class PostCreate(BaseModel):
title: str
content: str
class PostOut(BaseModel):
id: int
title: str
content: str
user_id: int
tags: List[str] = []
class Config:
orm_mode = True
class TagCreate(BaseModel):
name: str
class TagOut(BaseModel):
id: int
name: str
class Config:
orm_mode = True
class UserCreate(BaseModel):
username: str
email: EmailStr
password: str
class UserOut(BaseModel):
id: int
username: str
email: EmailStr
posts: List[PostOut] = []
class Config:
orm_mode = Truemain.py
from fastapi import FastAPI, HTTPException
from tortoise.contrib.fastapi import register_tortoise, HTTPNotFoundError
from models import User, Post, Tag
from schemas import UserCreate, UserOut, PostCreate, PostOut, TagCreate, TagOut
from typing import List
app = FastAPI()
register_tortoise(
app,
db_url='sqlite://db.sqlite3',
modules={'models': ['models']},
generate_schemas=True,
add_exception_handlers=True,
)
@app.post('/users/', response_model=UserOut)
async def create_user(user: UserCreate):
user_obj = await User.create(
username=user.username,
email=user.email,
hashed_password=user.password # In a real application, hash the password
)
return user_obj
@app.post('/users/{user_id}/posts/', response_model=PostOut)
async def create_post_for_user(user_id: int, post: PostCreate):
user = await User.get_or_none(id=user_id)
if not user:
raise HTTPException(status_code=404, detail="User not found")
post_obj = await Post.create(**post.dict(), user=user)
return post_obj
@app.get('/users/{user_id}/posts/', response_model=List[PostOut])
async def get_posts_for_user(user_id: int):
posts = await Post.filter(user_id=user_id).all()
return posts
@app.post('/tags/', response_model=TagOut)
async def create_tag(tag: TagCreate):
tag_obj = await Tag.create(**tag.dict())
return tag_obj
@app.post('/posts/{post_id}/tags/', response_model=PostOut)
async def add_tag_to_post(post_id: int, tag_ids: List[int]):
post = await Post.get_or_none(id=post_id)
if not post:
raise HTTPException(status_code=404, detail="Post not found")
tags = await Tag.filter(id__in=tag_ids).all()
await post.tags.add(*tags)
return post
@app.get('/tags/', response_model=List[TagOut])
async def get_tags():
tags = await Tag.all()
return tags
7.3 查询过滤器和分页
让我们向我们的 API 添加一些查询过滤器和分页。
main.py
from fastapi import FastAPI, HTTPException, Query
from tortoise.contrib.fastapi import register_tortoise, HTTPNotFoundError
from models import User, Post, Tag
from schemas import UserCreate, UserOut, PostCreate, PostOut, TagCreate, TagOut
from typing import List, Optional
app = FastAPI()
register_tortoise(
app,
db_url='sqlite://db.sqlite3',
modules={'models': ['models']},
generate_schemas=True,
add_exception_handlers=True,
)
@app.post('/users/', response_model=UserOut)
async def create_user(user: UserCreate):
user_obj = await User.create(
username=user.username,
email=user.email,
hashed_password=user.password # In a real application, hash the password
)
return user_obj
@app.post('/users/{user_id}/posts/', response_model=PostOut)
async def create_post_for_user(user_id: int, post: PostCreate):
user = await User.get_or_none(id=user_id)
if not user:
raise HTTPException(status_code=404, detail="User not found")
post_obj = await Post.create(**post.dict(), user=user)
return post_obj
@app.get('/users/{user_id}/posts/', response_model=List[PostOut])
async def get_posts_for_user(user_id: int):
posts = await Post.filter(user_id=user_id).all()
return posts
@app.post('/tags/', response_model=TagOut)
async def create_tag(tag: TagCreate):
tag_obj = await Tag.create(**tag.dict())
return tag_obj
@app.post('/posts/{post_id}/tags/', response_model=PostOut)
async def add_tag_to_post(post_id: int, tag_ids: List[int]):
post = await Post.get_or_none(id=post_id)
if not post:
raise HTTPException(status_code=404, detail="Post not found")
tags = await Tag.filter(id__in=tag_ids).all()
await post.tags.add(*tags)
return post
@app.get('/tags/', response_model=List[TagOut])
async def get_tags():
tags = await Tag.all()
return tags
@app.get('/posts/', response_model=List[PostOut])
async def get_posts(
tag: Optional[str] = None,
limit: int = Query(10, le=100),
offset: int = Query(0)
):
if tag:
posts = await Post.filter(tags__name=tag).offset(offset).limit(limit).all()
else:
posts = await Post.all().offset(offset).limit(limit)
return posts在本博客中，我们演示了如何将 Tortoise ORM 与 FastAPI 集成。我们介绍了如何设置项目、配置 Tortoise ORM、定义模型、创建 Pydantic 模式以及执行 CRUD 操作。Tortoise ORM 提供了一种异步且易于使用的 ORM 解决方案，与 FastAPI 的异步功能完美契合。
您可以随意扩展此基本设置，添加更复杂的模型、关系和业务逻辑，以满足您的应用程序需求。
相关推荐
- 3分钟让你的项目支持AI问答模块，完全开源!
-
hello，大家好，我是徐小夕。之前和大家分享了很多可视化，零代码和前端工程化的最佳实践，今天继续分享一下最近开源的Next-Admin的最新更新。最近对这个项目做了一些优化，并集成了大家比较关注...
- 干货|程序员的副业挂，12个平台分享
-
1、D2adminD2Admin是一个完全开源免费的企业中后台产品前端集成方案，使用最新的前端技术栈，小于60kb的本地首屏js加载，已经做好大部分项目前期准备工作，并且带有大量示例代码，助...
- Github标星超200K，这10个可视化面板你知道几个
-
在Github上有很多开源免费的后台控制面板可以选择，但是哪些才是最好、最受欢迎的可视化控制面板呢？今天就和大家推荐Github上10个好看又流行的可视化面板：1.AdminLTEAdminLTE是...
- 开箱即用的炫酷中后台前端开源框架第二篇
-
#头条创作挑战赛#1、SoybeanAdmin(1)介绍：SoybeanAdmin是一个基于Vue3、Vite3、TypeScript、NaiveUI、Pinia和UnoCSS的清新优...
- 搭建React+AntDeign的开发环境和框架
-
搭建React+AntDeign的开发环境和框架随着前端技术的不断发展，React和AntDesign已经成为越来越多Web应用程序的首选开发框架。React是一个用于构建用户界面的JavaScrip...
- 基于.NET 5实现的开源通用权限管理平台
-
??大家好，我是为广大程序员兄弟操碎了心的小编，每天推荐一个小工具/源码，装满你的收藏夹，每天分享一个小技巧，让你轻松节省开发效率，实现不加班不熬夜不掉头发，是我的目标！??今天小编推荐一款基于.NE...
- StreamPark - 大数据流计算引擎
-
使用Docker完成StreamPark的部署??1.基于h2和docker-compose进行StreamPark部署wgethttps://raw.githubusercontent.com/a...
- 教你使用UmiJS框架开发React
-
1、什么是Umi.js?umi，中文可发音为乌米，是一个可插拔的企业级react应用框架。你可以将它简单地理解为一个专注性能的类next.js前端框架，并通过约定、自动生成和解析代码等方式来辅助...
- 简单在线流程图工具在用例设计中的运用
-
敏捷模式下，测试团队的用例逐渐简化以适应快速的发版节奏，大家很早就开始运用思维导图工具比如xmind来编写测试方法、测试点。如今不少已经不少利用开源的思维导图组件（如百度脑图...）来构建测试测试...
- 【开源分享】神奇的大数据实时平台框架,让Flink&Spark开发更简单
-
这是一个神奇的框架，让Flink|Spark开发更简单，一站式大数据实时平台！他就是StreamX！什么是StreamX大数据技术如今发展的如火如荼，已经呈现百花齐放欣欣向荣的景象，实时处理流域...
- 聊聊规则引擎的调研及实现全过程
-
摘要本期主要以规则引擎业务实现为例,陈述在陌生业务前如何进行业务深入、调研、技术选型、设计及实现全过程分析,如果你对规则引擎不感冒、也可以从中了解一些抽象实现过程。诉求从硬件采集到的数据提供的形式多种...
- 【开源推荐】Diboot 2.0.5 发布，自动化开发助理
-
一、前言Diboot2.0.5版本已于近日发布，在此次发布中，我们新增了file-starter组件，完善了iam-starter组件，对core核心进行了相关优化，让devtools也支持对IAM...
- 微软推出Copilot Actions，使用人工智能自动执行重复性任务
-
IT之家11月19日消息，微软在今天举办的Ignite大会上宣布了一系列新功能，旨在进一步提升Microsoft365Copilot的智能化水平。其中最引人注目的是Copilot...
- Electron 使用Selenium和WebDriver
-
本节我们来学习如何在Electron下使用Selenium和WebDriver。SeleniumSelenium是ThoughtWorks提供的一个强大的基于浏览器的开源自动化测试工具...
- Quick 'n Easy Web Builder 11.1.0设计和构建功能齐全的网页的工具
-
一个实用而有效的应用程序，能够让您轻松构建、创建和设计个人的HTML网站。Quick'nEasyWebBuilder是一款全面且轻巧的软件，为用户提供了一种简单的方式来创建、编辑...
- 一周热门
- 最近发表
- 标签列表
-
- kubectlsetimage (56)
- mysqlinsertoverwrite (53)
- addcolumn (54)
- helmpackage (54)
- varchar最长多少 (61)
- 类型断言 (53)
- protoc安装 (56)
- jdk20安装教程 (60)
- rpm2cpio (52)
- 控制台打印 (63)
- 401unauthorized (51)
- vuexstore (68)
- druiddatasource (60)
- 企业微信开发文档 (51)
- rendertexture (51)
- speedphp (52)
- gitcommit-am (68)
- bashecho (64)
- str_to_date函数 (58)
- yum下载包及依赖到本地 (72)
- jstree中文api文档 (59)
- mvnw文件 (58)
- rancher安装 (63)
- nginx开机自启 (53)
- .netcore教程 (53)