Кратко про Nameko Python

Привет, Хабр!

Nameko, как фреймворк для Python, предлагает удобные абстракции и инструменты для создания микросервисов, ориентированных на сообщения и события.

С Nameko можно получить не только инструмент для создания микросервисов, но и платформу для реализации распределенных систем с поддержкой Dependency Injection, позволяющей легко интегрировать различные компоненты приложения.

Установим через пип:

pip install nameko

Это установит Nameko и все зависимости.

После установки проверим:

nameko --version

Если выдало версию — все ок.

Встроенные расширения

AMQP — это протокол обмена сообщениями, который Nameko использует для межсервисного взаимодействия. Встроенное расширение AMQP позволяет настроить и использовать очереди сообщений для связи между сервисами:

from nameko.messaging import Publisher, rpc  class ServiceA:     name = "service_a"      publisher = Publisher()      @rpc     def send_message(self, message):         self.publisher.publish(message)  class ServiceB:     name = "service_b"      @rpc     def process_message(self, message):         print("Received message:", message)

ServiceA отправляет сообщение через очередь, а ServiceB принимает и обрабатывает его.

DependencyProvider позволяет инжектировать зависимости в ваши сервисы, обеспечивая более чистый и модульный код. Создадим сервис, который использует БД:

from nameko.dependency_providers import DependencyProvider  class DatabaseWrapper:     def __init__(self, db_url):         self.db_url = db_url      def execute_query(self, query):         # Execute query logic here         pass  class DatabaseProvider(DependencyProvider):     def setup(self):         self.database = DatabaseWrapper(self.container.config['DATABASE_URL'])      def get_dependency(self, worker_ctx):         return self.database  class MyService:     name = "my_service"      db = DatabaseProvider()      def some_method(self):         result = self.db.execute_query("SELECT * FROM table")         return result

Юзаем DatabaseProvider для предоставления экземпляра DatabaseWrapper в сервис.

Встроенное расширение HTTP позволяет создавать веб-сервисы на базе Nameko, обеспечивая взаимодействие с внешними клиентами через HTTP. Например:

from nameko.web.handlers import http  class HttpService:     name = "http_service"      @http('GET', '/hello')     def hello_world(self, request):         return 200, "Hello, World!"

RPC также нашло свою реализацию в nameko, пример:

from nameko.rpc import rpc  class GreetingService:     name = "greeting_service"      @rpc     def greet(self, name):         return f"Hello, {name}!" 

GreetingService предоставляет метод greet, который может быть вызван удаленно другими сервисами.

Events позволяет сервисам подписываться на события и реагировать на них:

from nameko.events import event_handler  class NotificationService:     name = "notification_service"      @event_handler("email_service", "email_sent")     def handle_email_sent(self, payload):         print("Email sent:", payload)

NotificationService подписывается на событие email_sent от сервиса email_service и обрабатывает его.

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

from nameko.timer import timer  class CleanupService:     name = "cleanup_service"      @timer(interval=60)  # Выполнять каждую минуту     def cleanup(self):         print("Performing cleanup...")         # Логика очистки данных

Сервис будет выполнять метод cleanup каждую минуту.

Что там с тестами?

Для написания unit-тестов в Nameko можно использовать стандартные фреймворки для тестирования Python, какpytest или unittest. Для этого создаем экземпляр сервиса в тестовой среде и вызываем его методы для проверки их поведения. Предположим, у нас есть простой сервис для работы с мат. операциями:

# math_service.py from nameko.rpc import rpc  class MathService:     name = "math_service"      @rpc     def add(self, x, y):         return x + y

Напишем unit-тест для этого сервиса с использованием pytest:

# test_math_service.py from math_service import MathService  def test_add():     service = MathService()     assert service.add(2, 3) == 5

Интеграционные тесты в Nameko позволяют проверить взаимодействие между различными сервисами и убедиться, что они взаимодействуют правильно в рамках системы. Для этого можно использовать механизм запуска сервисов в тестовом режиме и проверять их взаимодействие через RPC вызовы или события. Пример интеграционного теста с использованием pytest:

# test_order_service.py from nameko.testing.services import entrypoint_hook  from order_service import OrderService from product_service import ProductService  def test_create_order():     # создаем экземпляр сервиса OrderService     order_service = OrderService()      # создаем экземпляр сервиса ProductService     product_service = ProductService()      # входим в контекст сервиса OrderService     with entrypoint_hook(order_service, "create_order") as create_order:         # вызываем метод create_order         order_id = create_order(user_id=1, product_id=1, quantity=2)          # проверяем, что заказ успешно создан         assert order_id is not None

Иногда может потребоваться использовать mock объекты для имитации зависимостей в тестах. В Nameko это можно реализовать с помощью entrypoint механизма:

# test_email_service.py from nameko.testing.services import entrypoint_hook from unittest.mock import Mock  from email_service import EmailService  def test_send_email():     # создаем экземпляр сервиса EmailService     email_service = EmailService()      # создаем mock объект для отправки email     mock_send_email = Mock()      # подменяем реальный метод отправки email на mock объект     email_service.send_email = mock_send_email      # входим в контекст сервиса EmailService     with entrypoint_hook(email_service, "send_email") as send_email:         # вызываем метод send_email         send_email(recipient="test@example.com", subject="Test", body="This is a test email")          # проверяем, что метод send_email был вызван с правильными аргументами         mock_send_email.assert_called_once_with("test@example.com", "Test", "This is a test email")

Написание расширений

Путем написания расширений можнонастраивать поведение Nameko под конкретные потребности.

Допустим, нужно добавить механизм аутентификации к сервисам Nameko. Можно написать расширение, которое будет обрабатывать проверку подлинности запросов к сервисам:

from nameko.extensions import Extension  class AuthExtension(Extension):     def __init__(self, auth_service):         self.auth_service = auth_service      def bind(self, container):         super(AuthExtension, self).bind(container)         self.container.auth = self      def get_dependency(self, worker_ctx):         return AuthServiceWrapper(self.auth_service)  class AuthServiceWrapper:     def __init__(self, auth_service):         self.auth_service = auth_service      def authenticate(self, token):         # логика аутентификации         pass

Или к примеру хотим добавить мониторинг производительности в сервисы Nameko. Можно написать расширение, которое будет собирать статистику о времени выполнения запросов и других метриках:

from nameko.extensions import Extension  class MonitoringExtension(Extension):     def __init__(self, statsd_client):         self.statsd_client = statsd_client      def bind(self, container):         super(MonitoringExtension, self).bind(container)         self.container.monitoring = self      def record_request_time(self, service_name, method_name, time_taken):         metric_name = f"{service_name}.{method_name}.request_time"         self.statsd_client.timing(metric_name, time_taken)

После написания расширения, можно использовать его в своих сервисах. Для этого просто добавляем его в контейнер Nameko:

from nameko.rpc import rpc from my_extensions import AuthExtension  class MyService:     name = "my_service"      auth = AuthExtension()      @rpc     def my_method(self, request):         # проверка аутентификации         self.auth.authenticate(request.token)         # логика метода

---

Статья подготовлена в рамках набора на курс Python Developer. Professional. По ссылке вы сможете узнать о курсе подробнее, а также зарегистрироваться на бесплатный вебинар.