FastAPI：从零开始构建简单的应用程序结构

在本篇博文中，我们将从零开始搭建一个简单的FastAPI应用程序。这可以作为中小型项目的良好起点。

系列内容📖

第一部分：奠定基础（本文）
第二部分：迁徙
第三部分：Docker化

这篇文章我们将介绍哪些内容？📝

使用Poetry生成基础项目。
安装FastAPI、SQLAlchemy和其他依赖项。
创建必要的文件，这些文件将作为应用程序的基础。

开始之前……⚠️

我将做出以下假设：

您对 Python 和 Python类型有基本的了解；
您已安装以下软件：
- Python 3.6+（我使用的是 Python 3.8）
- PostgreSQL；以及
- 诗

事不宜迟，我们开始吧🙂

用诗歌🏃创建一个基础项目

打开终端并输入以下命令。

poetry new app

运行上述命令后，app创建了一个名为 `<directory_name>` 的新目录。cd现在让我们进入该目录。

cd app

目录结构应如下所示。

.
├── app
│   └── __init__.py
├── pyproject.toml
├── README.rst
└── tests
    ├── __init__.py
    └── test_app.py

我们快速浏览一下这里都有哪些东西。

该app目录是我们的主要 Python 包。

在 Python 中，包含文件的目录__init__.py被视为一个包。.py添加到此目录的任何文件都将被视为该包的模块。通常，此文件为空，但在此示例中，Poetry 已经添加了__version__ = '0.1.0'.

该pyproject.toml文件将用于添加所有依赖项。稍后，当我们开始安装依赖项时，你会注意到poetry.lock会创建一个文件，稍后会详细介绍。

该README文件可用于添加项目详情或任何有助于其他开发人员参与项目的实用说明。就我个人而言，我更喜欢使用 Markdown 而不是 reStructuredText 来编写文档，所以我将把它重命名README.rst为README.md。您可以随意进行重命名，也可以保持原样。

mv README.rst README.md

最后，我们得到了tests包含所有单元测试的目录。

安装 FastAPI 和其他依赖项📦

在本节中，我们将仅安装必要的依赖项，以启动一个基本的 CRUD（创建、读取、更新、删除）应用程序。

我们将安装什么？

FastAPI——这还用说吗🙂
SQLAlchemy对象关系映射器 (ORM)
psycopg2-binary PostgreSQL 数据库适配器
Uvicorn 是一款速度极快的 ASGI 服务器

poetry add fastapi sqlalchemy psycopg2-binary uvicorn

我们还将安装以下开发依赖项，主要是为了保持代码质量和进行测试。

pytest测试框架
Mypy是 Python 的静态类型检查器。
sqlalchemy-stubs是 Mypy 的插件，也是 SQLAlchemy 的类型存根。
用于代码检查的Flake8
autoflake会移除未使用的导入语句和变量。
排序导入语句
黑色Python 代码格式化程序

poetry add -D mypy sqlalchemy-stubs flake8 autoflake isort black

注意：上述命令中未包含pytest该命令，因为在使用 Poetry 创建新项目时，该命令通常默认安装。

目前，我们的目录结构实际上并没有改变，但您会注意到一个pyproject.toml文件已更新，并且poetry.lock创建了一个新文件。该poetry.lock文件会将已安装的依赖项锁定到特定版本。当多个开发人员共同开发同一个项目时，这尤其有用，可以确保每个人都使用相同版本的软件包。

总结一下，我们的目录结构现在应该看起来像这样。

.
├── app
│   └── __init__.py
├── poetry.lock
├── pyproject.toml
├── README.md
└── tests
    ├── __init__.py
    └── test_app.py

添加项目文件📄

在本节中，我们将开始添加构成应用程序基础的文件。

我们要创建的第一个文件是 `.htaccess`main.py文件，它将作为我们应用程序的入口点，并包含我们所有的路由。现在让我们在app`package` 目录下创建这个文件。

main.py

app/main.py

from fastapi import FastAPI

app = FastAPI()


@app.get("/")
def index():
    return {"message": "Hello world!"}

至此，我们已经有了一个可以运行的基本应用程序。如果我们切换回终端并运行以下命令。

poetry shell
uvicorn app.main:app

