Работа с микросервисами через API-gateway

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import declarative_base, sessionmaker

engine = create_async_engine("sqlite+aiosqlite:///clients.sqlite")
sm = sessionmaker(
    engine, autocommit=False, autoflush=False, class_=AsyncSession
)

Base = declarative_base()

Для работы используем асинхронные сессии и SQLite в качестве СУБД.

from sqlalchemy import Column, Integer, Text
from sqlalchemy.orm.collections import InstrumentedList

from srv_clients.database import Base


class BaseModel(Base):
    __abstract__ = True
    id = Column(Integer, primary_key=True, comment="Идентификатор")

    def to_dict(self) -> dict:
        data = dict()
        for k, v in self.__dict__.items():
            if k.startswith("_"):
                continue
            elif isinstance(v, Base):
                v = v.to_dict()
            elif isinstance(v, InstrumentedList):
                v = [item.to_dict() for item in v]
            data[k] = v
        return data

Здесь мы создали базовую абстрактную модель. В ней объявлено поле уникального идентификатора, являющееся также первичным ключом. Кроме того, реализовали простейший рекурсивный метод to_dict для преобразования экземпляра класса в словарь. Да, он не охватывает все ситуации и может вызывать кучу ошибок и требует более детальной проработки, но для примера его более чем достаточно.

class Client(BaseModel):
    __tablename__ = "client"
    __table_args = {"comment": "Таблица клиентов"}

    surname = Column(Text(length=255), nullable=False, comment="Фамилия")
    name = Column(Text(length=255), nullable=False, comment="Имя")
    country = Column(Text(length=255), comment="Страна")

Создадим унаследованную от BaseModel модель Клиента, добавим к ней комментарий и несколько текстовых полей длиной до 255 символов так же с пояснениями.

from typing import List

from sqlalchemy import delete, select, update
from sqlalchemy.ext.asyncio import AsyncSession

from srv_clients.models import Client


class ClientQueryset:
    model = Client

    @classmethod
    async def create(cls, session: AsyncSession, **kwargs) -> int:
        created = cls.model(**kwargs)
        session.add(created)
        await session.flush([created])
        return created.id

Метод создания записи будет возвращать идентификатор созданной записи. Так как идентификатор является автоинкрементом и в простейших условиях генерируется только после закрытия транзакции, принудительно вызовем метод сессии flush, который сбросит данные в БД и атрибуту id присвоится значение.

    @classmethod
    async def get_by_id(cls, session: AsyncSession, id_: int) -> Client:
        return await session.scalar(
            select(cls.model).where(cls.model.id == id_)
        )

    @classmethod
    async def get_multiple(
        cls,
        session: AsyncSession,
        limit: int = None,
        offset: int = None,
        **filters,
    ) -> List[Client]:
        where = list()
        if surname := filters.get("surname"):
            where.append(cls.model.surname == surname)
        if name := filters.get("name"):
            where.append(cls.model.name == name)
        if country := filters.get("country"):
            where.append(cls.model.country == country)
        return await session.scalars(
            select(cls.model).where(*where).limit(limit).offset(offset)
        )

Для получения клиента по идентификатору и отфильтрованного списка записей используем методы сессии scalar и scalars соответственно. Условия фильтрации ожидаем в виде динамических именованных параметров. Используем простое сравнение полей и добавим возможность реализации пагинации с помощью параметров limit и offset.

    @classmethod
    async def update(cls, session: AsyncSession, id_: int, **kwargs) -> None:
        await session.execute(
            update(cls.model).where(cls.model.id == id_).values(**kwargs)
        )

    @classmethod
    async def delete(cls, session: AsyncSession, id_: int) -> None:
        await session.execute(delete(cls.model).where(cls.model.id == id_))

Все методы обернуты в декоратор @classmethod для возможности вызова без создания экземпляра класса.

from pydantic import BaseModel, Field


class ClientSchema(BaseModel):
    surname: str = Field(..., description="Фамилия")
    name: str = Field(..., description="Имя")
    country: str = Field(..., description="Страна")


class ClientViewSchema(ClientSchema):
    id: int = Field(..., description="Идентификатор")

Все атрибуты сделаем обязательными (в качестве значения по умолчанию передадим тип ellipsis — ...). К модели просмотра добавим атрибут идентификатора.

from typing import List

from fastapi import FastAPI, HTTPException, Request
from fastapi.params import Query
from starlette import status

from srv_clients.database import Base, engine
from srv_clients.database import sm as session_maker
from srv_clients.querysets import ClientQueryset
from srv_clients.schemas import ClientSchema, ClientViewSchema

app = FastAPI()
app.state.engine = engine
app.state.session_maker = session_maker

