Briefly about Nameko Python

Dependency Injectionallowing you to easily integrate various application components.

Install via pip:

pip install nameko

This will install Nameko and all dependencies.

After installation, let's check:

nameko --version

If it gave the version – all OK.

Built-in extensions

AMQP is a messaging protocol that Nameko uses for inter-service communication. The built-in AMQP extension allows you to configure and use message queues for communication between services:

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 sends a message through the queue, and ServiceB accepts and processes it.

DependencyProvider allows you to inject dependencies into your services, providing cleaner and more modular code. Let's create a service that uses the database:

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

Let's use it DatabaseProvider to provide a copy DatabaseWrapper to the service.

The built-in HTTP extension allows you to create web services based on Nameko, allowing interaction with external clients via HTTP. For example:

from nameko.web.handlers import http

class HttpService:
    name = "http_service"

    @http('GET', '/hello')
    def hello_world(self, request):
        return 200, "Hello, World!"

RPC also found its implementation in nameko, example:

from nameko.rpc import rpc

class GreetingService:
    name = "greeting_service"

    @rpc
    def greet(self, name):
        return f"Hello, {name}!"

GreetingService provides a method greetwhich can be called remotely by other services.

Events Allows services to subscribe to events and respond to them:

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 subscribes to the event email_sent from the service email_service and processes it.

Timer allows you to create periodic tasks in the application:

from nameko.timer import timer

class CleanupService:
    name = "cleanup_service"

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

The service will execute the method cleanup every minute.

What's up with the tests?

For writing unit tests in Nameko you can use standard Python testing frameworks likepytest or unittest. To do this, we create an instance of the service in a test environment and call its methods to test their behavior. Let's assume we have a simple service for working with mat. operations:

# math_service.py
from nameko.rpc import rpc

class MathService:
    name = "math_service"

    @rpc
    def add(self, x, y):
        return x + y

Let's write a unit test for this service using pytest:

# test_math_service.py
from math_service import MathService

def test_add():
    service = MathService()
    assert service.add(2, 3) == 5

Integration tests in Nameko allow you to check the interaction between different services and make sure that they interact correctly within the system. To do this, you can use the mechanism of launching services in test mode and checking their interaction through RPC calls or events. An example of an integration test using 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

Sometimes you may need to use mock objects to simulate dependencies in tests. In Nameko this can be implemented using entrypoint mechanism:

# 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")

Writing extensions

By writing extensions, you can customize Nameko's behavior to suit your specific needs.

Let's say you need to add an authentication mechanism to Nameko services. You can write an extension that will handle authentication of requests to services:

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

Or, for example, we want to add performance monitoring to Nameko services. You can write an extension that will collect statistics about query execution time and other metrics:

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)

After writing the extension, you can use it in your services. To do this, simply add it to the Nameko container:

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)
        # логика метода

---

The article was prepared as part of the course recruitment Python Developer. Professional. Follow the link to learn more about the course, as well as register for a free webinar.