使用 FastAPI 在 Python 中实现微服务 使用 Python 创建微服务 wemake-python-styleguide ~/python-microservices/movie-service/app/api/db.py

2025-05-26

使用 FastAPI 在 Python 中进行微服务

使用 Python 创建微服务

wemake-python-样式指南

〜/ python-microservices / movie-service / app / api / db.py

使用 Python 创建微服务

作为一名 Python 开发者，您可能听说过“微服务”这个术语，并且想要自己构建一个 Python 微服务。微服务是构建高度可扩展应用程序的绝佳架构。在开始使用微服务构建应用程序之前，您必须熟悉使用微服务的优缺点。在本文中，您将了解使用微服务的优缺点。您还将学习如何构建自己的微服务并使用 Docker Compose 进行部署。

在本教程中，您将学习：

  • 微服务的优点和缺点是什么
  • 为什么应该使用 Python 构建微服务
  • 如何使用 FastAPI 和 PostgreSQL 构建 REST API
  • 如何使用 FastAPI 构建微服务
  • 如何使用docker-compose
  • 如何使用 Nginx 管理微服务

您将首先使用 FastAPI 构建一个简单的 REST API，然后使用 PostgreSQL 作为数据库。之后，您将把同一个应用程序扩展为微服务。

微服务简介

微服务是一种将大型单体应用分解成多个独立应用，每个应用负责特定的服务/功能的方法。这种方法通常被称为面向服务的架构（SOA）。

在单体架构中，所有业务逻辑都驻留在同一个应用程序中。用户管理、身份验证等应用服务使用同一个数据库。

在微服务架构中，应用程序被分解为多个独立的服务，这些服务在不同的进程中运行。应用程序的不同功能使用不同的数据库，并且服务之间使用 HTTP、AMQP 或 TCP 等二进制协议进行通信，具体取决于每个服务的性质。服务间通信也可以使用RabbitMQ、Kafka或Redis等消息队列进行。

微服务的好处

微服务架构有很多好处。其中一些好处是：

  • 松耦合的应用意味着不同的服务可以使用最适合它们的技术来构建。因此，开发团队不必受制于项目启动时所做的选择。

  • 由于服务负责特定的功能，这使得应用程序更容易理解和控制。

  • 应用程序扩展也变得更加容易，因为如果其中一项服务需要高 GPU 使用率，那么只有包含该服务的服务器需要具有高 GPU，而其他服务器可以在普通服务器上运行。

微服务的缺点

微服务架构并非万能的灵丹妙药，它也有其缺点。其中一些缺点如下：

  • 由于不同的服务使用不同的数据库，涉及多个服务的事务需要使用最终一致性。

  • 第一次尝试时很难实现完美的服务拆分，需要反复尝试才能实现最佳的服务分离。

  • 由于服务之间通过网络交互进行通信，这会因网络延迟和服务速度慢而导致应用程序速度变慢。

为什么使用 Python 进行微服务

Python 是构建微服务的理想工具，因为它拥有强大的社区、简单的学习曲线和丰富的库。由于 Python 引入了异步编程，性能堪比 GO 和 Node.js 的Web 框架应运而生。

FastAPI 简介

FastAPI 是一个现代化的高性能 Web 框架，它拥有众多酷炫功能，例如基于 OpenAPI 的自动文档生成以及内置的序列化和验证库。FastAPI的所有酷炫功能列表请参阅此处。

为什么选择 FastAPI

我认为 FastAPI 是用 Python 构建微服务的绝佳选择的原因如下：

  • 自动文档
  • Async/Await 支持
  • 内置验证和序列化
  • 100% 类型注释，自动完成功能良好

安装 FastAPI

在安装 FastAPI 之前，请创建一个新目录，并使用virtualenvmovie_service在新创建的目录中创建一个新的虚拟环境。 如果您尚未安装：
virtualenv

pip install virtualenv
Enter fullscreen mode Exit fullscreen mode

现在，创建一个新的虚拟环境。

virtualenv env
Enter fullscreen mode Exit fullscreen mode

如果您使用的是 Mac/Linux，则可以使用以下命令激活虚拟环境：

source ./env/bin/activate
Enter fullscreen mode Exit fullscreen mode

Windows 用户可以运行此命令：

