Python使用fastapi快速编写一个增删改查的接口

心灵之窗2024

python用fastapi快速写一个增删改查的接口

1. from fastapi import FastAPI, HTTPException
   
2. from pydantic import BaseModel
   
3. from typing import Dict
   
4. 
   
5. app = FastAPI()
   
6. 
   
7. # Mock database
   
8. db = {}
   
9. 
   
10. 
    
11. # Model for the data
    
12. class Item(BaseModel):
    
13.     name: str
    
14.     description: str
    
15. 
    
16. 
    
17. # Create operation
    
18. @app.post("/items/")
    
19. def create_item(item: Item):
    
20.     if item.name in db:
    
21.         raise HTTPException(status_code=400, detail="Item already exists")
    
22.     db[item.name] = item.description
    
23.     return {"message": "Item created successfully"}
    
24. 
    
25. 
    
26. # Read operation
    
27. @app.get("/items/{name}")
    
28. def read_item(name: str):
    
29.     if name not in db:
    
30.         raise HTTPException(status_code=404, detail="Item not found")
    
31.     return {"name": name, "description": db[name]}
    
32. 
    
33. 
    
34. # Update operation
    
35. @app.put("/items/{name}")
    
36. def update_item(name: str, item: Item):
    
37.     if name not in db:
    
38.         raise HTTPException(status_code=404, detail="Item not found")
    
39.     db[name] = item.description
    
40.     return {"message": "Item updated successfully"}
    
41. 
    
42. 
    
43. # Delete operation
    
44. @app.delete("/items/{name}")
    
45. def delete_item(name: str):
    
46.     if name not in db:
    
47.         raise HTTPException(status_code=404, detail="Item not found")
    
48.     del db[name]
    
49.     return {"message": "Item deleted successfully"}

复制代码

这段代码设置了一个FastAPI应用程序，其中包含用于创建、读取、更新和删除物品的端点。数据以简单的内存数据库形式存储在字典（db）中。您可以使用诸如curl、Postman或任何其他HTTP客户端之类的工具来测试这些端点。
方法补充
除了上文的方法，小编还为大家整理了其他FastAPI实现高效的增删改查操作的方法，希望对大家有所帮助
环境搭建
在开始之前，请确保你的环境中安装了Python 3.6+和pip。使用以下命令安装FastAPI和Uvicorn，Uvicorn是一个轻量级的ASGI服务器。

1. pip install fastapi uvicorn

复制代码

快速启动
创建一个main.py文件，并写入以下代码来启动一个简单的FastAPI应用：

1. from fastapi import FastAPI
   
2. 
   
3. app = FastAPI()
   
4. 
   
5. @app.get("/")
   
6. async def read_root():
   
7.     return {"Hello": "World"}

复制代码

运行服务器：

1. uvicorn main:app --reload

复制代码

打开浏览器访问http://127.0.0.1:8000/，你将看到返回的JSON响应。
定义数据模型
在实现CRUD操作之前，需要定义数据模型。在models.py文件中，定义一个Pydantic模型，用于请求和响应数据的验证。

1. from pydantic import BaseModel
   
2. 
   
3. # 创建一个Item模型
   
4. class Item(BaseModel):
   
5.     id: int
   
6.     name: str
   
7.     description: str = None
   
8.     price: float
   
9.     tax: float = None

复制代码

创建CRUD操作
在crud.py文件中，实现CRUD操作的函数。这里简单地使用一个字典来存储数据，实际应用中应该使用数据库。

1. from models import Item
   
2. 
   
3. items = {}
   
4. 
   
5. def create_item(item_id: int, item: Item):
   
6.     items[item_id] = item
   
7.     return items[item_id]
   
8. 
   
9. def read_item(item_id: int):
   
10.     return items.get(item_id)
    
11. 
    
12. def update_item(item_id: int, item: Item):
    
13.     if item_id in items:
    
14.         items[item_id] = item
    
15.         return items[item_id]
    
16.     return None
    
17. 
    
18. def delete_item(item_id: int):
    
19.     if item_id in items:
    
20.         del items[item_id]
    
21.         return True
    
22.     return False

复制代码

实现API端点
在main.py中，将使用CRUD操作函数来实现API端点。

1. from fastapi import FastAPI, HTTPException
   
2. from models import Item
   
3. from crud import create_item, read_item, update_item, delete_item
   
4. 
   
5. app = FastAPI()
   
6. 
   