提示：如果您希望服务器在文件更改时重新加载，可以使用该--reload标志，如下所示uvicorn app.main:app --reload

现在，如果我们打开浏览器并访问http://127.0.0.1:8000 ，就会看到…… {"message":"Hello world!"}。FastAPI 还提供了开箱即用的 API 文档，所以如果您现在访问http://127.0.0.1:8000/docs，就会看到 Swagger UI。是不是很棒！🤘

我们稍后会回来更新main.py文件，但现在，让我们Ctrl+C在终端中输入命令停止 Uvicorn，然后继续添加其余文件。

接下来，我们db.py在同一目录下创建该文件。该文件将包含我们的数据库会话以及所有模型都将继承的基类。

db.py

app/db.py

from typing import Any

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import as_declarative
from sqlalchemy.orm import sessionmaker

from .config import settings

engine = create_engine(settings.SQLALCHEMY_DATABASE_URI, pool_pre_ping=True)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)


@as_declarative()
class Base:
    id: Any

信息：您可以在这里sessionmaker阅读更多关于该函数的信息，在这里阅读更多关于装饰器的信息。as_declarative

您可能已经注意到我们是settings从某个源导入的config，但我们实际上还没有创建该文件，所以现在让我们来创建它。

config.py

app/config.py

from typing import Any, Dict, Optional

from pydantic import BaseSettings, PostgresDsn, validator


class Settings(BaseSettings):
    POSTGRES_SERVER: str
    POSTGRES_USER: str
    POSTGRES_PASSWORD: str
    POSTGRES_DB: str

    SQLALCHEMY_DATABASE_URI: Optional[PostgresDsn] = None

    @validator("SQLALCHEMY_DATABASE_URI", pre=True)
    def assemble_db_connection(cls, v: Optional[str], values: Dict[str, Any]) -> Any:
        if isinstance(v, str):
            return v
        return PostgresDsn.build(
            scheme="postgresql",
            user=values.get("POSTGRES_USER"),
            password=values.get("POSTGRES_PASSWORD"),
            host=values.get("POSTGRES_SERVER"),
            path=f"/{values.get('POSTGRES_DB') or  ''}",
        )

    class Config:
        case_sensitive = True
        env_file = ".env"


settings = Settings()

提示：从.env文件加载配置时需要python-dotenv包。

这里我们使用了 Pydantic 的设置管理功能。默认情况下，该类会尝试使用`os.environ`BaseSettings读取系统级别的环境变量。但是，在我们的例子中，我们指定要从文件中读取环境变量。Pydantic 依赖于`python-dotenv`包来实现这一点，现在让我们将其添加为依赖项。.env

poetry add python-dotenv

现在我们将.env在项目目录的根目录下创建该文件。

.env

.env

# PostgreSQL
POSTGRES_SERVER=localhost
POSTGRES_USER=postgres
POSTGRES_PASSWORD=password
POSTGRES_DB=app

请务必编辑此文件以反映您的设置。

.gitignore

由于该.env文件可能包含敏感信息，我们不希望将其提交到版本控制系统中。因此，现在可能是将该文件添加到我们项目的合适时机。我们将复制GitHub 提供的.gitignorePython模板（链接在此）。.gitignore
.gitignore

wget https://raw.githubusercontent.com/github/gitignore/master/Python.gitignore
mv Python.gitignore .gitignore

接下来，我们将创建一个models.py文件schemas.py。

models.py

该models.py文件将包含我们所有继承自Base我们定义的 SQLAlchemy 类的模型db.py。现在我们将使用一个示例模型创建该文件User。

app/models.py

from uuid import uuid4
from sqlalchemy import Column, String, Text
from sqlalchemy.dialects.postgresql import UUID

from .db import Base


class Post(Base):
    __tablename__ = "posts"

    id = Column(UUID(as_uuid=True), primary_key=True, index=True, default=uuid4)
    title = Column(String)
    body = Column(Text)

schemas.py

现在我们来创建这个schemas.py文件。这个文件将包含我们所有的Pydantic 模型。FastAPI 在底层使用这些模型来验证传入的请求体、解析响应体，并自动生成API 文档。真是太棒了，至少我是这么认为的！👌

app/schemas.py

from typing import Optional

from pydantic import BaseModel, UUID4