.\env\Scripts\activate
Enter fullscreen mode Exit fullscreen mode

最后，您可以安装 FastAPI 了，运行以下命令：

pip install fastapi
Enter fullscreen mode Exit fullscreen mode

由于 FastAPI 不自带服务，uvicorn因此需要安装才能运行。uvicorn它是一个ASGI服务器，允许我们使用 async/await 功能。使用以下命令
安装uvicorn

pip install uvicorn
Enter fullscreen mode Exit fullscreen mode

使用 FastAPI 创建简单的 REST API

在开始使用 FastAPI 构建微服务之前，让我们先学习一下 FastAPI 的基础知识。在新建的目录中创建一个新目录app和一个新文件。main.py

在中添加以下代码main.py。

#~/movie_service/app/main.py

from fastapi import FastAPI

app = FastAPI()


@app.get('/')
async def index():
    return {"Real": "Python"}
Enter fullscreen mode Exit fullscreen mode

在这里，您首先导入并实例化 FastAPI，然后注册根端点/，然后返回一个JSON。

您可以使用 运行应用程序服务器uvicorn app.main:app --reload。这里app.main表示您使用目录main.py内的文件app，:app表示我们的FastAPI实例名称。

您可以从http://127.0.0.1:8000访问该应用程序。要访问酷炫的自动文档，请转到http://127.0.0.1:8000/docs。您可以在浏览器中试用并与 API 进行交互。

让我们为应用程序添加一些 CRUD 功能。
更新您的main.py代码，使其如下所示：

#~/movie_service/app/main.py

from fastapi import FastAPI
from pydantic import BaseModel
from typing import List

app = FastAPI()

fake_movie_db = [
    {
        'name': 'Star Wars: Episode IX - The Rise of Skywalker',
        'plot': 'The surviving members of the resistance face the First Order once again.',
        'genres': ['Action', 'Adventure', 'Fantasy'],
        'casts': ['Daisy Ridley', 'Adam Driver']
    }
]

class Movie(BaseModel):
    name: str
    plot: str
    genres: List[str]
    casts: List[str]


@app.get('/', response_model=List[Movie])
async def index():
    return fake_movie_db
Enter fullscreen mode Exit fullscreen mode

如您所见，您创建了一个从 pydanticMovie扩展的新类。 该模型包含名称、照片、类型和演员阵容。Pydantic内置了 FastAPI，这使得创建模型和请求验证变得轻而易举。BaseModel
Movie

如果你去文档网站看看，你会发现 Movies 模型中的一些字段已经在示例响应部分中提到了。这是因为你已经response_model在路由定义中定义了它们。

现在，让我们添加端点以将电影添加到我们的电影列表中。

添加新的端点定义来处理POST请求。

@app.post('/', status_code=201)
async def add_movie(payload: Movie):
    movie = payload.dict()
    fake_movie_db.append(movie)
    return {'id': len(fake_movie_db) - 1}
Enter fullscreen mode Exit fullscreen mode

现在，前往浏览器测试新的 API。尝试添加包含无效字段或缺少必填字段的电影，看看 FastAPI 是否自动处理了验证。

让我们添加一个新的端点来更新电影。

@app.put('/{id}')
async def update_movie(id: int, payload: Movie):
    movie = payload.dict()
    movies_length = len(fake_movie_db)
    if 0 <= id <= movies_length:
        fake_movie_db[id] = movie
        return None
    raise HTTPException(status_code=404, detail="Movie with given id not found")
Enter fullscreen mode Exit fullscreen mode

这id是我们的列表索引fake_movie_db。

注意：请记住HTTPException从fastapi

现在您还可以添加端点来删除电影。

@app.delete('/{id}')
async def delete_movie(id: int):
    movies_length = len(fake_movie_db)
    if 0 <= id <= movies_length:
        del fake_movie_db[id]
        return None
    raise HTTPException(status_code=404, detail="Movie with given id not found")
Enter fullscreen mode Exit fullscreen mode

在继续下一步之前，让我们先以更好的方式构建我们的应用程序。api在 中创建一个新文件夹app，并在新创建的文件夹中创建一个新文件movies.py。将所有与路由相关的代码从 移动main.py到movies.py。因此，movies.py应该如下所示：

#~/movie-service/app/api/movies.py

from typing import List
from fastapi import Header, APIRouter

