python用fastapi快速写一个增删改查的接口
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Dict
app = FastAPI()
# Mock database
db = {}
# Model for the data
class Item(BaseModel):
name: str
description: str
# Create operation
@app.post("/items/")
def create_item(item: Item):
if item.name in db:
raise HTTPException(status_code=400, detail="Item already exists")
db[item.name] = item.description
return {"message": "Item created successfully"}
# Read operation
@app.get("/items/{name}")
def read_item(name: str):
if name not in db:
raise HTTPException(status_code=404, detail="Item not found")
return {"name": name, "description": db[name]}
# Update operation
@app.put("/items/{name}")
def update_item(name: str, item: Item):
if name not in db:
raise HTTPException(status_code=404, detail="Item not found")
db[name] = item.description
return {"message": "Item updated successfully"}
# Delete operation
@app.delete("/items/{name}")
def delete_item(name: str):
if name not in db:
raise HTTPException(status_code=404, detail="Item not found")
del db[name]
return {"message": "Item deleted successfully"}
这段代码设置了一个FastAPI应用程序，其中包含用于创建、读取、更新和删除物品的端点。数据以简单的内存数据库形式存储在字典（db）中。您可以使用诸如curl、Postman或任何其他HTTP客户端之类的工具来测试这些端点。
方法补充
除了上文的方法，小编还为大家整理了其他FastAPI实现高效的增删改查操作的方法，希望对大家有所帮助
环境搭建
在开始之前，请确保你的环境中安装了Python 3.6+和pip。使用以下命令安装FastAPI和Uvicorn，Uvicorn是一个轻量级的ASGI服务器。
pip install fastapi uvicorn
快速启动
创建一个main.py文件，并写入以下代码来启动一个简单的FastAPI应用：
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
async def read_root():
return {"Hello": "World"}
运行服务器：
uvicorn main:app --reload
打开浏览器访问http://127.0.0.1:8000/，你将看到返回的JSON响应。
定义数据模型
在实现CRUD操作之前，需要定义数据模型。在models.py文件中，定义一个Pydantic模型，用于请求和响应数据的验证。
from pydantic import BaseModel
# 创建一个Item模型
class Item(BaseModel):
id: int
name: str
description: str = None
price: float
tax: float = None
创建CRUD操作
在crud.py文件中，实现CRUD操作的函数。这里简单地使用一个字典来存储数据，实际应用中应该使用数据库。
from models import Item
items = {}
def create_item(item_id: int, item: Item):
items[item_id] = item
return items[item_id]
def read_item(item_id: int):
return items.get(item_id)
def update_item(item_id: int, item: Item):
if item_id in items:
items[item_id] = item
return items[item_id]
return None
def delete_item(item_id: int):
if item_id in items:
del items[item_id]
return True
return False
实现API端点
在main.py中，将使用CRUD操作函数来实现API端点。
from fastapi import FastAPI, HTTPException
from models import Item
from crud import create_item, read_item, update_item, delete_item
app = FastAPI()
# 创建Item
@app.post("/items/{item_id}")
async def create(item_id: int, item: Item):
return create_item(item_id, item)
# 读取Item
@app.get("/items/{item_id}")
async def read(item_id: int):
item = read_item(item_id)
if item is None:
raise HTTPException(status_code=404, detail="Item not found")
return item
# 更新Item
@app.put("/items/{item_id}")
async def update(item_id: int, item: Item):
updated_item = update_item(item_id, item)
if updated_item is None:
raise HTTPException(status_code=404, detail="Item not found")
return updated_item
# 删除Item
@app.delete("/items/{item_id}")
async def delete(item_id: int):
if not delete_item(item_id):
raise HTTPException(status_code=404, detail="Item not found")
return {"detail": "Item deleted"}
测试API
我们可以使用FastAPI提供的API文档来测试我们的接口。重新运行服务器：
uvicorn main:app --reload
转到http://127.0.0.1:8000/docs，你将看到Swagger UI，一个自动生成的交互式API文档，你可以在这里测试你的API端点。
进阶：使用数据库
为了让CRUD操作更加实际，我们将使用SQLite数据库来存储数据。首先，我们需要安装databases和sqlalchemy。
pip install databases sqlalchemy
接下来，在database.py文件中设置数据库连接和表：
import databases
import sqlalchemy
from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String, Float
DATABASE_URL = "sqlite:///./test.db"
database = databases.Database(DATABASE_URL)
metadata = MetaData()
items = Table(
"items",
metadata,
Column("id", Integer, primary_key=True),
Column("name", String(50)),
Column("description", String(50)),
Column("price", Float),
Column("tax", Float, default=None)
)
engine = create_engine(DATABASE_URL)
metadata.create_all(engine)
在crud.py中，我们将函数更新为使用数据库：
from models import Item
from database import database, items
async def create_item(item_id: int, item: Item):
query = items.insert().values(id=item_id, **item.dict())
last_record_id = await database.execute(query)
return {**item.dict(), "id": last_record_id}
async def read_item(item_id: int):
query = items.select().where(items.c.id == item_id)
return await database.fetch_one(query)
async def update_item(item_id: int, item: Item):
query = items.update().where(items.c.id == item_id).values(**item.dict())
await database.execute(query)
return await read_item(item_id)
async def delete_item(item_id: int):
query = items.delete().where(items.c.id == item_id)
return await database.execute(query) > 0
最后，需要在main.py中启动和关闭数据库连接：
from fastapi import FastAPI, HTTPException
from models import Item
from crud import create_item, read_item, update_item, delete_item
from database import database
app = FastAPI()
@app.on_event("startup")
async def startup():
await database.connect()
@app.on_event("shutdown")
async def shutdown():
await database.disconnect()
# 其余的API端点保持不变
到此这篇关于Python使用fastapi快速编写一个增删改查的接口的文章就介绍到这了,更多相关Python fastapi增删改查内容请搜索IT俱乐部以前的文章或继续浏览下面的相关文章希望大家以后多多支持IT俱乐部！