# Shared properties
class PostBase(BaseModel):
    title: Optional[str] = None
    body: Optional[str] = None


# Properties to receive via API on creation
class PostCreate(PostBase):
    title: str
    body: str


# Properties to receive via API on update
class PostUpdate(PostBase):
    pass


class PostInDBBase(PostBase):
    id: Optional[UUID4] = None

    class Config:
        orm_mode = True


# Additional properties to return via API
class Post(PostInDBBase):
    pass


# Additional properties stored in DB
class PostInDB(PostInDBBase):
    pass

我们现在要创建的最后一个文件是该actions.py文件。该文件将包含我们所有的用例或要执行的操作，例如 CRUD 操作。

actions.py

app/actions.py

from typing import Any, Dict, Generic, List, Optional, Type, TypeVar, Union

from fastapi.encoders import jsonable_encoder
from pydantic import UUID4, BaseModel
from sqlalchemy.orm import Session

from . import schemas
from .db import Base
from .models import Post

# Define custom types for SQLAlchemy model, and Pydantic schemas
ModelType = TypeVar("ModelType", bound=Base)
CreateSchemaType = TypeVar("CreateSchemaType", bound=BaseModel)
UpdateSchemaType = TypeVar("UpdateSchemaType", bound=BaseModel)


class BaseActions(Generic[ModelType, CreateSchemaType, UpdateSchemaType]):
    def __init__(self, model: Type[ModelType]):
        """Base class that can be extend by other action classes.
           Provides basic CRUD and listing operations.

        :param model: The SQLAlchemy model
        :type model: Type[ModelType]
        """
        self.model = model

    def get_all(
        self, db: Session, *, skip: int = 0, limit: int = 100
    ) -> List[ModelType]:
        return db.query(self.model).offset(skip).limit(limit).all()

    def get(self, db: Session, id: UUID4) -> Optional[ModelType]:
        return db.query(self.model).filter(self.model.id == id).first()

    def create(self, db: Session, *, obj_in: CreateSchemaType) -> ModelType:
        obj_in_data = jsonable_encoder(obj_in)
        db_obj = self.model(**obj_in_data)  # type: ignore
        db.add(db_obj)
        db.commit()
        db.refresh(db_obj)
        return db_obj

    def update(
        self,
        db: Session,
        *,
        db_obj: ModelType,
        obj_in: Union[UpdateSchemaType, Dict[str, Any]]
    ) -> ModelType:
        obj_data = jsonable_encoder(db_obj)
        if isinstance(obj_in, dict):
            update_data = obj_in
        else:
            update_data = obj_in.dict(exclude_unset=True)
        for field in obj_data:
            if field in update_data:
                setattr(db_obj, field, update_data[field])
        db.add(db_obj)
        db.commit()
        db.refresh(db_obj)
        return db_obj

    def remove(self, db: Session, *, id: UUID4) -> ModelType:
        obj = db.query(self.model).get(id)
        db.delete(obj)
        db.commit()
        return obj


class PostActions(BaseActions[Post, schemas.PostCreate, schemas.PostUpdate]):
    """Post actions with basic CRUD operations"""

    pass


post = PostActions(Post)

在返回并更新main.py文件之前，让我们先回顾一下最终的目录结构。

.
├── app
│   ├── actions.py
│   ├── config.py
│   ├── __init__.py
│   ├── main.py
│   ├── models.py
│   └── schemas.py
├── poetry.lock
├── pyproject.toml
├── README.md
└── tests
    ├── __init__.py
    └── test_app.py

现在让我们更新main.py文件，把所有线索串联起来。

更新 main.py

app/main.py

from typing import Any, List

from fastapi import Depends, FastAPI, HTTPException
from pydantic import UUID4
from sqlalchemy.orm import Session
from starlette.status import HTTP_201_CREATED, HTTP_404_NOT_FOUND

from . import actions, models, schemas
from .db import SessionLocal, engine

# Create all tables in the database.
# Comment this out if you using migrations.
models.Base.metadata.create_all(bind=engine)

app = FastAPI()


# Dependency to get DB session.
def get_db():
    try:
        db = SessionLocal()
        yield db
    finally:
        db.close()


@app.get("/")
def index():
    return {"message": "Hello world!"}


