Реализация gRPC-сервиса на Python

Продолжая недавнюю статью про проектирование gRPC-сервиса, сегодня я расскажу про практическое использование сформированного proto-файла, реализовав небольшую систему работы с поставщиками на Python.

Архитектура gRPC-сервиса

Поскольку gRPC-сервис работает по принципу клиент-серверной архитектуры, нужно реализовать как серверную, так и клиентскую части. Причем, в отличие от веб-API в стиле REST, клиент в gRPC представляет собой полноценное приложение, а не просто HTML-страницу. Визуализируем эту архитектуру в виде набора диаграмм С4. В качестве примера я рассматриваю ранее спроектированную систему работы с поставщиками, которая имеет следующие функции:

• выдавать информацию о поставщике по его ИНН: название, адрес, телефон, email;
• принимать поток заказов этому поставщику до тех пор, пока количество единиц товара не превысит определенный лимит, например, 50 единиц. По завершении приема заказов система выдает общую сумму и суммарное количество товарных единиц, отправленных в заказ, а также дату и время окончания приема.

Уровень контекста выглядит так:

Скрипт PlantUML для этой диаграммы C4:

@startuml
!include <C4/C4_Container>
title Уровень контекста

Person(U, "Пользователь", "Пользователь сервиса")
System(S, "Система работы с поставщиками", "gRPC-сервис")
Rel(U, S, "Отправить заказы поставщику")
Rel(U, S, "Узнать подробности о поставщике по ИНН")
SHOW_LEGEND()
@enduml

Раскроем состав системы в виде контейнеров на следующем уровне С4:

Скрипт PlantUML для этой диаграммы C4:

@startuml
!include <C4/C4_Container>
title Уровень контейнеров

Person(U, "Пользователь", "Пользователь сервиса")
System_Boundary(S, "gRPC-сервис работы с поставщиками"){
  Container(F, "Клиентское приложение", Python Flask)
  Container(B, "Серверное приложение", Python)
}

Rel(U, F, "Отправить заказы поставщику")
Rel(U, F, "Узнать подробности о поставщике по ИНН")
Rel(F, B, "Отправить унарный запрос о поставщике по ИНН")
Rel(B, F, "Отправить унарный ответ с данными о поставщике")
Rel(F, B, "Отправить поток запросов с заказами")
Rel(B, F, "Отправить унарный ответ на поток запросов с заказами")
SHOW_LEGEND()
@enduml

Поскольку клиентское приложение представляет собой Flask-приложение, т.е. Python-процесс, запущенный на определенном сокете, т.е. конкретном порту вычислительного узла (в моем случае, моего личного ноутбука), и HTML-страницы как GUI, покажем это на схеме компонентов. Также на схеме компонентов отразим названия методов и сообщений, ранее определенных в proto-файле.

Скрипт PlantUML для этой диаграммы C4:

@startuml
!include https://raw.githubusercontent.com/plantuml-stdlib/C4-PlantUML/master/C4_Container.puml
!include https://raw.githubusercontent.com/plantuml-stdlib/C4-PlantUML/master/C4_Component.puml
title Уровень компонентов

Person(U, "Пользователь", "Пользователь сервиса")
System_Boundary(S, "gRPC-сервис работы с поставщиками"){
  Container(F, "Клиентское приложение", Python Flask){
  Component(GUI, "Интерфейс пользователя", "HTML")
  Component(Flask, "Клиентское приложение", "Flask")
}
  Container(B, "Серверное приложение", Python)
}

Rel(U, GUI, "Отправить заказы поставщику", HTTP)
Rel(U, GUI, "Узнать подробности о поставщике по ИНН", HTTP)
Rel(GUI, Flask, ProviderINN, HTTP)
Rel(GUI, Flask, OrderRequest, HTTP)
Rel(Flask, B, GetInfoByINN(ProviderINN), HTTP/2)
Rel(B, Flask, ProviderData, HTTP/2)
Rel(Flask, B, StreamOrders(stream OrderRequest), HTTP/2)
Rel(B, Flask, OrderResponse, HTTP/2)