Создаем приложение, добавляем к нему в качестве атрибутов подключение к базе данных и инициализатор сессий. Делаем это для того, чтобы иметь возможность обращаться к ним внутри эндпоинтов через экземплярRequest.

@app.on_event("startup")
async def on_startup():
    async with app.state.engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)


@app.on_event("shutdown")
async def on_shutdown():
    # async with app.state.engine.begin() as conn:
    #     await conn.run_sync(Base.metadata.drop_all)
    await app.state.engine.dispose()

Добавим два обработчика событий приложения:

• startup— переводим модели SQLAlchemy в таблицы БД;

• shutdown— очищаем БД путем удаления всех таблиц (закомментировано) и корректно завершаем соединение с БД.

@app.get("/clients", response_model=List[ClientViewSchema])
async def get_filtered_clients(
    request: Request,
    surname: str = Query(None, description="Фамилия"),
    name: str = Query(None, description="Имя"),
    country: str = Query(None, description="Страна"),
    limit: int = Query(
        None, description="Количество записей на странице результатов"
    ),
    offset: int = Query(
        None,
        description="Смещение страницы результатов относительно первой записи",
    ),
):
    async with request.app.state.session_maker() as session:
        rows = await ClientQueryset.get_multiple(
            session,
            surname=surname,
            name=name,
            country=country,
            limit=limit,
            offset=offset,
        )
        return [row.to_dict() for row in rows.all()]


@app.post(
    "/clients",
    response_model=ClientViewSchema,
    status_code=status.HTTP_201_CREATED,
)
async def create_client(request: Request, data: ClientSchema):
    sm = request.app.state.session_maker
    async with sm.begin() as session:
        id_ = await ClientQueryset.create(session, **data.dict())
    async with sm() as session:
        client = await ClientQueryset.get_by_id(session, id_)
        return client.to_dict()


@app.get("/clients/{client_id}", response_model=ClientViewSchema)
async def get_client(request: Request, client_id: int):
    async with request.app.state.session_maker() as session:
        client = await ClientQueryset.get_by_id(session, client_id)
        if not client:
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND,
                detail="Client not found",
            )
        return client.to_dict()


@app.patch(
    "/clients/{client_id}",
    response_model=ClientViewSchema,
    status_code=status.HTTP_202_ACCEPTED,
)
async def update_client(request: Request, client_id: int, data: ClientSchema):
    sm = request.app.state.session_maker
    async with sm.begin() as session:
        client = await ClientQueryset.get_by_id(session, client_id)
        if not client:
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND, detail="Client not found"
            )
        await ClientQueryset.update(session, client_id, **data.dict())

    async with sm() as session:
        client = await ClientQueryset.get_by_id(session, client_id)
        return client.to_dict()


@app.delete(
    "/clients/{client_id}",
    status_code=status.HTTP_202_ACCEPTED,
)
async def delete_client(request: Request, client_id: int):
    sm = request.app.state.session_maker
    async with sm.begin() as session:
        client = await ClientQueryset.get_by_id(session, client_id)
        if not client:
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND, detail="User not found"
            )
        await ClientQueryset.delete(session, client_id)

Реализуем здесь простой CRUD с транзакциями. Для операций, вносящих изменения в БД, откроем транзакцию (создадим точку сохранения — savepoint) с помощью метода begin и менеджера контекста, при выходе из которого она будет успешно закрыта (под капотом будет вызван оператор commit). В случае возникновения каких‑либо ошибок внутри блока with транзакция завершится откатом (будет вызван оператор rollback).

Для разнообразия добавим фильтрацию —get_filtered_clients.

Для случаев, когда запись из БД обязательно должна быть получена (например, при изменении или удалении), выбросим исключение HTTPException когда она не найдена.

Для преобразования ответа в экземпляр схемы, передаваемой в параметре response_modelэндпоинта, или их список, реализующая его функция должна возвращать соответственно словарь или список словарей. Тут нам и пригодится метод to_dict базовой модели данных.

Здесь отступление от первичной цели — хранение и получение настроек в yaml‑файле, а также обращение к переменным окружения из конфигурационного файла. Для этого используем библиотеку PyYaml‑Tags. Пока просто объявим класс и переопределим приватный метод _from_yaml. Вызов будет чуть ниже.

import os
from pathlib import Path

import yaml
from dotenv import load_dotenv
from yaml_tags import BaseTag, tag_registry


@tag_registry.register("env_tag")
class EnvTag(BaseTag):
    def _from_yaml(
        self,
        _loader,
        _work_dir,
        _prefix,
        _suffix,
        param=None,
        *args,
        **kwargs,
    ):
        result = os.environ.get(param, "")
        return f"{_prefix}{result}{_suffix}"

Регистрируем свой тэг env_tag и метод его и формирования.