from app.api.models import Movie

fake_movie_db = [
    {
        'name': 'Star Wars: Episode IX - The Rise of Skywalker',
        'plot': 'The surviving members of the resistance face the First Order once again.',
        'genres': ['Action', 'Adventure', 'Fantasy'],
        'casts': ['Daisy Ridley', 'Adam Driver']
    }
]

movies = APIRouter()

@movies.get('/', response_model=List[Movie])
async def index():
    return fake_movie_db

@movies.post('/', status_code=201)
async def add_movie(payload: Movie):
    movie = payload.dict()
    fake_movie_db.append(movie)
    return {'id': len(fake_movie_db) - 1}

@movies.put('/{id}')
async def update_movie(id: int, payload: Movie):
    movie = payload.dict()
    movies_length = len(fake_movie_db)
    if 0 <= id <= movies_length:
        fake_movie_db[id] = movie
        return None
    raise HTTPException(status_code=404, detail="Movie with given id not found")

@movies.delete('/{id}')
async def delete_movie(id: int):
    movies_length = len(fake_movie_db)
    if 0 <= id <= movies_length:
        del fake_movie_db[id]
        return None
    raise HTTPException(status_code=404, detail="Movie with given id not found")
Enter fullscreen mode Exit fullscreen mode

在这里，您使用FastAPI 的APIRouter注册了一个新的 API 路由。

models.py另外，在里面创建一个新文件，api用于保存我们的 Pydantic 模型。

#~/movie-service/api/models.py

from typing import List
from pydantic import BaseModel

class Movie(BaseModel):
    name: str
    plot: str
    genres: List[str]
    casts: List[str]
Enter fullscreen mode Exit fullscreen mode

现在注册这个新的路线文件main.py

#~/movie-service/app/main.py
from fastapi import FastAPI

from app.api.movies import movies

app = FastAPI()

app.include_router(movies)
Enter fullscreen mode Exit fullscreen mode

最后，我们的应用程序目录结构如下所示：

movie-service
├── app
│   ├── api
│   │   ├── models.py
│   │   ├── movies.py
│   |── main.py
└── env
Enter fullscreen mode Exit fullscreen mode

在继续下一步之前，请确保您的应用程序正常运行。

通过 FastAPI 使用 PostgreSQL 数据库

之前，你使用了伪造的 Python 列表来添加电影，但现在你终于可以使用真正的数据库来实现此目的了。你将使用PostgreSQL来实现此目的。如果你还没有安装PostgreSQL，请先安装它。安装 PostgreSQL 后，创建一个新的数据库，我将命名为movie_db。

您将使用encode/databases连接到数据库，并使用async和支持。点击此处await了解更多关于async/awaitPython 的使用方法

使用以下方法安装所需的库：

pip install 'databases[postgresql]'
Enter fullscreen mode Exit fullscreen mode

这也将安装sqlalchemy和asyncpg，它们是使用 PostgreSQL 所必需的。

在里面创建一个新文件api并将其命名为db.py。此文件将包含我们 REST API 的实际数据库模型。

#~/movie-service/app/api/db.py

from sqlalchemy import (Column, Integer, MetaData, String, Table,
                        create_engine, ARRAY)

from databases import Database

DATABASE_URL = 'postgresql://movie_user:movie_password@localhost/movie_db'

engine = create_engine(DATABASE_URL)
metadata = MetaData()

movies = Table(
    'movies',
    metadata,
    Column('id', Integer, primary_key=True),
    Column('name', String(50)),
    Column('plot', String(250)),
    Column('genres', ARRAY(String)),
    Column('casts', ARRAY(String))
)

database = Database(DATABASE_URL)
Enter fullscreen mode Exit fullscreen mode

这里，DATABASE_URI是用于连接 PostgreSQL 数据库的 URL。这里movie_user是数据库用户名，movie_password是数据库用户的密码，movie_db是数据库名称。

就像在SQLAlchemy中一样，您已经为电影数据库创建了表。

更新main.py以连接到数据库。main.py应该如下所示：

#~/movie-service/app/main.py

from fastapi import FastAPI
from app.api.movies import movies
from app.api.db import metadata, database, engine

metadata.create_all(engine)

app = FastAPI()

@app.on_event("startup")
async def startup():
    await database.connect()

