翼度科技»论坛 编程开发 python 查看内容

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

3

主题

3

帖子

9

积分

新手上路

Rank: 1

积分
9
python用fastapi快速写一个增删改查的接口
  1. from fastapi import FastAPI, HTTPException
  2. from pydantic import BaseModel
  3. from typing import Dict

  4. app = FastAPI()

  5. # Mock database
  6. db = {}


  7. # Model for the data
  8. class Item(BaseModel):
  9.     name: str
  10.     description: str


  11. # Create operation
  12. @app.post("/items/")
  13. def create_item(item: Item):
  14.     if item.name in db:
  15.         raise HTTPException(status_code=400, detail="Item already exists")
  16.     db[item.name] = item.description
  17.     return {"message": "Item created successfully"}


  18. # Read operation
  19. @app.get("/items/{name}")
  20. def read_item(name: str):
  21.     if name not in db:
  22.         raise HTTPException(status_code=404, detail="Item not found")
  23.     return {"name": name, "description": db[name]}


  24. # Update operation
  25. @app.put("/items/{name}")
  26. def update_item(name: str, item: Item):
  27.     if name not in db:
  28.         raise HTTPException(status_code=404, detail="Item not found")
  29.     db[name] = item.description
  30.     return {"message": "Item updated successfully"}


  31. # Delete operation
  32. @app.delete("/items/{name}")
  33. def delete_item(name: str):
  34.     if name not in db:
  35.         raise HTTPException(status_code=404, detail="Item not found")
  36.     del db[name]
  37.     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. app = FastAPI()

  3. @app.get("/")
  4. async def read_root():
  5.     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. # 创建一个Item模型
  3. class Item(BaseModel):
  4.     id: int
  5.     name: str
  6.     description: str = None
  7.     price: float
  8.     tax: float = None
复制代码
创建CRUD操作
在crud.py文件中,实现CRUD操作的函数。这里简单地使用一个字典来存储数据,实际应用中应该使用数据库。
  1. from models import Item

  2. items = {}

  3. def create_item(item_id: int, item: Item):
  4.     items[item_id] = item
  5.     return items[item_id]

  6. def read_item(item_id: int):
  7.     return items.get(item_id)

  8. def update_item(item_id: int, item: Item):
  9.     if item_id in items:
  10.         items[item_id] = item
  11.         return items[item_id]
  12.     return None

  13. def delete_item(item_id: int):
  14.     if item_id in items:
  15.         del items[item_id]
  16.         return True
  17.     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. app = FastAPI()

  5. # 创建Item
  6. @app.post("/items/{item_id}")
  7. async def create(item_id: int, item: Item):
  8.     return create_item(item_id, item)

  9. # 读取Item
  10. @app.get("/items/{item_id}")
  11. async def read(item_id: int):
  12.     item = read_item(item_id)
  13.     if item is None:
  14.         raise HTTPException(status_code=404, detail="Item not found")
  15.     return item

  16. # 更新Item
  17. @app.put("/items/{item_id}")
  18. async def update(item_id: int, item: Item):
  19.     updated_item = update_item(item_id, item)
  20.     if updated_item is None:
  21.         raise HTTPException(status_code=404, detail="Item not found")
  22.     return updated_item

  23. # 删除Item
  24. @app.delete("/items/{item_id}")
  25. async def delete(item_id: int):
  26.     if not delete_item(item_id):
  27.         raise HTTPException(status_code=404, detail="Item not found")
  28.     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. DATABASE_URL = "sqlite:///./test.db"

  5. database = databases.Database(DATABASE_URL)
  6. metadata = MetaData()

  7. items = Table(
  8.     "items",
  9.     metadata,
  10.     Column("id", Integer, primary_key=True),
  11.     Column("name", String(50)),
  12.     Column("description", String(50)),
  13.     Column("price", Float),
  14.     Column("tax", Float, default=None)
  15. )

  16. engine = create_engine(DATABASE_URL)
  17. metadata.create_all(engine)
复制代码
在crud.py中,我们将函数更新为使用数据库:
  1. from models import Item
  2. from database import database, items

  3. async def create_item(item_id: int, item: Item):
  4.     query = items.insert().values(id=item_id, **item.dict())
  5.     last_record_id = await database.execute(query)
  6.     return {**item.dict(), "id": last_record_id}

  7. async def read_item(item_id: int):
  8.     query = items.select().where(items.c.id == item_id)
  9.     return await database.fetch_one(query)

  10. async def update_item(item_id: int, item: Item):
  11.     query = items.update().where(items.c.id == item_id).values(**item.dict())
  12.     await database.execute(query)
  13.     return await read_item(item_id)

  14. async def delete_item(item_id: int):
  15.     query = items.delete().where(items.c.id == item_id)
  16.     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. app = FastAPI()

  6. @app.on_event("startup")
  7. async def startup():
  8.     await database.connect()

  9. @app.on_event("shutdown")
  10. async def shutdown():
  11.     await database.disconnect()

  12. # 其余的API端点保持不变
复制代码
到此这篇关于Python使用fastapi快速编写一个增删改查的接口的文章就介绍到这了,更多相关Python fastapi增删改查内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家!

来源:https://www.jb51.net/python/319208vrr.htm
免责声明:由于采集信息均来自互联网,如果侵犯了您的权益,请联系我们【E-Mail:cb@itdo.tech】 我们会及时删除侵权内容,谢谢合作!
来自手机

举报 回复 使用道具