SHOW_LEGEND()
@enduml

Таким образом, для реализации gRPC-сервиса, помимо определения его интерфейса в proto-файле с входными/выходными сообщениями и RPC-функциями, необходимо реализовать серверное и клиентские приложения, включая GUI для клиентской части в виде хотя бы одной HTML-страницы. Как это сделать, рассмотрим далее.

Основы архитектуры и интеграции информационных систем

Код курса

OAIS

Ближайшая дата курса

20 января, 2025

Продолжительность

16 ак.часов

Стоимость обучения

36 000 руб.

Программная реализация на Python

Как я рассказывала в прошлой статье, proto-файл с определением входных и выходных структур данных, а также RPC-функций является интерфейсом, который должны реализовать компоненты проектируемой системы, т.е. ее клиентская и серверная части. Заготовки для их фактической реализации генерируются автоматически из proto-файла с помощью protobuf-компилятора protoc. Мой proto-файл order.proto выглядит вот так:

syntax = "proto3";

import "google/protobuf/timestamp.proto"; //для работы с датой и временем

package order;

// ИНН поставщика
message ProviderINN {
  string inn = 1; // ИНН поставщика
}

// Информация о поставщике
message ProviderData {
  string inn = 1; // ИНН поставщика
  string name = 2; // Имя поставщика
  string phone = 3; // Телефон поставщика
  string email = 4; // Электронная почта поставщика
  string address = 5; // Адрес поставщика
}

// Информация о товаре в заказе
message OrderItem {
    string product = 1;
    int32 quantity = 2;
    double price = 3;
}

// Запрос на заказ
message OrderRequest {
  ProviderINN provider = 1; // Информация о поставщика
  repeated OrderItem item = 2; // Список товаров в заказе
}

// Унарный ответ на поток заказов
message OrderResponse {
    int32 quantity = 1;
    double amount = 2;
    google.protobuf.Timestamp orderdate = 3;
}

// Сервис
service OrderService {
  rpc GetInfoByINN (ProviderINN) returns (ProviderData);
  rpc StreamOrders (stream OrderRequest) returns (OrderResponse);
}

Чтобы сгенерировать из него заготовки клиентской и серверной частей на Python, в PyCharm для этого нужно запустить всего лишь пару строк:

import os
os.system('python -m grpc_tools.protoc -I. --python_out=. --grpc_python_out=. order.proto')

В Google Colab можно выполнить

!pip install grpcio grpcio-tools
!python -m grpc_tools.protoc -I. --python_out=. --grpc_python_out=. order.proto

Эти инструкции запускают модуль grpc_tools.protoc, который является компилятором для proto-файлов для кодогенерации. В результате выполнения этих инструкций в директории, где лежал proto-файл, появились 2 файла:

• order_pb2.py – содержит Python-классы, которые соответствуют сообщениям, определенным в proto-файле. Эти классы позволяют создавать экземпляры сообщений, устанавливать их поля и сериализовать или десериализовать данные в формате protobuf.
• order_pb2_grpc.py — содержит Python-классы и методы клиентов, а также серверных интерфейсов для работы с gRPC-сервисом, определенными в proto-файле. Клиентские классы позволяют вызывать методы, а серверные — реализовывать их.

Содержимое этих файлов генерируется автоматически и НЕ ПОДЛЕЖИТ РУЧНОМУ РЕДАКТИРОВАНИЮ, о чем честно предупреждают комментарии.

Содержимое файла order_pb2.py:

# -*- coding: utf-8 -*-
# Generated by the protocol buffer compiler.  DO NOT EDIT!
# NO CHECKED-IN PROTOBUF GENCODE
# source: order.proto
# Protobuf Python Version: 5.27.2
"""Generated protocol buffer code."""
from google.protobuf import descriptor as _descriptor
from google.protobuf import descriptor_pool as _descriptor_pool
from google.protobuf import runtime_version as _runtime_version
from google.protobuf import symbol_database as _symbol_database
from google.protobuf.internal import builder as _builder
_runtime_version.ValidateProtobufRuntimeVersion(
    _runtime_version.Domain.PUBLIC,
    5,
    27,
    2,
    '',
    'order.proto'
)
# @@protoc_insertion_point(imports)