@app.on_event("shutdown")
async def shutdown():
    await database.disconnect()


app.include_router(movies)
Enter fullscreen mode Exit fullscreen mode

FastAPI 提供了一些事件处理程序，您可以在应用程序启动时使用这些事件处理程序连接到我们的数据库，并在应用程序关闭时断开连接。

更新movies.py以便它使用数据库而不是假的 Python 列表。

#~/movie-service/app/api/movies.py


from typing import List
from fastapi import Header, APIRouter

from app.api.models import MovieIn, MovieOut
from app.api import db_manager

movies = APIRouter()

@movies.get('/', response_model=List[MovieOut])
async def index():
    return await db_manager.get_all_movies()

@movies.post('/', status_code=201)
async def add_movie(payload: MovieIn):
    movie_id = await db_manager.add_movie(payload)
    response = {
        'id': movie_id,
        **payload.dict()
    }

    return response

@movies.put('/{id}')
async def update_movie(id: int, payload: MovieIn):
    movie = payload.dict()
    fake_movie_db[id] = movie
    return None

@movies.put('/{id}')
async def update_movie(id: int, payload: MovieIn):
    movie = await db_manager.get_movie(id)
    if not movie:
        raise HTTPException(status_code=404, detail="Movie not found")

    update_data = payload.dict(exclude_unset=True)
    movie_in_db = MovieIn(**movie)

    updated_movie = movie_in_db.copy(update=update_data)

    return await db_manager.update_movie(id, updated_movie)

@movies.delete('/{id}')
async def delete_movie(id: int):
    movie = await db_manager.get_movie(id)
    if not movie:
        raise HTTPException(status_code=404, detail="Movie not found")
    return await db_manager.delete_movie(id)
Enter fullscreen mode Exit fullscreen mode

让我们添加db_manager.py操作数据库的功能。

#~/movie-service/app/api/db_manager.py

from app.api.models import MovieIn, MovieOut, MovieUpdate
from app.api.db import movies, database


async def add_movie(payload: MovieIn):
    query = movies.insert().values(**payload.dict())

    return await database.execute(query=query)

async def get_all_movies():
    query = movies.select()
    return await database.fetch_all(query=query)

async def get_movie(id):
    query = movies.select(movies.c.id==id)
    return await database.fetch_one(query=query)

async def delete_movie(id: int):
    query = movies.delete().where(movies.c.id==id)
    return await database.execute(query=query)

async def update_movie(id: int, payload: MovieIn):
    query = (
        movies
        .update()
        .where(movies.c.id == id)
        .values(**payload.dict())
    )
    return await database.execute(query=query)
Enter fullscreen mode Exit fullscreen mode

让我们更新一下，models.py以便您可以将 Pydantic 模型与 sqlalchemy 表一起使用。

#~/movie-service/app/api/models.py

from pydantic import BaseModel
from typing import List, Optional

class MovieIn(BaseModel):
    name: str
    plot: str
    genres: List[str]
    casts: List[str]


class MovieOut(MovieIn):
    id: int


class MovieUpdate(MovieIn):
    name: Optional[str] = None
    plot: Optional[str] = None
    genres: Optional[List[str]] = None
    casts: Optional[List[str]] = None
Enter fullscreen mode Exit fullscreen mode

这MovieIn是用于将电影添加到数据库的基础模型。id从数据库获取电影数据时，必须将 添加到此模型中，因此需要使用MovieOutmodel.model 。MovieUpdate该模型允许我们将模型中的值设置为可选，以便在更新电影时仅发送需要更新的字段。

现在，请转到浏览器文档站点并开始使用 API。

微服务数据管理模式

管理微服务中的数据是构建微服务最具挑战性的方面之一。由于应用程序的不同功能由不同的服务处理，因此数据库的使用可能会很棘手。

以下是一些可用于管理应用程序中的数据流的模式。

每个服务的数据库

如果您希望微服务尽可能松耦合，那么为每个服务使用一个数据库是一个不错的选择。为每个服务使用不同的数据库使我们能够独立地扩展不同的服务。涉及多个数据库的事务可以通过定义明确的 API 完成。但这也有一个缺点，因为实现涉及多个服务的业务事务并不简单。此外，网络开销的增加也会降低其使用效率。

共享数据库

