使用FastAPI重写Django官网Polls教程

Python编程与实战

发布于 2021-07-06 14:51:02

1.5K0

发布于 2021-07-06 14:51:02

译者按：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安装。

pip install fastapi

你同时需要安装ASGI服务器。

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}

现在启动服务器：

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个文件。

└───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/

列出所有投票问题

@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
  }]

获取、编辑和删除投票问题

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 为特定的投票问题创建选择

@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 更新特定问题的投票

@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

本文参与腾讯云自媒体同步曝光计划，分享自微信公众号。

原始发表：2021-06-16，如有侵权请联系 cloudcommunity@tencent.com 删除

api