_sym_db = _symbol_database.Default()


from google.protobuf import timestamp_pb2 as google_dot_protobuf_dot_timestamp__pb2


DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x0border.proto\x12\x05order\x1a\x1fgoogle/protobuf/timestamp.proto\"\x1a\n\x0bProviderINN\x12\x0b\n\x03inn\x18\x01 \x01(\t\"X\n\x0cProviderData\x12\x0b\n\x03inn\x18\x01 \x01(\t\x12\x0c\n\x04name\x18\x02 \x01(\t\x12\r\n\x05phone\x18\x03 \x01(\t\x12\r\n\x05\x65mail\x18\x04 \x01(\t\x12\x0f\n\x07\x61\x64\x64ress\x18\x05 \x01(\t\"=\n\tOrderItem\x12\x0f\n\x07product\x18\x01 \x01(\t\x12\x10\n\x08quantity\x18\x02 \x01(\x05\x12\r\n\x05price\x18\x03 \x01(\x01\"T\n\x0cOrderRequest\x12$\n\x08provider\x18\x01 \x01(\x0b\x32\x12.order.ProviderINN\x12\x1e\n\x04item\x18\x02 \x03(\x0b\x32\x10.order.OrderItem\"`\n\rOrderResponse\x12\x10\n\x08quantity\x18\x01 \x01(\x05\x12\x0e\n\x06\x61mount\x18\x02 \x01(\x01\x12-\n\torderdate\x18\x03 \x01(\x0b\x32\x1a.google.protobuf.Timestamp2\x84\x01\n\x0cOrderService\x12\x37\n\x0cGetInfoByINN\x12\x12.order.ProviderINN\x1a\x13.order.ProviderData\x12;\n\x0cStreamOrders\x12\x13.order.OrderRequest\x1a\x14.order.OrderResponse(\x01\x62\x06proto3')

_globals = globals()
_builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, _globals)
_builder.BuildTopDescriptorsAndMessages(DESCRIPTOR, 'order_pb2', _globals)
if not _descriptor._USE_C_DESCRIPTORS:
  DESCRIPTOR._loaded_options = None
  _globals['_PROVIDERINN']._serialized_start=55
  _globals['_PROVIDERINN']._serialized_end=81
  _globals['_PROVIDERDATA']._serialized_start=83
  _globals['_PROVIDERDATA']._serialized_end=171
  _globals['_ORDERITEM']._serialized_start=173
  _globals['_ORDERITEM']._serialized_end=234
  _globals['_ORDERREQUEST']._serialized_start=236
  _globals['_ORDERREQUEST']._serialized_end=320
  _globals['_ORDERRESPONSE']._serialized_start=322
  _globals['_ORDERRESPONSE']._serialized_end=418
  _globals['_ORDERSERVICE']._serialized_start=421
  _globals['_ORDERSERVICE']._serialized_end=553
# @@protoc_insertion_point(module_scope)

Содержимое файла order_pb2_grpc.py:

# Generated by the gRPC Python protocol compiler plugin. DO NOT EDIT!
"""Client and server classes corresponding to protobuf-defined services."""
import grpc
import warnings

import order_pb2 as order__pb2

GRPC_GENERATED_VERSION = '1.66.1'
GRPC_VERSION = grpc.__version__
_version_not_supported = False

try:
    from grpc._utilities import first_version_is_lower
    _version_not_supported = first_version_is_lower(GRPC_VERSION, GRPC_GENERATED_VERSION)
except ImportError:
    _version_not_supported = True