def load_config(config_path: Path) -> dict:
    tag_registry.require("env_tag")
    env_path = f"{config_path.absolute().parent}/.env"
    load_dotenv(dotenv_path=env_path)
    with open(config_path) as f:
        return yaml.load(f, Loader=yaml.Loader)


def get_config_path(file_name: str) -> Path:
    current_dir = Path(__file__).absolute().parent
    return current_dir / "config" / file_name

Тут мы считаем .env файл. Будем ожидать его в каталоге с конфигурационным(‑ми) файлом(‑ами).

local.yaml:

rmq:
  url: amqp://gateway:<% env_tag(param="RMQ_PASS") %>@127.0.0.1:5672/gateway

В данном случае храним только строку подключения к RabbitMQ.

.env:

RMQ_PASS=gateway

Обращение к зарегистрированному тэгу простое: <% [имя_тэга]([имя_именованного_параметра]=[имя_переменной_окружения]) %>

from typing import Tuple

import aiohttp

SERVICE_MAP = {
    "clients": "0.0.0.0:8001",
    "goods": "0.0.0.0:8002",
}


async def resolve(
    method: str,
    path: str,
    params: dict = None,
    data: dict = None,
    headers: dict = None,
) -> Tuple[dict, int]:
    """
    Функция определяет по первой части url-запроса к какому сервису 
    происходит обращение, перенаправляет в него запрос и возвращает 
    ответ в виде словаря и код статуса.
    """
    _, service_name, *_ = path.split("/")
    service_host = SERVICE_MAP[service_name]
    url = f"http://{service_host}{path}"
    response, status_code = await make_request(
        url, method, params, data, headers
    )
    return response, status_code


async def make_request(
    url: str,
    method: str,
    params: dict = None,
    data: dict = None,
    headers: dict = None,
) -> Tuple[dict, int]:
    """
    Выполнение запроса, тип которого определяется параметром method,
    по url-адресу. Возвращает ответ в виде словаря и код статуса.
    """
    
    if not headers:
        headers = {}
    if not data:
        data = {}

    async with aiohttp.ClientSession() as session:
        request = getattr(session, method)
        async with await request(
            url, params=params, json=data, headers=headers
        ) as response:
            response_json = await response.json()
            return response_json, response.status

import asyncio

from aio_pika import connect_robust
from aio_pika.patterns import RPC

from common import get_config_path, load_config
from resolver import resolve


async def server() -> None:
    config = load_config(get_config_path("local.yaml"))
    rmq = config["rmq"]

    connection = await connect_robust(rmq["url"])

    async with connection:
        channel = await connection.channel()
        rpc = await RPC.create(channel)
        await rpc.register(resolve.__name__, resolve, auto_delete=True)
    
        try:
            await asyncio.Future()
        finally:
            await connection.close()


if __name__ == "__main__":
    asyncio.run(server())

Здесь все совсем просто. Считываем конфигурационный файл, преобразовываем его в словарь и получаем настройки подключения к очереди RabbitMQ. Создаем это подключение, канал и регистрируем обработчик. Кстати, в пакете aio_pika есть и другие шаблоны, кроме RPC.

from typing import Tuple

from aio_pika import connect_robust
from aio_pika.patterns import RPC


async def send_request_to_queue(
    config: dict, message: dict
) -> Tuple[dict, int]:
    rmq = config["rmq"]
    connection = await connect_robust(rmq["url"])

    async with connection:
        channel = await connection.channel()
        rpc = await RPC.create(channel)
        result, status_code = await rpc.proxy.resolve(**message)
        return result, status_code

Из настроек получаем конфигурацию для работы с очередью и открываем соединение. Обозначаем, что хотим вызвать функцию resolve. Она, как известно, возвращает словарь и числовой код статуса.

По большей части он будет повторять код rpc‑сервера с одним дополнением — функцией router.

import os
from functools import wraps
from pathlib import Path
from typing import Any, Optional

import yaml
from dotenv import load_dotenv
from fastapi import HTTPException, Request, Response
from starlette import status
from yaml_tags import BaseTag, tag_registry

from gateway.rabbit import send_request_to_queue


@tag_registry.register("env_tag")
class EnvTag(BaseTag):
    def _from_yaml(
        self,
        _loader,
        _work_dir,
        _prefix,
        _suffix,
        param=None,
        *args,
        **kwargs,
    ) -> str:
        result = os.environ.get(param, "")
        return f"{_prefix}{result}{_suffix}"


def load_config(config_path: Path) -> dict:
    tag_registry.require("env_tag")
    env_path = f"{config_path.absolute().parent}/.env"
    load_dotenv(dotenv_path=env_path)
    with open(config_path) as f:
        return yaml.load(f, Loader=yaml.Loader)