@app.get("/posts", response_model=List[schemas.Post], tags=["posts"])
def list_posts(db: Session = Depends(get_db), skip: int = 0, limit: int = 100) -> Any:
    posts = actions.post.get_all(db=db, skip=skip, limit=limit)
    return posts


@app.post(
    "/posts", response_model=schemas.Post, status_code=HTTP_201_CREATED, tags=["posts"]
)
def create_post(*, db: Session = Depends(get_db), post_in: schemas.PostCreate) -> Any:
    post = actions.post.create(db=db, obj_in=post_in)
    return post


@app.put(
    "/posts/{id}",
    response_model=schemas.Post,
    responses={HTTP_404_NOT_FOUND: {"model": schemas.HTTPError}},
    tags=["posts"],
)
def update_post(
    *, db: Session = Depends(get_db), id: UUID4, post_in: schemas.PostUpdate,
) -> Any:
    post = actions.post.get(db=db, id=id)
    if not post:
        raise HTTPException(status_code=HTTP_404_NOT_FOUND, detail="Post not found")
    post = actions.post.update(db=db, db_obj=post, obj_in=post_in)
    return post


@app.get(
    "/posts/{id}",
    response_model=schemas.Post,
    responses={HTTP_404_NOT_FOUND: {"model": schemas.HTTPError}},
    tags=["posts"],
)
def get_post(*, db: Session = Depends(get_db), id: UUID4) -> Any:
    post = actions.post.get(db=db, id=id)
    if not post:
        raise HTTPException(status_code=HTTP_404_NOT_FOUND, detail="Post not found")
    return post


@app.delete(
    "/posts/{id}",
    response_model=schemas.Post,
    responses={HTTP_404_NOT_FOUND: {"model": schemas.HTTPError}},
    tags=["posts"],
)
def delete_post(*, db: Session = Depends(get_db), id: UUID4) -> Any:
    post = actions.post.get(db=db, id=id)
    if not post:
        raise HTTPException(status_code=HTTP_404_NOT_FOUND, detail="Post not found")
    post = actions.post.remove(db=db, id=id)
    return post

最后，如果我们再次运行服务器并访问http://127.0.0.1:8000/docs，我们现在就拥有了一个可以对我们的 Post 实体执行 CRUD 操作的基本 API。🚀

uvicorn app.main:app

结论💡

如果你已经看到这里了，太棒了！👍

我们创建了一个简单的应用程序，可以作为中小型项目的良好起点。在这个基础项目中，我们还可以添加一些功能，例如迁移或将 Docker 集成到我们的技术栈中。（提示：我们将在以后的文章中介绍这些内容，敬请期待 😉）

本文的最终代码可在GitHub上找到。

如果您喜欢阅读这篇文章，并希望了解更多内容，或者只是想与我联系，请在推特上关注我@alexvanzyl。

文章来源：https://dev.to/alexvanzyl/fastapi-simple-application-struct-from-scratch-2mem

菜单

分享

FastAPI：从零开始构建简单的应用程序结构

FastAPI：从零开始构建简单的应用程序结构

系列内容📖

这篇文章我们将介绍哪些内容？📝

开始之前……⚠️

用诗歌🏃创建一个基础项目

安装 FastAPI 和其他依赖项📦

我们将安装什么？

添加项目文件📄

main.py

db.py

config.py

.env

.gitignore

models.py

schemas.py

actions.py

更新 main.py

结论💡

系统设计面试中的 19 种微服务模式

使用 React 和 AWS Amplify 实现无服务器架构第三部分：跟踪应用使用情况

模型-视图-控制器（MVC）模式到底是什么？DEV 全球项目展示挑战赛，由 Mux 主办：快来展示你的项目吧！

我在两年内从 PHP 开发人员晋升为高级 C#/.NET 开发人员。

了解 Docker：第 12 部分 – 传递构建参数

Yarn 和第三方 NPM 客户端的黑暗未来 DEV 的全球展示与讲述挑战赛，由 Mux 呈现：展示你的项目！

CSS DEV 的全球展示挑战赛“响应式字体”由 Mux 呈现：展示你的项目！

我是如何以学生开发者的身份免费获得 Tabnine Pro 的，你也可以！

五大顶级JS框架

从 Rector PHP 开始：利用自动化改进您的 PHP 代码