if _version_not_supported:
    raise RuntimeError(
        f'The grpc package installed is at version {GRPC_VERSION},'
        + f' but the generated code in order_pb2_grpc.py depends on'
        + f' grpcio>={GRPC_GENERATED_VERSION}.'
        + f' Please upgrade your grpc module to grpcio>={GRPC_GENERATED_VERSION}'
        + f' or downgrade your generated code using grpcio-tools<={GRPC_VERSION}.'
    )


class OrderServiceStub(object):
    """Сервис
    """

    def __init__(self, channel):
        """Constructor.

        Args:
            channel: A grpc.Channel.
        """
        self.GetInfoByINN = channel.unary_unary(
                '/order.OrderService/GetInfoByINN',
                request_serializer=order__pb2.ProviderINN.SerializeToString,
                response_deserializer=order__pb2.ProviderData.FromString,
                _registered_method=True)
        self.StreamOrders = channel.stream_unary(
                '/order.OrderService/StreamOrders',
                request_serializer=order__pb2.OrderRequest.SerializeToString,
                response_deserializer=order__pb2.OrderResponse.FromString,
                _registered_method=True)


class OrderServiceServicer(object):
    """Сервис
    """

    def GetInfoByINN(self, request, context):
        """Missing associated documentation comment in .proto file."""
        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
        context.set_details('Method not implemented!')
        raise NotImplementedError('Method not implemented!')

    def StreamOrders(self, request_iterator, context):
        """Missing associated documentation comment in .proto file."""
        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
        context.set_details('Method not implemented!')
        raise NotImplementedError('Method not implemented!')


def add_OrderServiceServicer_to_server(servicer, server):
    rpc_method_handlers = {
            'GetInfoByINN': grpc.unary_unary_rpc_method_handler(
                    servicer.GetInfoByINN,
                    request_deserializer=order__pb2.ProviderINN.FromString,
                    response_serializer=order__pb2.ProviderData.SerializeToString,
            ),
            'StreamOrders': grpc.stream_unary_rpc_method_handler(
                    servicer.StreamOrders,
                    request_deserializer=order__pb2.OrderRequest.FromString,
                    response_serializer=order__pb2.OrderResponse.SerializeToString,
            ),
    }
    generic_handler = grpc.method_handlers_generic_handler(
            'order.OrderService', rpc_method_handlers)
    server.add_generic_rpc_handlers((generic_handler,))
    server.add_registered_method_handlers('order.OrderService', rpc_method_handlers)


 # This class is part of an EXPERIMENTAL API.
class OrderService(object):
    """Сервис
    """

    @staticmethod
    def GetInfoByINN(request,
            target,
            options=(),
            channel_credentials=None,
            call_credentials=None,
            insecure=False,
            compression=None,
            wait_for_ready=None,
            timeout=None,
            metadata=None):
        return grpc.experimental.unary_unary(
            request,
            target,
            '/order.OrderService/GetInfoByINN',
            order__pb2.ProviderINN.SerializeToString,
            order__pb2.ProviderData.FromString,
            options,
            channel_credentials,
            insecure,
            call_credentials,
            compression,
            wait_for_ready,
            timeout,
            metadata,
            _registered_method=True)

    @staticmethod
    def StreamOrders(request_iterator,
            target,
            options=(),
            channel_credentials=None,
            call_credentials=None,
            insecure=False,
            compression=None,
            wait_for_ready=None,
            timeout=None,
            metadata=None):
        return grpc.experimental.stream_unary(
            request_iterator,
            target,
            '/order.OrderService/StreamOrders',
            order__pb2.OrderRequest.SerializeToString,
            order__pb2.OrderResponse.FromString,
            options,
            channel_credentials,
            insecure,
            call_credentials,
            compression,
            wait_for_ready,
            timeout,
            metadata,
            _registered_method=True)

Те, кто внимательно рассмотрел сгенерированные файлы, заметили, что в них описана клиентская часть OrderServiceStub, которая используется для вызова методов на сервере. Она генерируется автоматически для того, чтобы клиент мог обращаться к сервису, описанному в proto-файле. А серверная часть, реализована не полностью, на что указывают исключения NotImplementedError в функциях GetInfoByINN() и StreamOrders() класса OrderServiceServicer. Класс OrderServiceServicer представляет собой шаблон для сервиса, который нуждается в реализации бизнес-логики указанных методов.

