FastAPI是一個(gè)開(kāi)源現(xiàn)代框架，用于在 Python 中構(gòu)建 API。
PostgreSQL是一個(gè)開(kāi)源的對(duì)象關(guān)系數(shù)據(jù)庫(kù)管理系統(tǒng)。
在本教程中，我們將使用 Fast API 構(gòu)建示例 RESTful API，并利用 PostgreSQL 持久數(shù)據(jù)的強(qiáng)大功能。然后，我們將使用Dockerfile和Docker Compose文件對(duì) API 和數(shù)據(jù)庫(kù)進(jìn)行容器化。Dockerfile 是一個(gè)文本文件，其中包含將在 Docker Compose 文件中執(zhí)行以構(gòu)建容器的一系列指令。Docker compose是一個(gè)定義和共享多容器Docker容器的工具。我們的應(yīng)用程序?qū)瑑蓚€(gè)容器。Fast API 容器和 PostgreSQL 容器。
前提要求與工具
Docker - 您需要對(duì) Docker 的工作原理有基本的了解。要了解 docker 的工作原理，您可以前往我之前的docker 入門(mén)帖子。您將學(xué)習(xí)如何安裝 docker、docker 的工作原理以及 docker 命令。
Python - 您需要在計(jì)算機(jī)上安裝 Python。最好是Python 3.10。
VSCode
相關(guān)網(wǎng)址
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 入門(mén) - https://dev.to/mbuthi/docker-2oge
Python官網(wǎng) - https://www.python.org/
VsCode官網(wǎng) - https://code.visualstudio.com/download
FastAPI 入門(mén)
我們將使用 Python 構(gòu)建一個(gè)產(chǎn)品列表示例應(yīng)用程序，用戶(hù)將能夠通過(guò) API 執(zhí)行 CRUD 操作。然后我們將使用 PostgreSQL 來(lái)保存產(chǎn)品數(shù)據(jù)。但是，我們需要了解項(xiàng)目目錄結(jié)構(gòu)是什么樣的。下面是 FastAPI 中項(xiàng)目目錄結(jié)構(gòu)的快照：
. └── 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 - 這是我們應(yīng)用程序的根目錄。
app - 保存我們 API 的服務(wù)。
main.py - 這是 API 入口點(diǎn)。
api - 包含 API 端點(diǎn)。
core - 包含核心功能，例如設(shè)置和日志記錄。
crud - 包含 CRUD（創(chuàng)建、讀取、更新、刪除）操作。
db - 包含與數(shù)據(jù)庫(kù)相關(guān)的代碼。
models - 包含數(shù)據(jù)庫(kù)模型。
utils - 包含實(shí)用函數(shù)和類(lèi)。
要開(kāi)始構(gòu)建 API，您需要安裝 Vscode 或您喜歡的 IDE。然后使用上面所示的目錄結(jié)構(gòu)創(chuàng)建一個(gè)新項(xiàng)目。
設(shè)置我們的docker環(huán)境
我們將首先為我們的應(yīng)用程序創(chuàng)建一個(gè) docker compose 文件和一個(gè) docker 文件。
因此，前往 vscode code 打開(kāi)名為Dockerfile的文件并粘貼以下說(shuō)明。
# 使用來(lái)自 docker hub 的 python 官方鏡像 FROM python:3.12-rc-slim-bullseye # 防止 pyc 文件被復(fù)制到容器中 ENV PYTHONDONTWRITEBYTECODE 1 # 確保 python 輸出記錄在容器的終端中 ENV PYTHONUNBUFFERED 1 RUN apt-get update \ # 構(gòu)建 Python 包的依賴(lài)項(xiàng) && apt-get install -y build-essential \ # psycopg2 依賴(lài)項(xiàng) && apt-get install -y libpq-dev \ # Translations 依賴(lài)項(xiàng) && apt-get install -y gettext \ # 清理未使用的文件 && apt-get purge -y --auto-remove -o APT::AutoRemove::RecommendsImportant=false \ && rm -rf /var/lib/apt/lists/* # 將文件“requirements.txt”從本地構(gòu)建上下文復(fù)制到容器的文件系統(tǒng)。 COPY ./requirements.txt /requirements.txt # 安裝python依賴(lài)項(xiàng) RUN pip install -r /requirements.txt # 設(shè)置工作目錄 WORKDIR /app # 運(yùn)行 Uvicorn 來(lái)啟動(dòng)您的 Python Web 應(yīng)用程序 CMD ["uvicorn", "main:app", "--host", "0.0.0.0" "--port" "8000"]
接下來(lái)，我們將創(chuàng)建 docker compose 文件。在 vscode 中打開(kāi)docker-compose.yml文件并粘貼以下說(shuō)明。
# 指定 compose 版本version: '3.8'# 為我們的 docker compose setupservices 指定服務(wù):
api:
build:
context: .
# 指定我們的 dockerfile 路徑。 我們的 Dockerfile 位于根文件夾中
dockerfile: .
# 指定圖像名稱(chēng)
image: products_api
# 該卷用于將主機(jī)上的文件和文件夾映射到容器
# 因此，如果我們更改主機(jī)上的代碼，docker容器中的代碼也會(huì)更改
volumes:
- .:/app
# 將主機(jī)上的8000端口映射到容器上的8000端口
ports:
- 8000:8000
# 指定.env文件路徑
env_file:
- ./env
# 定義對(duì)“products_db”服務(wù)的依賴(lài)，因此它首先啟動(dòng)
depends_on:
- products_db
products_db:
# 指定我們數(shù)據(jù)庫(kù)的圖像名稱(chēng)
# 如果在我們的本地存儲(chǔ)庫(kù)中找不到該圖像
# 將從 Docker Hub 的 docker 注冊(cè)表中拉取
image: 16rc1-alpine3.18
# 安裝卷以保存 postgreSQL 數(shù)據(jù)
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 的環(huán)境變量。
接下來(lái)，我們將創(chuàng)建一個(gè).env文件并實(shí)例化我們的環(huán)境變量。在你的 vscode 中打開(kāi) .env 文件并包含以下變量
# PostgreSQL 數(shù)據(jù)庫(kù)主機(jī) POSTGRES_HOST=products_db # PostgreSQL 數(shù)據(jù)庫(kù)用戶(hù) POSTGRES_USER=username # PostgreSQL 數(shù)據(jù)庫(kù)密碼 POSTGRES_PASSWORD=password # PostgreSQL 數(shù)據(jù)庫(kù)名稱(chēng) POSTGRES_DATABASE=database # PostgreSQL 數(shù)據(jù)庫(kù)端口 POSTGRES_PORT=5432 # 用于連接 PostgreSQL 的異步數(shù)據(jù)庫(kù) URI ASYNC_DATABASE_URI=postgresql+asyncpg://username:password@products_db:5432/database # 項(xiàng)目或應(yīng)用程序的名稱(chēng) PROJECT_NAME=Product Listings
.env 文件包含敏感變量。將這些
敏感變量包含在 .env 文件中始終是一個(gè)好習(xí)慣。
生成唯一 ID。
在我們的 FastAPI 應(yīng)用程序中，我們將定義一個(gè)強(qiáng)大的實(shí)用函數(shù)來(lái)生成唯一的 ID。該函數(shù)將使用 UUID 模塊。轉(zhuǎn)到 utils module/ 文件夾并打開(kāi) idgen.py 文件并粘貼下面的代碼片段。
import uuid def idgen() -> str: # 生成隨機(jī) uuid 字符串 return str(uuid.uuid4().hex)
FastAPI 中的設(shè)置配置
接下來(lái)，我們將創(chuàng)建一個(gè)配置類(lèi)。該類(lèi)將從 pydantic 基設(shè)置類(lèi)繼承。該類(lèi)將負(fù)責(zé)將環(huán)境變量加載到應(yīng)用程序上下文并定義其他應(yīng)用程序設(shè)置。在 vscode 中打開(kāi)名為 settings.py 的文件并插入以下代碼片段。
# 導(dǎo)入包
from pydantic_settings import BaseSettings
import os
from dotenv import load_dotenv
import secrets
load_dotenv()
class Settings(BaseSettings):
"""
應(yīng)用程序設(shè)置和配置參數(shù)
此類(lèi)使用 pydantic 數(shù)據(jù)驗(yàn)證庫(kù)定義應(yīng)用程序設(shè)置
"""
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 中創(chuàng)建我們的模型
我們將從創(chuàng)建一個(gè)基類(lèi)開(kāi)始?；?lèi)將包含所有模型中的公共屬性。這將有助于保持我們的代碼干燥。打開(kāi)位于 models 文件夾中名為 base.py 的文件。在文件內(nèi)粘貼以下代碼片段
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 模型的基類(lèi)，具有保持 DRY（不要重復(fù)）的公共屬性。 此類(lèi)旨在充當(dāng) SQLAlchemy 模型的基類(lèi)。 它定義了常見(jiàn)的屬性，例如表名、創(chuàng)建時(shí)間戳、 并更新可以被其他模型繼承的時(shí)間戳，幫助您 堅(jiān)持 DRY（不要重復(fù)自己）原則。 屬性: __tablename__ (str): 表名，源自小寫(xiě)的類(lèi)名。 id (str): 每條記錄的唯一ID。 created_on (datetime): 創(chuàng)建記錄的時(shí)間戳。 updated_on (datetime, optional): 上次更新記錄的時(shí)間戳。 默認(rèn)為“無(wú)”，直到發(fā)生更新。 示例: 要使用此基類(lèi)創(chuàng)建 SQLAlchemy 模型： class YourModel(Base_): # 在此處為您的模型定義其他屬性。 """ @declared_attr def __tablename__(cls): # 表名源自小寫(xiě)的類(lèi)名 return cls.__name__.lower() # 每條記錄的唯一 UUID ID id: Mapped[str] = mapped_column(primary_key=True, default=idgen,index=True) # 記錄創(chuàng)建的時(shí)間戳 created_on: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now()) # 記錄更新的時(shí)間戳，最初為 None，直到發(fā)生更新 updated_on: Mapped[datetime] = mapped_column(DateTime(timezone=True), onupdate=func.now(), nullable=True)
基類(lèi)包含 id 屬性，它是每個(gè)記錄的唯一 UUID，還包含created_on 屬性，它是記錄創(chuàng)建的時(shí)間戳，還包含updated_on，它是記錄更新的時(shí)間戳。
接下來(lái)，我們將定義我們產(chǎn)品的模型。該模型將從基類(lèi)繼承Base_。打開(kāi)名為 products.py 的文件，該文件包含在名為 models 的文件夾中。在文件內(nèi)，粘貼以下代碼片段。
from sqlalchemy.orm import Mapped, mapped_column from sqlalchemy import String from .base import Base_ class Product(Base_): """ 這是用于定義產(chǎn)品模型的 SQLAlchemy 類(lèi)。 它繼承了Base_類(lèi)的所有屬性和方法。 該類(lèi)定義了常用屬性，例如名稱(chēng)、價(jià)格圖像、和重量。 屬性： name (str): 產(chǎn)品名稱(chēng) price (str): 產(chǎn)品價(jià)格 image (str): 產(chǎn)品圖片 url weight (str): 產(chǎn)品價(jià)格 'nullable=False' 表示這些列在數(shù)據(jù)庫(kù)中不能有 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]只是一個(gè) Python 類(lèi)型的提示。它強(qiáng)調(diào)該屬性將保存字符串類(lèi)型的值。取代mapped_column了之前的 sqlalchemy Column。
在 Fast API 中創(chuàng)建模式。
我們現(xiàn)在將定義我們的pydantic模式。這些模式充當(dāng)數(shù)據(jù)類(lèi)，定義某個(gè) API 端點(diǎn)期望接收哪些數(shù)據(jù)，以便將請(qǐng)求視為有效請(qǐng)求。它們還可以在 Fast API 中使用來(lái)定義響應(yīng)模型，即端點(diǎn)返回的響應(yīng)。打開(kāi)名為product.py 的文件（包含在名為 的文件夾中）schemas并粘貼以下代碼片段。
from typing import Optional from pydantic import BaseModel class ProductBase(BaseModel): name: str # 產(chǎn)品名稱(chēng)（必填） price: str # 產(chǎn)品價(jià)格（必填） image: str # 產(chǎn)品圖片的 URL 或路徑（必填） weight: str # 產(chǎn)品的重量（必填） class ProductCreate(ProductBase): ... class ProductUpdate(ProductBase): ... class ProductPatch(ProductBase): name: Optional[str] # 修補(bǔ)時(shí)名稱(chēng)是可選的 price: Optional[str] # 修補(bǔ)時(shí)價(jià)格是可選的 image: Optional[str] # 修補(bǔ)時(shí)圖像是可選的 weight: Optional[str] # 修補(bǔ)時(shí)權(quán)重是可選的 class Product(ProductBase): id: str class Config: orm_mode = True
Optional從 Python 類(lèi)型模塊導(dǎo)入，該模塊定義該字段不是必需的，因此可以為 None。
在 Fast API 中創(chuàng)建 CRUD 操作
我們現(xiàn)在將定義創(chuàng)建、讀取、更新和刪除方法。首先，我們將為操作創(chuàng)建一個(gè)基類(lèi)?；?lèi)類(lèi)將有助于維護(hù) Python 中的 DRY 代碼設(shè)計(jì)。各種SQLAlchemy模型也會(huì)繼承該類(lèi)來(lái)執(zhí)行數(shù)據(jù)庫(kù)操作。因此，打開(kāi)名為 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（創(chuàng)建、讀取、更新、刪除）操作。
此類(lèi)提供了一組可與 SQLAlchemy 模型一起使用的通用 CRUD 操作。
它包括創(chuàng)建、檢索、更新和刪除數(shù)據(jù)庫(kù)中記錄的方法。
參數(shù):
model (Type[ModelType]): 要執(zhí)行 CRUD 操作的 SQLAlchemy 模型類(lèi)。
例子:
為特定模型（例如用戶(hù)模型）創(chuàng)建 CRUD 實(shí)例：
```
python
crud_user = CRUDBase[Prodcut, ProductCreateSchema, ProductUpdateSchema]
```
"""
def __init__(self, model: Type[ModelType]):
self.model = model
# 獲取單個(gè)實(shí)例
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()
# 獲取所有多個(gè)實(shí)體
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()
# 搜索特定實(shí)體
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()
# 添加實(shí)體
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
# 部分更新實(shí)體
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)
# 完全更新實(shí)體
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
# 從數(shù)據(jù)庫(kù)中完全刪除實(shí)體
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 方法從數(shù)據(jù)庫(kù)中獲取與對(duì)象 ID 匹配的單個(gè)記錄。get_multi 方法從數(shù)據(jù)庫(kù)獲取分頁(yè)文檔。get_by_params 方法根據(jù)匹配的參數(shù)搜索匹配的記錄。get_or_create方法首先檢查實(shí)體是否存在，如果不存在，則在數(shù)據(jù)庫(kù)中創(chuàng)建實(shí)體。patch 方法更新記錄字段。update方法完全更新記錄字段。remove 方法從數(shù)據(jù)庫(kù)中刪除一條記錄。
定義了 CRUD 操作的基類(lèi)后，我們現(xiàn)在將定義產(chǎn)品 CRUD 操作。Product CRUD 操作將從基類(lèi)繼承CRUDBase。打開(kāi) 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)
創(chuàng)建數(shù)據(jù)庫(kù)會(huì)話(huà)
這里我們將定義一個(gè)異步數(shù)據(jù)庫(kù)引擎來(lái)對(duì)數(shù)據(jù)庫(kù)執(zhí)行異步操作。然后我們將引擎綁定到 sessionmaker，它將與數(shù)據(jù)庫(kù)異步交互。打開(kāi)名為 db 的文件夾中包含的名為 session.py 的文件。粘貼下面的代碼片段。
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession from sqlalchemy.orm import sessionmaker from app.core.settings import settings # 使用應(yīng)用程序設(shè)置中的 ASYNC_DATABASE_URI 創(chuàng)建異步 SQLAlchemy 引擎。 engine = create_async_engine( settings.ASYNC_DATABASE_URI, ) # 使用 sessionmaker 創(chuàng)建一個(gè) AsyncSession 類(lèi)，綁定到 SQLAlchemy 引擎。 # 該會(huì)話(huà)類(lèi)將用于與數(shù)據(jù)庫(kù)異步交互。 SessionLocal = sessionmaker( engine, expire_on_commit=False, class_=AsyncSession )
create_async_engine- 這將使用應(yīng)用程序設(shè)置中的 ASYNC_DATABASE_URI 創(chuàng)建異步 SQLAlchemy 引擎。
sessionmaker- 這將使用 sessionmaker 創(chuàng)建一個(gè) AsyncSession 類(lèi)，綁定到 SQLAlchemy 引擎。
創(chuàng)建快速 API 依賴(lài)項(xiàng)
在這里，我們將定義將在我們的應(yīng)用程序中使用的所有依賴(lài)項(xiàng)。這可能包括數(shù)據(jù)庫(kù)會(huì)話(huà)。打開(kāi) 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函數(shù)是一個(gè)異步生成函數(shù)，用于生成數(shù)據(jù)庫(kù)會(huì)話(huà)。
創(chuàng)建產(chǎn)品列表端點(diǎn)
這里我們將定義 POST、GET、PUT、PATCH 和 DELETE 方法。
POST 將創(chuàng)建一個(gè)新產(chǎn)品。
GET 將檢索一個(gè)或多個(gè)產(chǎn)品。
PUT 將完全更新產(chǎn)品。
PATCH 將更新為產(chǎn)品指定的字段。
DELETE 將從數(shù)據(jù)庫(kù)中刪除產(chǎn)品。
轉(zhuǎn)到代碼編輯器并打開(kāi)名為 products.py 的文件，該文件包含在名為 endpoints 的文件夾中。在文件內(nèi)粘貼下面的代碼片段。
# 導(dǎo)入必要的模塊和組件
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
# 創(chuàng)建 APIRouter 實(shí)例
router = APIRouter()
# 定義創(chuàng)建新產(chǎn)品的路線
@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（創(chuàng)建、讀取、更新、刪除）操作
# 創(chuàng)建新產(chǎn)品或返回現(xiàn)有產(chǎn)品（如果已存在）
product, created = await crud.product.get_or_create(
db=db, defaults=product_in.dict()
)
# 如果產(chǎn)品已存在，則引發(fā)帶有 400 狀態(tài)代碼的 HTTPException
if not created:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Product exists"
)
# 返回創(chuàng)建的或現(xiàn)有的產(chǎn)品
return product
# 定義通過(guò) ID 檢索產(chǎn)品的路由
@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 操作通過(guò) ID 檢索產(chǎn)品
product = await crud.product.get(db=db, obj_id=productId)
# 如果產(chǎn)品不存在，則引發(fā)帶有 404 狀態(tài)代碼的 HTTPException
if not product:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Product not found"
)
# 返回檢索到的產(chǎn)品
return product
# 定義一個(gè)路由來(lái)檢索分頁(yè)的產(chǎn)品列表
@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操作檢索多個(gè)帶分頁(yè)的產(chǎn)品
products = await crud.product.get_multi(db=db, skip=skip, limit=limit)
# 如果未找到產(chǎn)品，則引發(fā)帶有 404 狀態(tài)代碼的 HTTPException
if not products:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Products not found"
)
# 返回分頁(yè)的產(chǎn)品列表
return paginate(products)
# 定義部分更新產(chǎn)品的路線
@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 操作通過(guò) ID 檢索產(chǎn)品
product = await crud.product.get(db=db, obj_id=product_Id)
# 如果產(chǎn)品不存在，則引發(fā)帶有 404 狀態(tài)代碼的 HTTPException
if not product:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Product not found"
)
# 使用CRUD操作來(lái)修補(bǔ)（部分更新）產(chǎn)品
product_patched = await crud.product.patch(db=db, obj_id=product_Id, obj_in=product_in.dict())
# 返回已修補(bǔ)的產(chǎn)品
return product_patched
# 定義完全更新產(chǎn)品的路線
@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 操作通過(guò) ID 檢索產(chǎn)品
product = await crud.product.get(db=db, obj_id=productId)
# 如果產(chǎn)品不存在，則引發(fā)帶有 404 狀態(tài)代碼的 HTTPException
if not product:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Product not found"
)
# 使用CRUD操作來(lái)全面更新產(chǎn)品
product_updated = await crud.product.update(
db=db, obj_current=product, obj_new=product_in
)
# 返回更新后的產(chǎn)品
return product_updated
# 定義刪除產(chǎn)品的路線
@router.delete("/{productId}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_product(
db: Annotated[AsyncSession, Depends(get_db)],
productId: str
):
# 使用 CRUD 操作通過(guò) ID 檢索產(chǎn)品
product = await crud.product.get(db=db, obj_id=productId)
# 如果產(chǎn)品不存在，則引發(fā)帶有 404 狀態(tài)代碼的 HTTPException
if not product:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Product not found"
)
# 使用CRUD操作移除（刪除）產(chǎn)品
await crud.product.remove(db=db, obj_id=productId)
# 返回 204 No Content 響應(yīng)表示刪除成功
return端點(diǎn)由注釋組成，解釋每個(gè)端點(diǎn)中發(fā)生的情況。
現(xiàn)在我們需要將端點(diǎn)公開(kāi)給 API 入口點(diǎn)，以便我們編輯兩個(gè)文件。對(duì)于第一個(gè)文件，我們將打開(kāi)名為 v1 的文件夾內(nèi)的 api.py 文件。然后粘貼下面的代碼片段。
# 從 FastAPI 導(dǎo)入 APIRouter 類(lèi) from fastapi import APIRouter # 從“app.api.v1.endpoints”模塊導(dǎo)入“products”路由器 from app.api.v1.endpoints import products # 創(chuàng)建 APIRouter 的實(shí)例 router = APIRouter() # 將“products”路由器作為子路由器包含在“/products”前綴下 # 并分配標(biāo)簽“Products”以對(duì)相關(guān) API 端點(diǎn)進(jìn)行分組 router.include_router(products.router, prefix="/products", tags=["Products"])
然后打開(kāi) main.py 文件并粘貼下面的代碼片段。
# 從FastAPI框架導(dǎo)入FastAPI類(lèi)
from fastapi import FastAPI
# 導(dǎo)入add_pagination
from fastapi_pagination import add_pagination
# 從“app.api.v1.api”模塊導(dǎo)入“路由器”
from app.api.v1.api import router
#從“app.core.settings”模塊導(dǎo)入“settings”對(duì)象
from app.core.settings import settings
# 創(chuàng)建 FastAPI 應(yīng)用程序的實(shí)例
# - “title”設(shè)置為“settings”中的項(xiàng)目名稱(chēng)
# - 'openapi_url' 指定 OpenAPI 文檔的 URL
app = FastAPI(
title=settings.PROJECT_NAME,
openapi_url=f"{settings.API_V1_STR}/openapi.json"
)
# 為所有使用 paginate 的路由添加必要的分頁(yè)參數(shù)
add_pagination(app)
# 在 FastAPI 應(yīng)用程序中包含“路由器”（其中包含您的 API 路由）
app.include_router(router)到目前為止，我們可以嘗試啟動(dòng)我們的服務(wù)器。為此，我們必須使用 docker 容器和 docker 鏡像構(gòu)建我們的應(yīng)用程序。
運(yùn)行產(chǎn)品列表 API
在這里我們將嘗試運(yùn)行我們的 API。
假設(shè)您本地計(jì)算機(jī)上安裝了docker，請(qǐng)打開(kāi) vscode 終端。
要打開(kāi)終端：
Windows 使用快捷鍵ctrl + `。
Mac OS 使用快捷鍵 ? +`。
Linux 使用快捷鍵Ctrl+Shift+`。
在終端中寫(xiě)入以下命令：
docker-compose -f docker-compose.yml up -d
docker-compose- 此命令用于使用 Docker compose 管理 Docker 容器。
-f - 用于指定撰寫(xiě)文件的路徑。
docker-compose.yml - 這是定義容器的 compose 文件的路徑。在我們的例子中是 docker-compose.yml。
up - 用于初始化和啟動(dòng) compose 文件中指定的服務(wù)。在我們的例子中，它啟動(dòng)products_db和api服務(wù)。
-d - 這指定容器應(yīng)以分離模式啟動(dòng)，即容器作為后臺(tái)服務(wù)啟動(dòng)。
成功執(zhí)行命令后，您可以通過(guò)在 vscode 終端中執(zhí)行以下命令來(lái)驗(yàn)證容器確實(shí)已啟動(dòng)并正在運(yùn)行：
docker ps
您應(yīng)該能夠看到以下輸出：
要通過(guò) Swagger 查看 API 文檔，您可以打開(kāi)您的首選瀏覽器并粘貼以下 URL：
http://localhost:8000/docs
默認(rèn)情況下，我們將通過(guò)端口 8000 訪問(wèn) API，因?yàn)檫@是我們之前在 docker compose 文件中指定的映射到主機(jī)的端口。
在您的瀏覽器中，您將能夠看到類(lèi)似以下內(nèi)容：
我們現(xiàn)在已經(jīng)成功設(shè)置了用于產(chǎn)品列表的 API。但是，如果我們嘗試在 Swagger 中執(zhí)行 POST 請(qǐng)求，我們將收到 500 內(nèi)部服務(wù)器錯(cuò)誤。
要查看導(dǎo)致錯(cuò)誤的原因，我們將查看api容器日志。要查看日志，我們可以使用docker 桌面或使用我們的終端來(lái)查看日志。為此，我們將在 vscode 終端中執(zhí)行以下命令：
docker logs <CONTAINER ID>
是當(dāng)前正在運(yùn)行的容器CONTAINER ID的 ID.api
為了獲得CONTAINER ID我們將運(yùn)行：
docker ps
成功運(yùn)行docker logs命令后，我們會(huì)得到如下錯(cuò)誤，如下圖所示：
在最后一行，我們可以清楚地看到日志表明"database" does not exist. 之前，我們已經(jīng)在**.env**我們的 POSTGRES_DATABASE=database 中定義了。而這個(gè)名為database的數(shù)據(jù)庫(kù)不存在。這意味著我們實(shí)際上必須首先創(chuàng)建數(shù)據(jù)庫(kù)本身。
為了創(chuàng)建數(shù)據(jù)庫(kù)，我們將使用products_db容器。
在你的 vscode 終端中：
運(yùn)行下面的命令
docker exec -it <CONTAINER ID> /bin/bash
上面的命令在容器內(nèi)啟動(dòng) Bash 終端。
運(yùn)行docker ps以獲取products_dbID 并將其替換CONTAINER ID為您的映像實(shí)例的 ID products_db。
我們需要?jiǎng)?chuàng)建數(shù)據(jù)庫(kù)。為此，我們將在容器 Bash 終端中運(yùn)行以下一系列命令：
psql -U username
上述命令啟動(dòng) PostgreSQL 的基于終端的前端。它允許我們交互式地輸入查詢(xún)。
CREATE DATABASE database;
上面的命令在 PostgreSQL 中創(chuàng)建一個(gè)名為database的數(shù)據(jù)庫(kù)。
ALTER ROLE username WITH PASSWORD 'password';
上述命令更改角色用戶(hù)名并為其分配密碼password。
GRANT ALL PRIVILEGES ON DATABASE database TO username;
上述命令將所有數(shù)據(jù)庫(kù)權(quán)限授予名為 username 的用戶(hù)。
完成此操作后，我們現(xiàn)在需要執(zhí)行數(shù)據(jù)庫(kù)遷移。
快速 API 數(shù)據(jù)庫(kù)遷移
數(shù)據(jù)庫(kù)遷移或模式遷移是為修改關(guān)系數(shù)據(jù)庫(kù)中對(duì)象的結(jié)構(gòu)而開(kāi)發(fā)的受控更改集。
為了在我們的 API 中執(zhí)行遷移，我們將在項(xiàng)目根目錄中創(chuàng)建一個(gè) alembic.ini 文件和一個(gè) alembic 文件夾。在 alembic 文件夾內(nèi)創(chuàng)建另一個(gè)名為 versions 的文件夾以及兩個(gè)名為 env.py 和 script.py.mako 的文件。
現(xiàn)在項(xiàng)目目錄結(jié)構(gòu)如下所示：
. └── 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
我們現(xiàn)在將編輯已添加的文件。
打開(kāi) alembic.ini 文件并粘貼以下腳本：
# 通用的單一數(shù)據(jù)庫(kù)配置。 [alembic] # 遷移腳本的路徑 script_location = alembic # 用于生成遷移文件名的模板； 默認(rèn)值為 %%(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 一起使用的配置文件，它提供用于管理數(shù)據(jù)庫(kù)架構(gòu)隨時(shí)間變化的設(shè)置和選項(xiàng)。
打開(kāi) 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 中作為異步數(shù)據(jù)庫(kù)引擎運(yùn)行數(shù)據(jù)庫(kù)遷移。
打開(kāi) 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.
"""定義了用于處理遷移的腳本后，我們現(xiàn)在可以在 api 容器中執(zhí)行它們。為此，運(yùn)行以下命令：
docker exec -it <CONTAINER ID> /bin/bash
上面的命令在容器內(nèi)啟動(dòng) Bash 終端。
將 替換為 api 容器的實(shí)際 ID。要獲取 api 容器，請(qǐng)運(yùn)行docker ps命令。
alembic revision --autogenerate -m "Migrate products table"
上述命令生成一個(gè)新的遷移腳本。新的遷移腳本包含當(dāng)前數(shù)據(jù)庫(kù)架構(gòu)與代碼中的模型定義之間的差異。
alembic upgrade head
上述命令適用于所有待處理的遷移。
測(cè)試我們的 API
由于我們已經(jīng)執(zhí)行了數(shù)據(jù)庫(kù)模式的遷移，因此我們現(xiàn)在可以通過(guò) swagger 文檔自信地測(cè)試我們的 API。
要訪問(wèn) Swagger 文檔，請(qǐng)?jiān)跒g覽器中輸入以下 URL：
http://localhost:8000/docs
我們可以通過(guò)執(zhí)行 POST 請(qǐng)求開(kāi)始。
POST 請(qǐng)求快速 API
在 Swagger 中，展開(kāi)可折疊的 POST 請(qǐng)求，然后單擊Try it out按鈕。在“響應(yīng)正文”部分中，將 JSON 架構(gòu)的鍵值更改為您的首選項(xiàng)，如下所示。
對(duì)于圖像鍵，您可以輸入圖像 URL。然后單擊執(zhí)行按鈕。成功 POST 后，您將看到 201 創(chuàng)建的狀態(tài)代碼以及響應(yīng)正文，如下所示：
根據(jù)您分配給 JSON 架構(gòu)的值，響應(yīng)正文可能與上面顯示的有所不同。
GET 請(qǐng)求（分頁(yè)數(shù)據(jù)）
在 GET 請(qǐng)求中，我們想要獲取多個(gè)項(xiàng)目。為此，我們可以指定跳過(guò)和限制。
Skip 與 OFFSET 類(lèi)似，是在檢索任何行之前要跳過(guò)的結(jié)果表的行數(shù)。
Limit 是指定獲取結(jié)果表的前 N 行的語(yǔ)法。
單擊獲取請(qǐng)求，對(duì)于跳過(guò)參數(shù)，我們可以使用默認(rèn)值 0，對(duì)于限制，我們也可以使用默認(rèn)值 20。
單擊執(zhí)行按鈕，您將看到包含分頁(yè)產(chǎn)品數(shù)據(jù)的響應(yīng)正文。
獎(jiǎng)勵(lì)積分
作為額外的好處，您可以選擇探索其余端點(diǎn)并在評(píng)論部分分享您對(duì)“響應(yīng)正文”的想法。
您可以通過(guò)以下 URL 訪問(wèn)相關(guān)的 GitHub 存儲(chǔ)庫(kù)中的項(xiàng)目：
https://github.com/mbuthi/product_listing_API
將項(xiàng)目克隆到本地存儲(chǔ)庫(kù)，然后繼續(xù)運(yùn)行它。
結(jié)論
總之，本文引導(dǎo)您完成了使用 docker 對(duì) Fast API 應(yīng)用程序和 PostgreSQL 數(shù)據(jù)庫(kù)進(jìn)行容器化的過(guò)程。通過(guò)將 API 和數(shù)據(jù)庫(kù)捆綁到單獨(dú)的容器中，我們實(shí)現(xiàn)了可移植性和易于部署。
我們首先為 docker 環(huán)境設(shè)置 dockerfile 和 docker compose 文件，設(shè)置模型、模式、CRUD 操作和端點(diǎn)。
在整篇文章中，我們定義了如何使用 docker 中的卷來(lái)持久化數(shù)據(jù)，以及 docker 最佳實(shí)踐，并強(qiáng)調(diào)了 DRY 編程設(shè)計(jì)。
我希望本文能讓您深入了解如何使用 docker 將 Fast API 應(yīng)用程序和 PostgreSQL 數(shù)據(jù)庫(kù)容器化，從而使您的 Web 應(yīng)用程序更上一層樓。當(dāng)您繼續(xù)接觸容器化之旅時(shí)，請(qǐng)?zhí)剿?Fast API 和 docker 中的更多高級(jí)主題。文章來(lái)源：http://www.zghlxwxcb.cn/article/345.html
文章來(lái)源地址http://www.zghlxwxcb.cn/article/345.html
到此這篇關(guān)于使用 Fast API 和 PostgreSQL 進(jìn)行 DevOps：如何使用 Docker 容器化 Fast API 應(yīng)用程序的文章就介紹到這了,更多相關(guān)內(nèi)容可以在右上角搜索或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！