如果涉及多个服务的事务很多，最好使用共享数据库。共享数据库虽然能带来应用程序高度一致的好处，但却会抵消微服务架构的大部分优势。负责某个服务的开发人员需要协调其他服务中的模式变更。

API 组成

在涉及多个数据库的事务中，API Composer 充当 API 网关，并按所需顺序执行对其他微服务的 API 调用。最后，执行内存连接后，每个微服务的结果将返回给客户端服务。这种方法的缺点是大型数据集的内存连接效率低下。

在 Docker 中创建 Python 微服务

使用Docker可以大大减少部署微服务的痛苦。Docker有助于封装每个服务并独立地扩展它们。

安装 Docker 和 Docker Compose

如果您尚未在系统中安装 docker，请运行以下命令验证 docker 是否已安装docker。安装完 Docker 后，请安装 Docker Compose。Docker Compose 用于定义和运行多个 Docker 容器，并有助于容器之间轻松交互。

创建电影服务

由于构建电影服务的大部分工作在开始使用 FastAPI 时已经完成，因此您将重用已编写的代码。创建一个全新的文件夹，我将命名为python-microservices。将您之前编写的代码（我将其命名为 ）移入其中movie-service。
因此，文件夹结构将如下所示：

python-microservices/
└── movie-service/
    ├── app/
    └── env/
Enter fullscreen mode Exit fullscreen mode

首先，让我们创建一个requirements.txt文件，用于保存所有需要用到的依赖项movie-service。在里面
创建一个新文件，并添加以下内容：requirements.txtmovie-service

asyncpg==0.20.1
databases[postgresql]==0.2.6
fastapi==0.48.0
SQLAlchemy==1.3.13
uvicorn==0.11.2
httpx==0.11.1
Enter fullscreen mode Exit fullscreen mode

您已经使用了那里提到的所有库，除了在进行服务到服务 API 调用时要使用的httpx 。

创建一个Dockerfile内部文件movie-service，内容如下：

FROM python:3.8-slim

WORKDIR /app

COPY ./requirements.txt /app/requirements.txt

RUN apt-get update \
    && apt-get install gcc -y \
    && apt-get clean

RUN pip install -r /app/requirements.txt \
    && rm -rf /root/.cache/pip

COPY . /app/
Enter fullscreen mode Exit fullscreen mode

首先，您需要定义要使用的 Python 版本。然后将 设置WORKDIR为appDocker 容器内文件夹。之后，gcc安装应用程序中使用的库所需的 。
最后，安装所有依赖项requirements.txt，并复制其中的所有文件movie-service/app。

更新db.py和替换

DATABASE_URI = 'postgresql://movie_user:movie_password@localhost/movie_db'
Enter fullscreen mode Exit fullscreen mode

和

DATABASE_URI = os.getenv('DATABASE_URI')
Enter fullscreen mode Exit fullscreen mode

注意：不要忘记os在文件顶部导入。

您需要这样做，以便稍后可以将其DATABASE_URI作为环境变量提供。

此外，更新main.py和替换

app.include_router(movies)
Enter fullscreen mode Exit fullscreen mode

和

app.include_router(movies, prefix='/api/v1/movies', tags=['movies'])
Enter fullscreen mode Exit fullscreen mode

如此一来，管理不同版本的 API 就变得更容易了。此外，标签还能让您在 FastAPI 文档中更轻松地prefix /api/v1/movies查找相关 API 。movies

另外，你需要更新我们的模型，以便casts存储演员的 ID，而不是实际姓名。因此，请将 更新models.py为如下所示：

#~/python-microservices/movie-service/app/api/models.py

from pydantic import BaseModel
from typing import List, Optional

class MovieIn(BaseModel):
    name: str
    plot: str
    genres: List[str]
    casts_id: List[int]


class MovieOut(MovieIn):
    id: int


class MovieUpdate(MovieIn):
    name: Optional[str] = None
    plot: Optional[str] = None
    genres: Optional[List[str]] = None
    casts_id: Optional[List[int]] = None
Enter fullscreen mode Exit fullscreen mode

同样，您需要更新数据库表，让我们更新db.py：

#~/python-microservices/movie-service/app/api/db.py

import os

from sqlalchemy import (Column, DateTime, Integer, MetaData, String, Table,
                        create_engine, ARRAY)