Полноценная реализация делается в коде серверного и клиентского приложения. Например, серверная часть у меня описана в файле order_server.py:

import grpc
import datetime
from concurrent import futures
import order_pb2
import order_pb2_grpc
from faker import Faker
from collections import defaultdict

class OrderServiceServicer(order_pb2_grpc.OrderServiceServicer):
    def __init__(self):
        self.fake = Faker('ru_RU')

    def GetInfoByINN(self, request, context):
        data = order_pb2.ProviderData(
            inn=request.inn,
            name=self.fake.name(),
            phone=self.fake.phone_number(),
            address=self.fake.address(),
            email=self.fake.free_email()
        )
        print('Принят от клиента запрос на поставщика с ИНН ', data.inn)
        print('Поставщик с ИНН ', data.inn, ' имеет следующие реквизиты: ')
        print('Название ', data.name)
        print('Телефон ', data.phone)
        print('Адрес ', data.address)
        print('Емейл ', data.email)
        return data

    def StreamOrders(self, request_iterator, context):
        total_quantity = 0
        total_amount = 0
        items = defaultdict(int)

        for order_request in request_iterator:
            print('Принят от клиента заказ с товарами', order_request)
            print('текущее дата и время: ', datetime.datetime.now())
            for item in order_request.item:
                print('товар: ', item)
                total_quantity += item.quantity
                items[item.product] += item.quantity
                print('общее количество товаров', total_quantity)
                total_amount += item.price * item.quantity
                total_amount = round(total_amount, 2)
                print('общая сумма заказа', total_amount)
                print('')

        return order_pb2.OrderResponse(
            quantity=total_quantity,
            amount=total_amount,
            orderdate = datetime.datetime.now()
        )

def serve():
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
    order_pb2_grpc.add_OrderServiceServicer_to_server(OrderServiceServicer(), server)
    server.add_insecure_port('[::]:50051')
    server.start()
    server.wait_for_termination()

if __name__ == '__main__':
    serve()

Этот Python-скрипт реализует gRPC-сервер для обработки информации о заказах и поставщиках, генерируемые с помощью библиотеки Faker. Для отладки я, как обычно, вывожу данные в консоль, используя print(). Функция serve() создаёт gRPC-сервер с пулом потоков (до 10 штук), регистрирует сервис OrderServiceServicer на сервере, который запускается на порту 50051, и работает до его завершения.

Клиентская часть реализована в виде Flask-приложения так:

from flask import Flask, request, render_template, jsonify
import grpc
import order_pb2
import order_pb2_grpc
import random
import datetime
from datetime import datetime, timedelta

app = Flask(__name__)

# списки полей в заявке
products = ['яблоки желтые', 'малина', 'вода', 'хлеб белый','хлеб серый', 'креветки', 'форель', 'апельсины', 'кета','курица','яйцо перепелиное','яйцо куриное','лаваш',
            'булка сдобная','булка сахарная','помидоры бакинские','помидоры чери','огурцы','перец сладкий','перец острый','перец болгарский','мандарины','укроп свежий',
            'укроп сушеный','клубника свежая','клубника мороженная','мороженое','картошка','морковь', 'свекла','пангасиус','семга','кальмар замороженный','горошек зеленый',
            'смородина черная','смородина красная','соль поваренная пищевая йодированная','чай черный байховый','чай зеленый','чай красный','кофе','кофе с молоком','какао',
            'молоко','кефир','сыр с плесенью','сыр плавленый','сыр твердый','сыр мягкий','яблоки красные','яблоки зеленые','яблоки сушеные','икра красная',
            'икра черная','икра заморская баклажанная','масло сливочное','масло оливковое','масло подсолнечное','масло кокосовое','орех грецкий','орех бразильский',
            'лист лавровый','куркума','кукуруза','печенье сладкое','пряники сдобные','тесто слоеное','варенц','ряженка','снежок','шоколад молочный']