7. # 创建Item
   
8. @app.post("/items/{item_id}")
   
9. async def create(item_id: int, item: Item):
   
10.     return create_item(item_id, item)
    
11. 
    
12. # 读取Item
    
13. @app.get("/items/{item_id}")
    
14. async def read(item_id: int):
    
15.     item = read_item(item_id)
    
16.     if item is None:
    
17.         raise HTTPException(status_code=404, detail="Item not found")
    
18.     return item
    
19. 
    
20. # 更新Item
    
21. @app.put("/items/{item_id}")
    
22. async def update(item_id: int, item: Item):
    
23.     updated_item = update_item(item_id, item)
    
24.     if updated_item is None:
    
25.         raise HTTPException(status_code=404, detail="Item not found")
    
26.     return updated_item
    
27. 
    
28. # 删除Item
    
29. @app.delete("/items/{item_id}")
    
30. async def delete(item_id: int):
    
31.     if not delete_item(item_id):
    
32.         raise HTTPException(status_code=404, detail="Item not found")
    
33.     return {"detail": "Item deleted"}

复制代码

测试API
我们可以使用FastAPI提供的API文档来测试我们的接口。重新运行服务器：

1. uvicorn main:app --reload

复制代码

转到http://127.0.0.1:8000/docs，你将看到Swagger UI，一个自动生成的交互式API文档，你可以在这里测试你的API端点。
进阶：使用数据库
为了让CRUD操作更加实际，我们将使用SQLite数据库来存储数据。首先，我们需要安装databases和sqlalchemy。

1. pip install databases sqlalchemy

复制代码

接下来，在database.py文件中设置数据库连接和表：

1. import databases
   
2. import sqlalchemy
   
3. from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String, Float
   
4. 
   
5. DATABASE_URL = "sqlite:///./test.db"
   
6. 
   
7. database = databases.Database(DATABASE_URL)
   
8. metadata = MetaData()
   
9. 
   
10. items = Table(
    
11.     "items",
    
12.     metadata,
    
13.     Column("id", Integer, primary_key=True),
    
14.     Column("name", String(50)),
    
15.     Column("description", String(50)),
    
16.     Column("price", Float),
    
17.     Column("tax", Float, default=None)
    
18. )
    
19. 
    
20. engine = create_engine(DATABASE_URL)
    
21. metadata.create_all(engine)

复制代码

在crud.py中，我们将函数更新为使用数据库：

1. from models import Item
   
2. from database import database, items
   
3. 
   
4. async def create_item(item_id: int, item: Item):
   
5.     query = items.insert().values(id=item_id, **item.dict())
   
6.     last_record_id = await database.execute(query)
   
7.     return {**item.dict(), "id": last_record_id}
   
8. 
   
9. async def read_item(item_id: int):
   
10.     query = items.select().where(items.c.id == item_id)
    
11.     return await database.fetch_one(query)
    
12. 
    
13. async def update_item(item_id: int, item: Item):
    
14.     query = items.update().where(items.c.id == item_id).values(**item.dict())
    
15.     await database.execute(query)
    
16.     return await read_item(item_id)
    
17. 
    
18. async def delete_item(item_id: int):
    
19.     query = items.delete().where(items.c.id == item_id)
    
20.     return await database.execute(query) > 0

复制代码

最后，需要在

1. main.py

复制代码

中启动和关闭数据库连接：

1. from fastapi import FastAPI, HTTPException
   
2. from models import Item
   
3. from crud import create_item, read_item, update_item, delete_item
   
4. from database import database
   
5. 
   
6. app = FastAPI()
   
7. 
   
8. @app.on_event("startup")
   
9. async def startup():
   
10.     await database.connect()
    
11. 
    
12. @app.on_event("shutdown")
    
13. async def shutdown():
    
14.     await database.disconnect()
    
15. 
    
16. # 其余的API端点保持不变

复制代码

到此这篇关于Python使用fastapi快速编写一个增删改查的接口的文章就介绍到这了,更多相关Python fastapi增删改查内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！

绝影刀

已测试，非常不错

冒险任东

谢谢你的提醒，我会注意的。

爱编程的牧童

让我们一起努力

十年国战玩家ok

谢谢你的提醒，我会注意的。

糖醋奥利奥

同意！

用户0736329

谢谢你分享这个信息

余生泪角

已测试，非常不错

树上的猫

666666666666

叭叭的土鼠

太棒了！感谢分享这个信息！