from databases import Database

DATABASE_URL = os.getenv('DATABASE_URL')

engine = create_engine(DATABASE_URL)
metadata = MetaData()

movies = Table(
    'movies',
    metadata,
    Column('id', Integer, primary_key=True),
    Column('name', String(50)),
    Column('plot', String(250)),
    Column('genres', ARRAY(String)),
    Column('casts_id', ARRAY(Integer))
)

database = Database(DATABASE_URL)
Enter fullscreen mode Exit fullscreen mode

现在，movies.py在添加新电影或更新电影之前，进行更新以检查具有给定 ID 的演员是否存在于演员服务中。

#~/python-microservices/movie-service/app/api/movies.py

from typing import List
from fastapi import APIRouter, HTTPException

from app.api.models import MovieOut, MovieIn, MovieUpdate
from app.api import db_manager
from app.api.service import is_cast_present

movies = APIRouter()

@movies.post('/', response_model=MovieOut, status_code=201)
async def create_movie(payload: MovieIn):
    for cast_id in payload.casts_id:
        if not is_cast_present(cast_id):
            raise HTTPException(status_code=404, detail=f"Cast with id:{cast_id} not found")

    movie_id = await db_manager.add_movie(payload)
    response = {
        'id': movie_id,
        **payload.dict()
    }

    return response

@movies.get('/', response_model=List[MovieOut])
async def get_movies():
    return await db_manager.get_all_movies()

@movies.get('/{id}/', response_model=MovieOut)
async def get_movie(id: int):
    movie = await db_manager.get_movie(id)
    if not movie:
        raise HTTPException(status_code=404, detail="Movie not found")
    return movie

@movies.put('/{id}/', response_model=MovieOut)
async def update_movie(id: int, payload: MovieUpdate):
    movie = await db_manager.get_movie(id)
    if not movie:
        raise HTTPException(status_code=404, detail="Movie not found")

    update_data = payload.dict(exclude_unset=True)

    if 'casts_id' in update_data:
        for cast_id in payload.casts_id:
            if not is_cast_present(cast_id):
                raise HTTPException(status_code=404, detail=f"Cast with given id:{cast_id} not found")

    movie_in_db = MovieIn(**movie)

    updated_movie = movie_in_db.copy(update=update_data)

    return await db_manager.update_movie(id, updated_movie)

@movies.delete('/{id}', response_model=None)
async def delete_movie(id: int):
    movie = await db_manager.get_movie(id)
    if not movie:
        raise HTTPException(status_code=404, detail="Movie not found")
    return await db_manager.delete_movie(id)
Enter fullscreen mode Exit fullscreen mode

让我们添加一个服务来对转换服务进行 API 调用：

#~/python-microservices/movie-service/app/api/service.py

import os
import httpx

CAST_SERVICE_HOST_URL = 'http://localhost:8002/api/v1/casts/'
url = os.environ.get('CAST_SERVICE_HOST_URL') or CAST_SERVICE_HOST_URL

def is_cast_present(cast_id: int):
    r = httpx.get(f'{url}{cast_id}')
    return True if r.status_code == 200 else False
Enter fullscreen mode Exit fullscreen mode

您进行 api 调用以获取具有给定 id 的转换，如果转换存在则返回 true，否则返回 false。

创建 Casts 服务

与 类似movie-service，要创建 ，casts-service您将使用 FastAPI 和 PostgreSQL 数据库。

创建如下所示的文件夹结构：

python-microservices/
.
├── cast_service/
│   ├── app/
│   │   ├── api/
│   │   │   ├── casts.py
│   │   │   ├── db_manager.py
│   │   │   ├── db.py
│   │   │   ├── models.py
│   │   ├── main.py
│   ├── Dockerfile
│   └── requirements.txt
├── movie_service/
...
Enter fullscreen mode Exit fullscreen mode

添加以下内容requirements.txt：

asyncpg==0.20.1
databases[postgresql]==0.2.6
fastapi==0.48.0
SQLAlchemy==1.3.13
uvicorn==0.11.2
Enter fullscreen mode Exit fullscreen mode

Dockerfile：

FROM python:3.8-slim

WORKDIR /app

COPY ./requirements.txt /app/requirements.txt

RUN apt-get update \
    && apt-get install gcc -y \
    && apt-get clean