# Функция для получения информации по ИНН от gRPC сервера
def get_info_by_inn(inn):
    with grpc.insecure_channel('localhost:50051') as channel:
        stub = order_pb2_grpc.OrderServiceStub(channel)
        provider_request = order_pb2.ProviderINN(inn=inn)
        response = stub.GetInfoByINN(provider_request)
        provider_info = {
            'inn': response.inn,
            'name': response.name,
            'phone': response.phone,
            'email': response.email,
            'address': response.address
        }
    return provider_info

# Функция для отправки потока заказов на gRPC сервер
def stream_orders(provider_info, items):
    with grpc.insecure_channel('localhost:50051') as channel:
        stub = order_pb2_grpc.OrderServiceStub(channel)

        def order_request_iterator():
            for item in items:
                yield order_pb2.OrderRequest(
                    provider=order_pb2.ProviderINN(
                        inn=provider_info['inn']
                    ),
                    item=[
                        order_pb2.OrderItem(
                            product=o['product'],
                            quantity=int(o['quantity']),
                            price=float(o['price'])
                        )
                        for o in item
                    ]
                )
        server_answer = stub.StreamOrders(order_request_iterator())
        return server_answer

#Обработчик HTTP-запросов
@app.route('/', methods=['GET', 'POST'])
def index():
    if request.method == 'POST':
        if 'inn' in request.form:
            inn = request.form['inn']
            provider_info = get_info_by_inn(inn)
            return render_template('index.html', provider_info=provider_info)
        elif 'send_orders' in request.form:
            provider_info = {
                'inn': request.form['inn_r'],
                'name': request.form['name'],
                'email': request.form['email'],
                'phone': request.form['phone'],
                'address': request.form['address']
            }
            k = 0
            items = []  # Создаем список заказов вне цикла
            while True:
                item = {
                    'product': random.choice(products),
                    'quantity': random.randint(1, 10),
                    'price': round(random.uniform(1, 1000), 2)
                }
                items.append(item)  # Добавляем новый элемент в список

                # Отправка заказа и получение ответа от сервера
                unary_server_answer = stream_orders(provider_info, [items])
                # Преобразуем временную метку в объект datetime
                timestamp_seconds = unary_server_answer.orderdate.seconds
                timestamp_nanos = unary_server_answer.orderdate.nanos
                order_datetime = datetime.fromtimestamp(timestamp_seconds) + timedelta(
                    microseconds=timestamp_nanos / 1000)

                # Форматируем дату и время в нужный формат
                formatted_date = order_datetime.strftime('%Y-%m-%d %H:%M:%S')

                if unary_server_answer.quantity>50:
                    break

            return render_template('index.html', provider_info=provider_info, orderdate=formatted_date,
                                   quantity=unary_server_answer.quantity,amount=unary_server_answer.amount, products=items)
    else:
        return render_template('index.html', provider_info=None, server_answer=None)

if __name__ == '__main__':
    app.run()

Библиотеки Flask используется для создания веб-сервера, который будет обрабатывать HTTP-запросы, а gRPC – для взаимодействия с удалённым сервером через методы, определённые в файлах order_pb2 и order_pb2_grpc. Функция get_info_by_inn() принимает ИНН и запрашивает информацию о поставщике у gRPC-сервиса, создавая с ним канал связи по сокету localhost:50051, т.е. порт 50051 на локальном хосте, и вызывая метод GetInfoByINN(ProviderINN). Функция stream_orders() также устанавливает канал связи с gRPC-сервером и отправляет ему поток заказов, используя генератор order_request_iterator, который создаёт поток объектов OrderRequest. Метод StreamOrders(OrderRequest) отправляет эти заказы на gRPC-сервер и получает от него унарный ответ, когда количество единиц товара превысит 50.

