译者按:FastAPI越来越火了,基本上和Django, Flask一起站稳了Python Web框架前3的位置。尽管Django已经很优秀了,但是新鲜事物和技术还是要关注下的。本文使用FastAPI重构了Django官网的Polls API,能让你对FastAPI的使用过程有个初步了解。
- 原文链接:https://www.agiliq.com/blog/2020/05/polls-api-using-fastapi/
- 原作:Manjunath Hugar
- 翻译:大江狗
什么是FastAPI?
FastAPI 是一个用于构建 API 的现代、快速(高性能)的 web 框架,使用 Python 3.6 并基于标准的 Python 类型提示。其性能可以与NodeJS和GO比肩。
安装
打开终端,使用pip安装。
代码语言:javascript复制pip install fastapi
你同时需要安装ASGI服务器。
代码语言:javascript复制pip install uvicorn
安装就是这么简单,现在让我们开始利用它编写API终点吧。首先创建 main.py
并添加如下代码:
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
def index():
return {"message": "Welcome to the world of FastAPI!"}
@app.get("/items/{item}")
def read_item(item: str, q: str = None):
return {"item": item, "q": q}
现在启动服务器:
代码语言:javascript复制uvicorn main:app
打开浏览器,访问: http://127.0.0.1:8000/
,你应该能看到如下响应。
{"message":"Welcome to the world of FastAPI!"}
访问 http://127.0.0.1:8000/items/apple?q=delicious
,你应该可以看到如下响应。
{"item":"apple","q":"delicious"}
这太好了,我们已经创建了API有两个终点:
http://127.0.0.1:8000/
不接收任何参数,它只是返回一个JSON响应。http://127.0.0.1:8000/items/{item}"
采取str类型参数item和可选查询参数q。
FastAPI 的另一个优点是它提供了一个交互式 API 文档,只需访问http://127.0.0.1:8000/docs
或http://127.0.0.1:8000/redoc
正式开始
现在,让我们继续重建我们的民意调查(Polls)教程API。我们上面创建的端点是静态的,它们不与数据库交互。在下一节中,您将了解如何使用SQLAlchemy进行 ORM 和Pydantic创建模型/计划,使我们的 API 充满活力。
我们将创建以下API端点:
- 创建投票问题
- 列出所有投票问题
- 获取问题详细信息
- 编辑投票问题
- 删除投票问题
- 为特定的投票问题创建选择
- 更新特定问题的投票
我们的项目结构如下所示,一共就5个文件。
代码语言:javascript复制└───pollsapi
│---crud.py
│---database.py
│---main.py
│---models.py
│---schemas.py
现在,让我们将以下代码添加到pollsapi/database.py
它的作用是创建数据库连接。
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
SQLALCHEMY_DATABASE_URL = "postgresql://YOUR_USERNAME:YOUR_PASSWORD@localhost:5432/DATABASE_NAME"
engine = create_engine(
SQLALCHEMY_DATABASE_URL
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
之后,将以下代码添加到pollsapi/models.py
它的作用与Django的models模型很类似,定义了我们的数据表的结构, 只不过是通过sqlalchemy实现的。
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, String, Integer, DateTime, ForeignKey
from sqlalchemy.orm import relationship
from database import Base
class Question(Base):
__tablename__ = "question"
id = Column(Integer, primary_key=True)
question_text = Column(String(200))
pub_date = Column(DateTime)
choices = relationship('Choice', back_populates="question")
class Choice(Base):
__tablename__ = "choice"
id = Column(Integer, primary_key=True)
question_id = Column(Integer, ForeignKey('question.id', ondelete='CASCADE'))
choice_text = Column(String(200))
votes = Column(Integer, default=0)
question = relationship("Question", back_populates="choices")
我们已经创建了由SQLAlchemy ORM提供的模型,通过模型我们可以简单地访问属性,如获得该特定问题的所有选择,比如question.choices 或choice.question。
好的,到目前为止还不错,我们现在将使用pydantic库来创建数据接口schema,它的主要作用是做类型强制检查,有点类似DRF的序列化器。继续将以下代码添加到pollsapi/schemas.py
class Config:
SQLAlchemy 中的定义参数类型与 Pydantic 不同,在 SQLAlchemy 使用的是大写String,并且将类型作为参数传递为此类=Column
question_text = Column(String)
而Pydantic风格声明使用:
和小写的str。
question_text: str
Pyndatic 模型/模组将映射到传入数据(POST、PUT 中的请求数据)和从 API 返回的响应数据。另一个重要的事情要了解的是我们在Question类里设置了orm_mode = True,这是因为默认Pydantic模型可以读取dict类型数据,不能直接读取ORM类型数据。如果数据是ORM模型,需要进行此项设置。
好的,我们现在将创建包含执行CRUD操作的所有功能。将以下代码添加到pollsapi/crud.py
from sqlalchemy.orm import Session
我们创建了所有用于 API 功能的实用功能。现在来了真正的文件,这将利用我们上面创建的所有文件。
创建投票问题
将以下行添加到pollsapi/main.py
访问:http://127.0.0.1:8000/docs/questions/
列出所有投票问题
代码语言:javascript复制@app.get("/questions/", response_model=List[schema.Question])
def get_questions(db: Session = Depends(get_db)):
return crud.get_all_questions(db=db)
注意在,返回对象列表,而不仅仅是一个对象,所以我们应该让我们的框架知道。尝试删除,我们的应用程序将抛出一个错误。Listresponse_modelcrud.get_all_questionsList
在这一点上,当你访问,你应该看到两个部分 - 和,点击 GET 部分,并尝试一下,你应该看到一个响应类似下面的东西http://127.0.0.1:8000/docsPOST /questions/GET /questions/
[
{
"question_text": "What is fastAPI?",
"pub_date": "2020-05-14T12:58:05.043000",
"id": 1
}]
获取、编辑和删除投票问题
代码语言:javascript复制def get_question_obj(db, qid):
obj = crud.get_question(db=db, qid=qid)
if obj is None:
raise HTTPException(status_code=404, detail="Question not found")
return obj
@app.get("/questions/{qid}", response_model=schema.QuestionInfo)
def get_question(qid: int, db: Session = Depends(get_db)):
return get_question_obj(db=db, qid=qid)
@app.put("/questions/{qid}", response_model=schema.QuestionInfo)
def edit_question(qid: int, question: schema.QuestionCreate, db: Session = Depends(get_db)):
get_question_obj(db=db, qid=qid)
obj = crud.edit_question(db=db, qid=qid, question=question)
return obj
@app.delete("/questions/{qid}")
def delete_question(qid: int, db: Session = Depends(get_db)):
get_question_obj(db=db, qid=qid)
crud.delete_question(db=db, qid=qid)
return {"detail": "Question deleted", "status_code": 204}
我们使用了不同的response_model,这是因为我们希望仅在问题详细信息 API 的情况下才显示 选项。
API 为特定的投票问题创建选择
代码语言:javascript复制@app.post("/questions/{qid}/choice", response_model=schema.ChoiceList)
def create_choice(qid: int, choice: schema.ChoiceCreate, db: Session = Depends(get_db)):
get_question_obj(db=db, qid=qid)
return crud.create_choice(db=db, qid=qid, choice=choice)
API 更新特定问题的投票
代码语言:javascript复制@app.put("/choices/{choice_id}/vote", response_model=schema.ChoiceList)
def update_vote(choice_id: int, db: Session = Depends(get_db)):
return crud.update_vote(choice_id=choice_id, db=db)
以下是问题和选择的终点
1. 创建问题 -POST http://127.0.0.1:8000/questions/
2. 列出所有问题 -GET http://127.0.0.1:8000/questions/
3. 检索特定问题-GET http://127.0.0.1:8000/questions/{qid}
4. 编辑特定问题 -PUT http://127.0.0.1:8000/questions/{qid}
5. 删除特定问题 -DELETE http://127.0.0.1:8000/questions/{qid}
6. 为特定的投票问题创建选项 -POST http://127.0.0.1:8000/questions/{qid}/choice
7. 更新特定问题的投票结果-PUT http://127.0.0.1:8000/choices/{choice_id}/vote