RUN pip install -r /app/requirements.txt \
    && rm -rf /root/.cache/pip

COPY . /app/
Enter fullscreen mode Exit fullscreen mode

main.py

#~/python-microservices/cast-service/app/main.py

from fastapi import FastAPI
from app.api.casts import casts
from app.api.db import metadata, database, engine

metadata.create_all(engine)

app = FastAPI()

@app.on_event("startup")
async def startup():
    await database.connect()

@app.on_event("shutdown")
async def shutdown():
    await database.disconnect()

app.include_router(casts, prefix='/api/v1/casts', tags=['casts'])
Enter fullscreen mode Exit fullscreen mode

您添加了 前缀，/api/v1/casts以便更轻松地管理 API。此外，添加 前缀还能让您更轻松地在 FastAPI 文档中tags查找 相关的文档。casts

casts.py

#~/python-microservices/cast-service/app/api/casts.py

from fastapi import APIRouter, HTTPException
from typing import List

from app.api.models import CastOut, CastIn, CastUpdate
from app.api import db_manager

casts = APIRouter()

@casts.post('/', response_model=CastOut, status_code=201)
async def create_cast(payload: CastIn):
    cast_id = await db_manager.add_cast(payload)

    response = {
        'id': cast_id,
        **payload.dict()
    }

    return response

@casts.get('/{id}/', response_model=CastOut)
async def get_cast(id: int):
    cast = await db_manager.get_cast(id)
    if not cast:
        raise HTTPException(status_code=404, detail="Cast not found")
    return cast
Enter fullscreen mode Exit fullscreen mode

db_manager.py

#~/python-microservices/cast-service/app/api/db_manager.py

from app.api.models import CastIn, CastOut, CastUpdate
from app.api.db import casts, database


async def add_cast(payload: CastIn):
    query = casts.insert().values(**payload.dict())

    return await database.execute(query=query)

async def get_cast(id):
    query = casts.select(casts.c.id==id)
    return await database.fetch_one(query=query)
Enter fullscreen mode Exit fullscreen mode

db.py

#~/python-microservices/cast-service/app/api/db.py

import os

from sqlalchemy import (Column, Integer, MetaData, String, Table,
                        create_engine, ARRAY)

from databases import Database

DATABASE_URI = os.getenv('DATABASE_URI')

engine = create_engine(DATABASE_URI)
metadata = MetaData()

casts = Table(
    'casts',
    metadata,
    Column('id', Integer, primary_key=True),
    Column('name', String(50)),
    Column('nationality', String(20)),
)

database = Database(DATABASE_URI)
Enter fullscreen mode Exit fullscreen mode

models.py

#~/python-microservices/cast-service/app/api/models.py

from pydantic import BaseModel
from typing import List, Optional

class CastIn(BaseModel):
    name: str
    nationality: Optional[str] = None


class CastOut(CastIn):
    id: int


class CastUpdate(CastIn):
    name: Optional[str] = None
Enter fullscreen mode Exit fullscreen mode

使用 Docker Compose 运行微服务

要运行微服务，请创建一个docker-compose.yml文件并向其中添加以下内容：

version: '3.7'

services:
  movie_service:
    build: ./movie-service
    command: uvicorn app.main:app --reload --host 0.0.0.0 --port 8000
    volumes:
      - ./movie-service/:/app/
    ports:
      - 8001:8000
    environment:
      - DATABASE_URI=postgresql://movie_db_username:movie_db_password@movie_db/movie_db_dev
      - CAST_SERVICE_HOST_URL=http://cast_service:8000/api/v1/casts/

  movie_db:
    image: postgres:12.1-alpine
    volumes:
      - postgres_data_movie:/var/lib/postgresql/data/
    environment:
      - POSTGRES_USER=movie_db_username
      - POSTGRES_PASSWORD=movie_db_password
      - POSTGRES_DB=movie_db_dev

  cast_service:
    build: ./cast-service
    command: uvicorn app.main:app --reload --host 0.0.0.0 --port 8000
    volumes:
      - ./cast-service/:/app/
    ports:
      - 8002:8000
    environment:
      - DATABASE_URI=postgresql://cast_db_username:cast_db_password@cast_db/cast_db_dev

  cast_db:
    image: postgres:12.1-alpine
    volumes:
      - postgres_data_cast:/var/lib/postgresql/data/
    environment:
      - POSTGRES_USER=cast_db_username
      - POSTGRES_PASSWORD=cast_db_password
      - POSTGRES_DB=cast_db_dev