def get_config_path(file_name: str) -> Path:
    current_dir = Path(__file__).absolute().parent
    return current_dir / "config" / file_name


def router(
    method,
    path: str,
    data_key: Optional[str] = None,
    status_code: Optional[int] = status.HTTP_200_OK,
    response_model: Optional[Any] = None,
):
    """
    Обертка функции эндпоинта, реализующая обращение к RPC-серверу.

    :param method: Вызываемый объект (функция) реализующая тот или 
                   иной метод http-запроса.
    :param path: Адрес эндпоинта.
    :param data_key: Имя ключа, в котором передаются данные 
                     на эндпоинт.
    :param status_code: Ожидаемый код ответа сервера.
    :param response_model: Модель данных для преобразования ответа.
    """
    app_method = method(
        path, status_code=status_code, response_model=response_model
    )

    def wrapper(endpoint):
        @app_method
        @wraps(endpoint)
        async def decorator(request: Request, response: Response, **kwargs):
            request_method = request.scope["method"].lower()
            data = kwargs.get(data_key)
            data = data.dict() if data else {}
            response_data, response_status_code = await send_request_to_queue(
                config=request.app.config,
                message={
                    "method": request_method,
                    "path": request.scope["path"],
                    "params": dict(request.query_params),
                    "data": data,
                    "headers": {},
                },
            )
            if response_status_code >= status.HTTP_400_BAD_REQUEST:
                raise HTTPException(
                    status_code=response_status_code, detail=response_data
                )
            response.status_code = response_status_code
            return response_data

    return wrapper

Функция router реализует декоратор, который в свою очередь традиционным методом оборачивает эндпоинт в параметризированный декоратор. Из переданных параметров собирается объект сообщения для направления в один из сервисов. Сообщение направляется в очередь и ожидается ответ. Если код ответа информирует об ошибке (>= 400), выбрасываем исключение с полученным статусом и сообщением. В случае успеха, подменяем статус объекта response кодом, полученным от сервиса.

На данном этапе функционал можно дополнить, например, проверкой учетных данных, каким‑то логированием и всяческими плюшками. Для примера ограничимся простыми запросом‑ответом.

from fastapi import FastAPI

from common import get_config_path, load_config
from gateway.srv_clients.routes import clients_router

app = FastAPI()
app.config = load_config(get_config_path("local.yaml"))
app.include_router(clients_router)

Здесь мы инициализируем приложение, добавляем к нему в качестве атрибута config словарь с настройками. Делаем это опять же для того, чтобы иметь возможность обращаться к ним внутри эндпоинтов через экземплярRequest, а именно в декораторе route.

from typing import List

from fastapi import APIRouter, Response
from fastapi.params import Query
from starlette import status
from starlette.requests import Request

from gateway.common import router
from gateway.srv_clients.schemas import ClientSchema, ClientViewSchema

clients_router = APIRouter(prefix="/clients")


@router(
    method=clients_router.get,
    path="",
    response_model=List[ClientViewSchema],
)
async def get_filtered_clients(
    request: Request,
    response: Response,
    surname: str = Query(None, description="Фамилия"),
    name: str = Query(None, description="Имя"),
    country: str = Query(None, description="Страна"),
    limit: int = Query(
        None, description="Количество записей на странице результатов"
    ),
    offset: int = Query(
        None,
        description="Смещение страницы результатов относительно первой записи",
    ),
):
    pass


@router(
    method=clients_router.post,
    path="",
    response_model=ClientViewSchema,
    status_code=status.HTTP_201_CREATED,
    data_key="data",
)
async def create_client(
    request: Request, response: Response, data: ClientSchema
):
    pass


@router(
    method=clients_router.get,
    path="/{client_id}",
    response_model=ClientViewSchema,
)
async def get_client(request: Request, response: Response, client_id: int):
    pass


@router(
    method=clients_router.patch,
    path="/{client_id}",
    response_model=ClientViewSchema,
    status_code=status.HTTP_202_ACCEPTED,
    data_key="data",
)
async def update_client(
    request: Request, response: Response, client_id: int, data: ClientSchema
):
    pass


@router(
    method=clients_router.delete,
    path="/{client_id}",
    status_code=status.HTTP_202_ACCEPTED,
)
async def delete_client(request: Request, response: Response, client_id: int):
    pass

Все эндпоинты не содержат код. Запросы выполняются внутри декоратора, в который мы оборачиваем функцию.

from pydantic import BaseModel, Field


class ClientSchema(BaseModel):
    surname: str = Field(..., description="Фамилия")
    name: str = Field(..., description="Имя")
    country: str = Field(..., description="Страна")


class ClientViewSchema(ClientSchema):
    id: int = Field(..., description="Идентификатор")

Здесь полная копия схем сервиса.