FastAPI是一个开源现代框架，用于在 Python 中构建 API。
PostgreSQL是一个开源的对象关系数据库管理系统。
在本教程中，我们将使用 Fast API 构建示例 RESTful API，并利用 PostgreSQL 持久数据的强大功能。然后，我们将使用Dockerfile和Docker Compose文件对 API 和数据库进行容器化。Dockerfile 是一个文本文件，其中包含将在 Docker Compose 文件中执行以构建容器的一系列指令。Docker compose是一个定义和共享多容器Docker容器的工具。我们的应用程序将包含两个容器。Fast API 容器和 PostgreSQL 容器。
前提要求与工具
Docker - 您需要对 Docker 的工作原理有基本的了解。要了解 docker 的工作原理，您可以前往我之前的docker 入门帖子。您将学习如何安装 docker、docker 的工作原理以及 docker 命令。
Python - 您需要在计算机上安装 Python。最好是Python 3.10。
VSCode
相关网址
FastAPI - https://fastapi.tiangolo.com/
PostgreSQL - https://www.postgresql.org
Dockerfile - https://docs.docker.com/engine/reference/builder/
Docker Compose - https://docs.docker.com/get-started/08_using_compose/
docker 入门 - https://dev.to/mbuthi/docker-2oge
Python官网 - https://www.python.org/
VsCode官网 - https://code.visualstudio.com/download
FastAPI 入门
我们将使用 Python 构建一个产品列表示例应用程序，用户将能够通过 API 执行 CRUD 操作。然后我们将使用 PostgreSQL 来保存产品数据。但是，我们需要了解项目目录结构是什么样的。下面是 FastAPI 中项目目录结构的快照：
. └── FastAPI_APP/ ├── app/ │ ├── api/ │ │ ├── v1/ │ │ │ ├── endpoints/ │ │ │ │ ├── __init__.py │ │ │ │ └── products.py │ │ │ ├── __init__.py │ │ │ └── api.py │ │ ├── __init__.py │ │ └── deps.py │ ├── core/ │ │ ├── __init__.py │ │ └── settings.py │ ├── crud/ │ │ ├── __init__.py │ │ ├── base.py │ │ └── product.py │ ├── db/ │ │ ├── __init__.py │ │ └── session.py │ ├── models/ │ │ ├── __init__.py │ │ ├── basemodel.py │ │ └── products.py │ ├── schemas/ │ │ ├── __init__.py │ │ └── product.py │ └── utils/ │ ├── __init__.py │ └── idgen.py └── main.py
粗略地概括一下：
FastAPI_APP - 这是我们应用程序的根目录。
app - 保存我们 API 的服务。
main.py - 这是 API 入口点。
api - 包含 API 端点。
core - 包含核心功能，例如设置和日志记录。
crud - 包含 CRUD（创建、读取、更新、删除）操作。
db - 包含与数据库相关的代码。
models - 包含数据库模型。
utils - 包含实用函数和类。
要开始构建 API，您需要安装 Vscode 或您喜欢的 IDE。然后使用上面所示的目录结构创建一个新项目。
设置我们的docker环境
我们将首先为我们的应用程序创建一个 docker compose 文件和一个 docker 文件。
因此，前往 vscode code 打开名为Dockerfile的文件并粘贴以下说明。
# 使用来自 docker hub 的 python 官方镜像 FROM python:3.12-rc-slim-bullseye # 防止 pyc 文件被复制到容器中 ENV PYTHONDONTWRITEBYTECODE 1 # 确保 python 输出记录在容器的终端中 ENV PYTHONUNBUFFERED 1 RUN apt-get update \ # 构建 Python 包的依赖项 && apt-get install -y build-essential \ # psycopg2 依赖项 && apt-get install -y libpq-dev \ # Translations 依赖项 && apt-get install -y gettext \ # 清理未使用的文件 && apt-get purge -y --auto-remove -o APT::AutoRemove::RecommendsImportant=false \ && rm -rf /var/lib/apt/lists/* # 将文件“requirements.txt”从本地构建上下文复制到容器的文件系统。 COPY ./requirements.txt /requirements.txt # 安装python依赖项 RUN pip install -r /requirements.txt # 设置工作目录 WORKDIR /app # 运行 Uvicorn 来启动您的 Python Web 应用程序 CMD ["uvicorn", "main:app", "--host", "0.0.0.0" "--port" "8000"]
接下来，我们将创建 docker compose 文件。在 vscode 中打开docker-compose.yml文件并粘贴以下说明。
# 指定 compose 版本version: '3.8'# 为我们的 docker compose setupservices 指定服务:
api:
build:
context: .
# 指定我们的 dockerfile 路径。 我们的 Dockerfile 位于根文件夹中
dockerfile: .
# 指定图像名称
image: products_api
# 该卷用于将主机上的文件和文件夹映射到容器
# 因此，如果我们更改主机上的代码，docker容器中的代码也会更改
volumes:
- .:/app
# 将主机上的8000端口映射到容器上的8000端口
ports:
- 8000:8000
# 指定.env文件路径
env_file:
- ./env
# 定义对“products_db”服务的依赖，因此它首先启动
depends_on:
- products_db
products_db:
# 指定我们数据库的图像名称
# 如果在我们的本地存储库中找不到该图像
# 将从 Docker Hub 的 docker 注册表中拉取
image: 16rc1-alpine3.18
# 安装卷以保存 postgreSQL 数据
volumes:
- postgres_data:/var/lib/postgresql/data/
environment: # Use environment variables for db configuration
- POSTGRES_USER=${POSTGRES_USER}
- POSTGRES_PASSWORD=${POSTGRES_PASSWORD}
- POSTGRES_DATABASE=${POSTGRES_DATABASE}# Define a volume for persisting postgreSQL datavolumes:
postgres_data:FastAPI 的环境变量。
接下来，我们将创建一个.env文件并实例化我们的环境变量。在你的 vscode 中打开 .env 文件并包含以下变量
# PostgreSQL 数据库主机 POSTGRES_HOST=products_db # PostgreSQL 数据库用户 POSTGRES_USER=username # PostgreSQL 数据库密码 POSTGRES_PASSWORD=password # PostgreSQL 数据库名称 POSTGRES_DATABASE=database # PostgreSQL 数据库端口 POSTGRES_PORT=5432 # 用于连接 PostgreSQL 的异步数据库 URI ASYNC_DATABASE_URI=postgresql+asyncpg://username:password@products_db:5432/database # 项目或应用程序的名称 PROJECT_NAME=Product Listings
.env 文件包含敏感变量。将这些
敏感变量包含在 .env 文件中始终是一个好习惯。
生成唯一 ID。
在我们的 FastAPI 应用程序中，我们将定义一个强大的实用函数来生成唯一的 ID。该函数将使用 UUID 模块。转到 utils module/ 文件夹并打开 idgen.py 文件并粘贴下面的代码片段。
import uuid def idgen() -> str: # 生成随机 uuid 字符串 return str(uuid.uuid4().hex)
FastAPI 中的设置配置
接下来，我们将创建一个配置类。该类将从 pydantic 基设置类继承。该类将负责将环境变量加载到应用程序上下文并定义其他应用程序设置。在 vscode 中打开名为 settings.py 的文件并插入以下代码片段。
# 导入包
from pydantic_settings import BaseSettings
import os
from dotenv import load_dotenv
import secrets
load_dotenv()
class Settings(BaseSettings):
"""
应用程序设置和配置参数
此类使用 pydantic 数据验证库定义应用程序设置
"""
PROJECT_NAME: str = os.getenv("PROJECT_NAME")
API_V1_STR: str = "/api/v1"
ASYNC_DATABASE_URI: str = os.getenv("ASYNC_DATABASE_URI")
SECRET_KEY: str = secrets.token_urlsafe(32)
settings = Settings()在 Fast API 中创建我们的模型
我们将从创建一个基类开始。基类将包含所有模型中的公共属性。这将有助于保持我们的代码干燥。打开位于 models 文件夹中名为 base.py 的文件。在文件内粘贴以下代码片段
from sqlalchemy import DateTime, func from sqlalchemy.orm import Mapped, declared_attr, DeclarativeBase, mapped_column from app.utils.idgen import idgen from datetime import datetime class Base_(DeclarativeBase): """ SQLAlchemy 模型的基类，具有保持 DRY（不要重复）的公共属性。 此类旨在充当 SQLAlchemy 模型的基类。 它定义了常见的属性，例如表名、创建时间戳、 并更新可以被其他模型继承的时间戳，帮助您 坚持 DRY（不要重复自己）原则。 属性: __tablename__ (str): 表名，源自小写的类名。 id (str): 每条记录的唯一ID。 created_on (datetime): 创建记录的时间戳。 updated_on (datetime, optional): 上次更新记录的时间戳。 默认为“无”，直到发生更新。 示例: 要使用此基类创建 SQLAlchemy 模型： class YourModel(Base_): # 在此处为您的模型定义其他属性。 """ @declared_attr def __tablename__(cls): # 表名源自小写的类名 return cls.__name__.lower() # 每条记录的唯一 UUID ID id: Mapped[str] = mapped_column(primary_key=True, default=idgen,index=True) # 记录创建的时间戳 created_on: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now()) # 记录更新的时间戳，最初为 None，直到发生更新 updated_on: Mapped[datetime] = mapped_column(DateTime(timezone=True), onupdate=func.now(), nullable=True)
基类包含 id 属性，它是每个记录的唯一 UUID，还包含created_on 属性，它是记录创建的时间戳，还包含updated_on，它是记录更新的时间戳。
接下来，我们将定义我们产品的模型。该模型将从基类继承Base_。打开名为 products.py 的文件，该文件包含在名为 models 的文件夹中。在文件内，粘贴以下代码片段。
from sqlalchemy.orm import Mapped, mapped_column from sqlalchemy import String from .base import Base_ class Product(Base_): """ 这是用于定义产品模型的 SQLAlchemy 类。 它继承了Base_类的所有属性和方法。 该类定义了常用属性，例如名称、价格图像、和重量。 属性： name (str): 产品名称 price (str): 产品价格 image (str): 产品图片 url weight (str): 产品价格 'nullable=False' 表示这些列在数据库中不能有 NULL 值。 """ name: Mapped[str] = mapped_column(String(30), index=True, nullable=False) price: Mapped[str] = mapped_column(String(30), nullable=False) image: Mapped[str] = mapped_column(String, nullable=False) weight: Mapped[str] = mapped_column(String, nullable=False)
这Mapped[str]只是一个 Python 类型的提示。它强调该属性将保存字符串类型的值。取代mapped_column了之前的 sqlalchemy Column。
在 Fast API 中创建模式。
我们现在将定义我们的pydantic模式。这些模式充当数据类，定义某个 API 端点期望接收哪些数据，以便将请求视为有效请求。它们还可以在 Fast API 中使用来定义响应模型，即端点返回的响应。打开名为product.py 的文件（包含在名为 的文件夹中）schemas并粘贴以下代码片段。
from typing import Optional from pydantic import BaseModel class ProductBase(BaseModel): name: str # 产品名称（必填） price: str # 产品价格（必填） image: str # 产品图片的 URL 或路径（必填） weight: str # 产品的重量（必填） class ProductCreate(ProductBase): ... class ProductUpdate(ProductBase): ... class ProductPatch(ProductBase): name: Optional[str] # 修补时名称是可选的 price: Optional[str] # 修补时价格是可选的 image: Optional[str] # 修补时图像是可选的 weight: Optional[str] # 修补时权重是可选的 class Product(ProductBase): id: str class Config: orm_mode = True
Optional从 Python 类型模块导入，该模块定义该字段不是必需的，因此可以为 None。
在 Fast API 中创建 CRUD 操作
我们现在将定义创建、读取、更新和删除方法。首先，我们将为操作创建一个基类。基类类将有助于维护 Python 中的 DRY 代码设计。各种SQLAlchemy模型也会继承该类来执行数据库操作。因此，打开名为 crud 的文件夹中名为 base.py 的文件。粘贴下面的代码片段。
from typing import Any, Dict, Generic, Optional, Type, TypeVar
from pydantic import BaseModel
from sqlalchemy.ext.declarative import DeclarativeMeta
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.future import select
from sqlalchemy import func, update
from fastapi.encoders import jsonable_encoder
ModelType = TypeVar("ModelType", bound=DeclarativeMeta)
CreateSchemaType = TypeVar("CreateSchemaType", bound=BaseModel)
UpdateSchemaType = TypeVar("UpdateSchemaType", bound=BaseModel)
class CRUDBase(Generic[ModelType, CreateSchemaType, UpdateSchemaType]):
"""
SQLAlchemy 模型的通用 CRUD（创建、读取、更新、删除）操作。
此类提供了一组可与 SQLAlchemy 模型一起使用的通用 CRUD 操作。
它包括创建、检索、更新和删除数据库中记录的方法。
参数:
model (Type[ModelType]): 要执行 CRUD 操作的 SQLAlchemy 模型类。
例子:
为特定模型（例如用户模型）创建 CRUD 实例：
```
python
crud_user = CRUDBase[Prodcut, ProductCreateSchema, ProductUpdateSchema]
```
"""
def __init__(self, model: Type[ModelType]):
self.model = model
# 获取单个实例
async def get(self, db: AsyncSession, obj_id: str) -> Optional[ModelType]:
query = await db.execute(select(self.model).where(self.model.id == obj_id))
return query.scalar_one_or_none()
# 获取所有多个实体
async def get_multi(self, db: AsyncSession, *, skip: int = 0, limit: int = 100) -> ModelType:
query = await db.execute(select(self.model))
return query.scalars().all()
# 搜索特定实体
async def get_by_params(self, db: AsyncSession, **params: Any) -> Optional[ModelType]:
query = select(self.model)
for key, value in params.items():
if isinstance(value, str):
query = query.where(func.lower(getattr(self.model, key)) == func.lower(value))
else:
query = query.where(getattr(self.model, key) == value)
result = await db.execute(query)
return result.scalar_one_or_none()
# 添加实体
async def get_or_create(self, db: AsyncSession,
defaults: Optional[Dict[str, Any]], **kwargs: Any) -> ModelType:
instance = await self.get_by_params(db, **kwargs)
if instance:
return instance, False
params = defaults or {}
params.update(kwargs)
instance = self.model(**params)
db.add(instance)
await db.commit()
await db.refresh(instance)
return instance, True
# 部分更新实体
async def patch(self, db: AsyncSession,
*, obj_id: str,
obj_in: UpdateSchemaType | Dict[str, Any]
) -> Optional[ModelType]:
db_obj = await self.get(db=db, obj_id=obj_id)
if not db_obj:
return None
update_data = obj_in if isinstance(obj_in, dict) else obj_in.model_dump(exclude_unset=True)
query = (
update(self.model)
.where(self.model.id == obj_id)
.values(**update_data)
)
await db.execute(query)
return await self.get(db, obj_id)
# 完全更新实体
async def update(
self,
db: AsyncSession,
*,
obj_current: ModelType,
obj_new: UpdateSchemaType | Dict[str, Any] | ModelType
):
obj_data = jsonable_encoder(obj_current)
if isinstance(obj_new, dict):
update_data = obj_new
else:
update_data = obj_new.model_dump(exclude_unset=True)
for field in obj_data:
if field in update_data:
setattr(obj_current, field, update_data[field])
db.add(obj_current)
await db.commit()
await db.refresh(obj_current)
return obj_current
# 从数据库中完全删除实体
async def remove(self, db: AsyncSession, *, obj_id: str) -> Optional[ModelType]:
db_obj = await self.get(db, obj_id)
if not db_obj:
return None
await db.delete(db_obj)
await db.commit()
return db_obj我们定义了各种方法。get 方法从数据库中获取与对象 ID 匹配的单个记录。get_multi 方法从数据库获取分页文档。get_by_params 方法根据匹配的参数搜索匹配的记录。get_or_create方法首先检查实体是否存在，如果不存在，则在数据库中创建实体。patch 方法更新记录字段。update方法完全更新记录字段。remove 方法从数据库中删除一条记录。
定义了 CRUD 操作的基类后，我们现在将定义产品 CRUD 操作。Product CRUD 操作将从基类继承CRUDBase。打开 crud 文件夹中名为product.py 的文件。粘贴下面的代码片段。
from typing import Any, Coroutine, Dict, Optional from fastapi_pagination import Page from sqlalchemy.ext.asyncio import AsyncSession from .base import CRUDBase from app.schemas.product import ProductUpdate, ProductCreate from app.models.product import Product class CRUDProduct(CRUDBase[Product, ProductCreate, ProductUpdate]): async def get(self, db: AsyncSession, obj_id: str) -> Product: return await super().get(db, obj_id) async def get_or_create(self, db: AsyncSession, defaults: Dict[str, Any] | None, **kwargs: Any) -> Product: return await super().get_or_create(db, defaults, **kwargs) async def get_multi(self, db: AsyncSession, *, skip: int = 0, limit: int = 20) -> Page[Product]: return await super().get_multi(db, skip=skip, limit=limit) async def update(self, db: AsyncSession, *, obj_current: Product, obj_new: ProductUpdate | Dict[str, Any] | Product): return await super().update(db, obj_current=obj_current, obj_new=obj_new) async def remove(self, db: AsyncSession, *, obj_id: str) -> Product | None: return await super().remove(db, obj_id=obj_id) product = CRUDProduct(Product)
创建数据库会话
这里我们将定义一个异步数据库引擎来对数据库执行异步操作。然后我们将引擎绑定到 sessionmaker，它将与数据库异步交互。打开名为 db 的文件夹中包含的名为 session.py 的文件。粘贴下面的代码片段。
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession from sqlalchemy.orm import sessionmaker from app.core.settings import settings # 使用应用程序设置中的 ASYNC_DATABASE_URI 创建异步 SQLAlchemy 引擎。 engine = create_async_engine( settings.ASYNC_DATABASE_URI, ) # 使用 sessionmaker 创建一个 AsyncSession 类，绑定到 SQLAlchemy 引擎。 # 该会话类将用于与数据库异步交互。 SessionLocal = sessionmaker( engine, expire_on_commit=False, class_=AsyncSession )
create_async_engine- 这将使用应用程序设置中的 ASYNC_DATABASE_URI 创建异步 SQLAlchemy 引擎。
sessionmaker- 这将使用 sessionmaker 创建一个 AsyncSession 类，绑定到 SQLAlchemy 引擎。
创建快速 API 依赖项
在这里，我们将定义将在我们的应用程序中使用的所有依赖项。这可能包括数据库会话。打开 API 文件夹中包含的名为 deps.py 的文件并粘贴下面的代码片段。
from typing import AsyncGenerator from sqlalchemy.ext.asyncio import AsyncSession from app.db.session import SessionLocal async def get_db() -> AsyncGenerator[AsyncSession, None]: async with SessionLocal() as db: yield db
该get_db函数是一个异步生成函数，用于生成数据库会话。
创建产品列表端点
这里我们将定义 POST、GET、PUT、PATCH 和 DELETE 方法。
POST 将创建一个新产品。
GET 将检索一个或多个产品。
PUT 将完全更新产品。
PATCH 将更新为产品指定的字段。
DELETE 将从数据库中删除产品。
转到代码编辑器并打开名为 products.py 的文件，该文件包含在名为 endpoints 的文件夹中。在文件内粘贴下面的代码片段。
# 导入必要的模块和组件
from typing import Annotated
from fastapi import APIRouter, status, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from fastapi_pagination import Page, paginate
from app.schemas.product import Product, ProductCreate, ProductPatch, ProductUpdate
from app.api.deps import get_db
from app import crud
# 创建 APIRouter 实例
router = APIRouter()
# 定义创建新产品的路线
@router.post("/", response_model=Product, status_code=status.HTTP_201_CREATED)
async def create_product(
db: Annotated[AsyncSession, Depends(get_db)],
product_in: ProductCreate
):
# 使用“crud”模块中的 CRUD（创建、读取、更新、删除）操作
# 创建新产品或返回现有产品（如果已存在）
product, created = await crud.product.get_or_create(
db=db, defaults=product_in.dict()
)
# 如果产品已存在，则引发带有 400 状态代码的 HTTPException
if not created:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Product exists"
)
# 返回创建的或现有的产品
return product
# 定义通过 ID 检索产品的路由
@router.get("/{productId}", response_model=Product, status_code=status.HTTP_200_OK)
async def get_product(
db: Annotated[AsyncSession, Depends(get_db)],
productId: str
):
# 使用 CRUD 操作通过 ID 检索产品
product = await crud.product.get(db=db, obj_id=productId)
# 如果产品不存在，则引发带有 404 状态代码的 HTTPException
if not product:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Product not found"
)
# 返回检索到的产品
return product
# 定义一个路由来检索分页的产品列表
@router.get("/", response_model=Page[Product], status_code=status.HTTP_200_OK)
async def get_products(
db: Annotated[AsyncSession, Depends(get_db)],
skip: int = 0,
limit: int = 20
):
# 使用CRUD操作检索多个带分页的产品
products = await crud.product.get_multi(db=db, skip=skip, limit=limit)
# 如果未找到产品，则引发带有 404 状态代码的 HTTPException
if not products:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Products not found"
)
# 返回分页的产品列表
return paginate(products)
# 定义部分更新产品的路线
@router.patch("/{productId}", status_code=status.HTTP_200_OK)
async def patch_product(
db: Annotated[AsyncSession, Depends(get_db)],
product_Id: str,
product_in: ProductPatch
):
# 使用 CRUD 操作通过 ID 检索产品
product = await crud.product.get(db=db, obj_id=product_Id)
# 如果产品不存在，则引发带有 404 状态代码的 HTTPException
if not product:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Product not found"
)
# 使用CRUD操作来修补（部分更新）产品
product_patched = await crud.product.patch(db=db, obj_id=product_Id, obj_in=product_in.dict())
# 返回已修补的产品
return product_patched
# 定义完全更新产品的路线
@router.put("/{productId}", response_model=Product, status_code=status.HTTP_200_OK)
async def update_product(
db: Annotated[AsyncSession, Depends(get_db)],
productId: str,
product_in: ProductUpdate
):
# 使用 CRUD 操作通过 ID 检索产品
product = await crud.product.get(db=db, obj_id=productId)
# 如果产品不存在，则引发带有 404 状态代码的 HTTPException
if not product:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Product not found"
)
# 使用CRUD操作来全面更新产品
product_updated = await crud.product.update(
db=db, obj_current=product, obj_new=product_in
)
# 返回更新后的产品
return product_updated
# 定义删除产品的路线
@router.delete("/{productId}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_product(
db: Annotated[AsyncSession, Depends(get_db)],
productId: str
):
# 使用 CRUD 操作通过 ID 检索产品
product = await crud.product.get(db=db, obj_id=productId)
# 如果产品不存在，则引发带有 404 状态代码的 HTTPException
if not product:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Product not found"
)
# 使用CRUD操作移除（删除）产品
await crud.product.remove(db=db, obj_id=productId)
# 返回 204 No Content 响应表示删除成功
return端点由注释组成，解释每个端点中发生的情况。
现在我们需要将端点公开给 API 入口点，以便我们编辑两个文件。对于第一个文件，我们将打开名为 v1 的文件夹内的 api.py 文件。然后粘贴下面的代码片段。
# 从 FastAPI 导入 APIRouter 类 from fastapi import APIRouter # 从“app.api.v1.endpoints”模块导入“products”路由器 from app.api.v1.endpoints import products # 创建 APIRouter 的实例 router = APIRouter() # 将“products”路由器作为子路由器包含在“/products”前缀下 # 并分配标签“Products”以对相关 API 端点进行分组 router.include_router(products.router, prefix="/products", tags=["Products"])
然后打开 main.py 文件并粘贴下面的代码片段。
# 从FastAPI框架导入FastAPI类
from fastapi import FastAPI
# 导入add_pagination
from fastapi_pagination import add_pagination
# 从“app.api.v1.api”模块导入“路由器”
from app.api.v1.api import router
#从“app.core.settings”模块导入“settings”对象
from app.core.settings import settings
# 创建 FastAPI 应用程序的实例
# - “title”设置为“settings”中的项目名称
# - 'openapi_url' 指定 OpenAPI 文档的 URL
app = FastAPI(
title=settings.PROJECT_NAME,
openapi_url=f"{settings.API_V1_STR}/openapi.json"
)
# 为所有使用 paginate 的路由添加必要的分页参数
add_pagination(app)
# 在 FastAPI 应用程序中包含“路由器”（其中包含您的 API 路由）
app.include_router(router)到目前为止，我们可以尝试启动我们的服务器。为此，我们必须使用 docker 容器和 docker 镜像构建我们的应用程序。
运行产品列表 API
在这里我们将尝试运行我们的 API。
假设您本地计算机上安装了docker，请打开 vscode 终端。
要打开终端：
Windows 使用快捷键ctrl + `。
Mac OS 使用快捷键 ⌘ +`。
Linux 使用快捷键Ctrl+Shift+`。
在终端中写入以下命令：
docker-compose -f docker-compose.yml up -d
docker-compose- 此命令用于使用 Docker compose 管理 Docker 容器。
-f - 用于指定撰写文件的路径。
docker-compose.yml - 这是定义容器的 compose 文件的路径。在我们的例子中是 docker-compose.yml。
up - 用于初始化和启动 compose 文件中指定的服务。在我们的例子中，它启动products_db和api服务。
-d - 这指定容器应以分离模式启动，即容器作为后台服务启动。
成功执行命令后，您可以通过在 vscode 终端中执行以下命令来验证容器确实已启动并正在运行：
docker ps
您应该能够看到以下输出：
要通过 Swagger 查看 API 文档，您可以打开您的首选浏览器并粘贴以下 URL：
http://localhost:8000/docs
默认情况下，我们将通过端口 8000 访问 API，因为这是我们之前在 docker compose 文件中指定的映射到主机的端口。
在您的浏览器中，您将能够看到类似以下内容：
我们现在已经成功设置了用于产品列表的 API。但是，如果我们尝试在 Swagger 中执行 POST 请求，我们将收到 500 内部服务器错误。
要查看导致错误的原因，我们将查看api容器日志。要查看日志，我们可以使用docker 桌面或使用我们的终端来查看日志。为此，我们将在 vscode 终端中执行以下命令：
docker logs <CONTAINER ID>
是当前正在运行的容器CONTAINER ID的 ID.api
为了获得CONTAINER ID我们将运行：
docker ps
成功运行docker logs命令后，我们会得到如下错误，如下图所示：
在最后一行，我们可以清楚地看到日志表明"database" does not exist. 之前，我们已经在**.env**我们的 POSTGRES_DATABASE=database 中定义了。而这个名为database的数据库不存在。这意味着我们实际上必须首先创建数据库本身。
为了创建数据库，我们将使用products_db容器。
在你的 vscode 终端中：
运行下面的命令
docker exec -it <CONTAINER ID> /bin/bash
上面的命令在容器内启动 Bash 终端。
运行docker ps以获取products_dbID 并将其替换CONTAINER ID为您的映像实例的 ID products_db。
我们需要创建数据库。为此，我们将在容器 Bash 终端中运行以下一系列命令：
psql -U username
上述命令启动 PostgreSQL 的基于终端的前端。它允许我们交互式地输入查询。
CREATE DATABASE database;
上面的命令在 PostgreSQL 中创建一个名为database的数据库。
ALTER ROLE username WITH PASSWORD 'password';
上述命令更改角色用户名并为其分配密码password。
GRANT ALL PRIVILEGES ON DATABASE database TO username;
上述命令将所有数据库权限授予名为 username 的用户。
完成此操作后，我们现在需要执行数据库迁移。
快速 API 数据库迁移
数据库迁移或模式迁移是为修改关系数据库中对象的结构而开发的受控更改集。
为了在我们的 API 中执行迁移，我们将在项目根目录中创建一个 alembic.ini 文件和一个 alembic 文件夹。在 alembic 文件夹内创建另一个名为 versions 的文件夹以及两个名为 env.py 和 script.py.mako 的文件。
现在项目目录结构如下所示：
. └── FastAPI_APP/ ├── app/ │ ├── alembic.ini │ ├── alembic/ │ │ ├── versions │ │ ├── env.py │ │ └── script.py.mako │ ├── api/ │ │ ├── v1/ │ │ │ ├── endpoints/ │ │ │ │ ├── __init__.py │ │ │ │ └── products.py │ │ │ ├── __init__.py │ │ │ └── api.py │ │ ├── __init__.py │ │ └── deps.py │ ├── core/ │ │ ├── __init__.py │ │ └── settings.py │ ├── crud/ │ │ ├── __init__.py │ │ ├── base.py │ │ └── product.py │ ├── db/ │ │ ├── __init__.py │ │ └── session.py │ ├── models/ │ │ ├── __init__.py │ │ ├── basemodel.py │ │ └── products.py │ ├── schemas/ │ │ ├── __init__.py │ │ └── product.py │ └── utils/ │ ├── __init__.py │ └── idgen.py └── main.py
我们现在将编辑已添加的文件。
打开 alembic.ini 文件并粘贴以下脚本：
# 通用的单一数据库配置。 [alembic] # 迁移脚本的路径 script_location = alembic # 用于生成迁移文件名的模板； 默认值为 %%(rev)s_%%(slug)s file_template = %%(year)d-%%(month).2d-%%(day).2d-%%(hour).2d-%%(minute).2d_%%(rev)s # sys.path path, will be prepended to sys.path if present. # defaults to the current working directory. prepend_sys_path = . version_path_separator = os # Use os.pathsep. Default configuration used for new projects. sqlalchemy.url = [post_write_hooks] # Logging configuration [loggers] keys = root,sqlalchemy,alembic [handlers] keys = console [formatters] keys = generic [logger_root] level = WARN handlers = console qualname = [logger_sqlalchemy] level = WARN handlers = qualname = sqlalchemy.engine [logger_alembic] level = INFO handlers = qualname = alembic [handler_console] class = StreamHandler args = (sys.stderr,) level = NOTSET formatter = generic [formatter_generic] format = %(levelname)-5.5s [%(name)s] %(message)s datefmt = %H:%M:%S
alembci.ini 文件是与 Alembic 一起使用的配置文件，它提供用于管理数据库架构随时间变化的设置和选项。
打开 alembic 文件夹或模块中包含的 env.py 文件并粘贴以下代码片段：
# Import necessary modules
import asyncio
import sys
import pathlib
from alembic import context
from sqlalchemy.ext.asyncio import create_async_engine
# Import the necessary database models and settings
from app.models.product import Product
from app.core.settings import settings
from app.models.base import Base_
from sqlalchemy.orm import declarative_base
# Define the target metadata for migrations
target_metadata = Base_.metadata
# Append the parent directory of the current file to the sys.path
# This allows importing modules from the parent directory
sys.path.append(str(pathlib.Path(__file__).resolve().parents[1]))
# Define a function to run migrations
def do_run_migrations(connection):
context.configure(
compare_type=True,
dialect_opts={"paramstyle": "named"},
connection=connection,
target_metadata=target_metadata,
include_schemas=True,
version_table_schema=target_metadata.schema,
)
with context.begin_transaction():
context.run_migrations()
# Define an asynchronous function to run migrations online
async def run_migrations_online():
"""Run migrations in 'online' mode.
In this scenario, we create an Engine
and associate a connection with the context.
"""
# Create an asynchronous database engine using the URI from settings
connectable = create_async_engine(settings.ASYNC_DATABASE_URI, future=True)
# Connect to the database and run migrations within a transaction
async with connectable.connect() as connection:
await connection.run_sync(do_run_migrations)
# Run the migrations online using asyncio
asyncio.run(run_migrations_online())上面的脚本使用 Alembic 在 Fast API 中作为异步数据库引擎运行数据库迁移。
打开 alembic 模块中包含的名为 script.py.mako 的文件。粘贴下面的脚本：
"""
Revision ID: ${up_revision}
Revises: ${down_revision | comma,n}
Create Date: ${create_date}
"""
# Import necessary modules from Alembic and SQLAlchemy
from alembic import op
import sqlalchemy as sa
# Import any additional necessary modules (if specified)
${imports if imports else ""}
# Define revision identifiers used by Alembic
revision = ${repr(up_revision)} # The unique identifier for this revision
down_revision = ${repr(down_revision)} # The revision to which this one applies (if any)
branch_labels = ${repr(branch_labels)} # Labels associated with this revision (if any)
depends_on = ${repr(depends_on)} # Dependencies for this revision (if any)
def upgrade():
${upgrades if upgrades else "pass"}
"""
This function is called when upgrading the database schema.
You can specify SQL operations to apply schema changes.
If no operations are specified, 'pass' can be used.
"""
def downgrade():
${downgrades if downgrades else "pass"}
"""
This function is called when downgrading the database schema.
You can specify SQL operations to reverse schema changes.
If no operations are specified, 'pass' can be used.
"""定义了用于处理迁移的脚本后，我们现在可以在 api 容器中执行它们。为此，运行以下命令：
docker exec -it <CONTAINER ID> /bin/bash
上面的命令在容器内启动 Bash 终端。
将 替换为 api 容器的实际 ID。要获取 api 容器，请运行docker ps命令。
alembic revision --autogenerate -m "Migrate products table"
上述命令生成一个新的迁移脚本。新的迁移脚本包含当前数据库架构与代码中的模型定义之间的差异。
alembic upgrade head
上述命令适用于所有待处理的迁移。
测试我们的 API
由于我们已经执行了数据库模式的迁移，因此我们现在可以通过 swagger 文档自信地测试我们的 API。
要访问 Swagger 文档，请在浏览器中输入以下 URL：
http://localhost:8000/docs
我们可以通过执行 POST 请求开始。
POST 请求快速 API
在 Swagger 中，展开可折叠的 POST 请求，然后单击Try it out按钮。在“响应正文”部分中，将 JSON 架构的键值更改为您的首选项，如下所示。
对于图像键，您可以输入图像 URL。然后单击执行按钮。成功 POST 后，您将看到 201 创建的状态代码以及响应正文，如下所示：
根据您分配给 JSON 架构的值，响应正文可能与上面显示的有所不同。
GET 请求（分页数据）
在 GET 请求中，我们想要获取多个项目。为此，我们可以指定跳过和限制。
Skip 与 OFFSET 类似，是在检索任何行之前要跳过的结果表的行数。
Limit 是指定获取结果表的前 N 行的语法。
单击获取请求，对于跳过参数，我们可以使用默认值 0，对于限制，我们也可以使用默认值 20。
单击执行按钮，您将看到包含分页产品数据的响应正文。
奖励积分
作为额外的好处，您可以选择探索其余端点并在评论部分分享您对“响应正文”的想法。
您可以通过以下 URL 访问相关的 GitHub 存储库中的项目：
https://github.com/mbuthi/product_listing_API
将项目克隆到本地存储库，然后继续运行它。
结论
总之，本文引导您完成了使用 docker 对 Fast API 应用程序和 PostgreSQL 数据库进行容器化的过程。通过将 API 和数据库捆绑到单独的容器中，我们实现了可移植性和易于部署。
我们首先为 docker 环境设置 dockerfile 和 docker compose 文件，设置模型、模式、CRUD 操作和端点。
在整篇文章中，我们定义了如何使用 docker 中的卷来持久化数据，以及 docker 最佳实践，并强调了 DRY 编程设计。
我希望本文能让您深入了解如何使用 docker 将 Fast API 应用程序和 PostgreSQL 数据库容器化，从而使您的 Web 应用程序更上一层楼。当您继续接触容器化之旅时，请探索 Fast API 和 docker 中的更多高级主题。文章来源：https://www.toymoban.com/diary/apps/345.html
文章来源地址https://www.toymoban.com/diary/apps/345.html
到此这篇关于使用 Fast API 和 PostgreSQL 进行 DevOps：如何使用 Docker 容器化 Fast API 应用程序的文章就介绍到这了,更多相关内容可以在右上角搜索或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！