volumes:
  postgres_data_movie:
  postgres_data_cast:
Enter fullscreen mode Exit fullscreen mode

这里有 4 个不同的服务：movie_service、movie_service 的数据库、cast_service 以及 cast 服务的数据库。您已将 暴露movie_service给 port ，8001类似于cast_serviceport 8002。

对于数据库，您已经使用了卷，以便在 Docker 容器关闭时数据不会被破坏。

使用以下命令运行 docker-compose：

docker-compose up -d
Enter fullscreen mode Exit fullscreen mode

如果不存在，这将创建 docker 镜像并运行它们。

前往http://localhost:8002/docs在 Casts 服务中添加演员阵容。同样，前往http://localhost:8001/docs在 Movie 服务中添加电影。

使用 Nginx 通过单个主机地址访问两个服务

您已经使用 Docker Compose 部署了微服务，但还有一个小问题。每个微服务都需要使用不同的端口访问。您可以使用Nginx 反向代理来解决这个问题。使用 Nginx，您可以添加一个中间件来重定向请求，该中间件会根据 API URL 将请求路由到不同的服务。

nginx_config.conf在里面添加一个新文件python-microservices，内容如下。


server {
  listen 8080;

  location /api/v1/movies {
    proxy_pass http://movie_service:8000/api/v1/movies;
  }

  location /api/v1/casts {
    proxy_pass http://cast_service:8000/api/v1/casts;
  }

}
Enter fullscreen mode Exit fullscreen mode

这里，您在端口上运行 Nginx 8080，如果端点以 开头，则将请求路由到电影服务；/api/v1/movies如果端点以 开头，则将请求路由到演员服务。/api/v1/casts

现在，您需要在我们的 中添加 nginx 服务docker-compose-yml。在服务之后添加以下服务cast_db：

nginx:
    image: nginx:latest
    ports:
      - "8080:8080"
    volumes:
      - ./nginx_config.conf:/etc/nginx/conf.d/default.conf
    depends_on:
      - cast_service
      - movie_service
Enter fullscreen mode Exit fullscreen mode

现在，使用以下命令关闭容器：

docker-compose down
Enter fullscreen mode Exit fullscreen mode

然后再次运行：

docker-compose up -d
Enter fullscreen mode Exit fullscreen mode

现在，你可以在端口 访问电影服务和演员服务8080。
访问http://localhost:8080/api/v1/movies/获取电影列表。

现在，你可能想知道如何访问服务的文档。例如，main.py电影服务的更新和替换

app = FastAPI()
Enter fullscreen mode Exit fullscreen mode

和

app = FastAPI(openapi_url="/api/v1/movies/openapi.json", docs_url="/api/v1/movies/docs")
Enter fullscreen mode Exit fullscreen mode

同样，对于演员服务将其替换为

app = FastAPI(openapi_url="/api/v1/casts/openapi.json", docs_url="/api/v1/casts/docs")
Enter fullscreen mode Exit fullscreen mode

在这里，您更改了文档的服务端点和openapi.json服务地点。

现在，您可以从http://localhost:8080/api/v1/movies/docs访问电影服务的文档，从http://localhost:8080/api/v1/casts/docs访问演员服务的文档。

如果你在某个时候遇到困难或者只是想看看完整的代码，请前往Github Repo

结论和下一步

微服务架构非常适合将大型单体应用分解为独立的业务逻辑，但这也带来了一些复杂性。Python 非常适合构建微服务，因为它拥有丰富的开发者经验，并且拥有大量的软件包和框架，可以提高开发人员的生产力。

得益于 Docker，部署微服务变得更加轻松。了解更多关于如何使用 Docker 和 Docker Compose 开发微服务的信息

想让我聊聊任何话题吗？欢迎在Twitter上或在下方留言。

文章来源：https://dev.to/paurakhsharma/microservice-in-python-using-fastapi-24cc
Devto 翻译
PREV
2024 年每位云工程师都应该了解的 7 种编程语言！
NEXT
最适合初学者的 JavaScript 编程项目！👨🏻‍💻