Наконец, поскольку Flask использует шаблоны для HTML-страниц, чтобы отделить логику приложения от представления, необходимо также написать разметку для страницы. Библиотека Jinja2 для рендеринга HTML позволяет динамически генерировать Flask-приложению веб-страницу, вставляя значения переменных в двойных фигурных скобках. Код моей HTML-страницы index.html выглядит так:

<!DOCTYPE html>
<html lang="ru">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>Отправка заказов поставщику</title>
    <style>
        input[type="text"] {
            width: 40%;
        }
    </style>
</head>
<body>
    <h2>Введите ИНН поставщика, чтобы узнать о нем полную информацию</h2>
    <form method="POST">
        <label for="inn">ИНН:</label>
        <input type="text" id="inn" name="inn">
        <button type="submit">Получить информацию</button>
    </form>
    {% if provider_info %}
        <h2>Информация о поставщике с ИНН {{ provider_info.inn }}</h2>
        <form method="POST">
            <input type="hidden" name="send_orders" value="true">
            <label for="inn-r">ИНН:</label>
            <input type="text" id="inn-r" name="inn_r" value="{{ provider_info.inn }}" readonly>
            <br>
            <label for="name">Имя:</label>
            <input type="text" id="name" name="name" value="{{ provider_info.name }}" readonly>
            <br>
            <label for="phone">Телефон:</label>
            <input type="text" id="phone" name="phone" value="{{ provider_info.phone }}" readonly>
            <br>
            <label for="email">Email:</label>
            <input type="text" id="email" name="email" value="{{ provider_info.email }}" readonly>
            <br>
            <label for="address">Адрес:</label>
            <input type="text" id="address" name="address" value="{{ provider_info.address }}" readonly>
            <h2>Отправьте заказы этому поставщику</h2>
            <button type="submit">Отправить заказы</button>
            {% if products %}
            <h3>Заказы, отправленные этому поставщику:</h3>
            <table>
                <thead>
                    <tr>
                        <th>Продукт</th>
                        <th>Количество, шт.</th>
                        <th>Цена, руб</th>
                    </tr>
                </thead>
                <tbody>
                    {% for order in products %}
                        <tr>
                            <td>{{ order.product }}</td>
                            <td>{{ order.quantity }}</td>
                            <td>{{ order.price }}</td>
                        </tr>
                    {% endfor %}
                </tbody>
            </table>
            {% endif %}
        </form>
    {% endif %}
    {% if orderdate %}
        <p><b>Дата и время</b> {{ orderdate }}</p>
    {% endif %}
    {% if quantity %}
        <p><b>Заказано товаров</b> {{ quantity }}</p>
    {% endif %}
    {% if amount %}
        <p><b>На общую сумму</b> {{ amount }}</p>
    {% endif %}
</body>
</html>

Запустив серверную и клиентскую части, можно протестировать разработанную систему.

Клиентская часть в виде Flask-приложения запущена на 5000 порту локального хоста, что показывает HTML-страница созданного веб-сервиса. Наконец, можно поработать с созданным gRPC-сервисом, протестировав оба его метода.

В терминале PyCharm также выводится отладочная информация.

Весь код можно, как обычно, доступен в моем Github-репозитории: https://github.com/AnnaVichugova/PythonApps/tree/main/gRPC

В заключение отмечу, что работать с gRPC довольно интересно, однако, большое количество объектов разработки увеличивает поле возможных ошибок. Впрочем, жесткие контракты данных, определенные в proto-файле, стараются сократить их количество.

Дизайн API — проектирование веб-приложений

Код курса

DAPI

Ближайшая дата курса

27 января, 2025

Продолжительность

16 ак.часов

Стоимость обучения

36 000 руб.

Больше примеров и подробностей про архитектуру и интеграцию информационных систем вы узнаете на моих курсах в Школе прикладного бизнес-анализа и проектирования информационных систем на базе нашего лицензированного учебного центра обучения и повышения квалификации системных и бизнес-аналитиков в Москве:

• Дизайн API — проектирование веб-приложений
• Основы архитектуры и интеграции информационных систем