Building Python Microservices with Nameko

Overview

When I worked in Student.com, we used a microservice archtecture for our backend. The system was divided into several services, such as user-service, order-service, landlord-serivce, property-service, etc. We used Nameko to handle commnication between services.

Simple Example

In nameko, you use @rpc entrypoint to define APIs that other service can call, and RpcProxy to call APIs on other services. Nameko is an asynchronous messaging system, based on AMQP (RabbitMQ), and also provides @event_handler, EventDispatcher to for pub-sub messaging.

The following example shows how the order-service gets user, product info when creating an order, and how it sends an asynchronous message to message-service to notify the user by SMS.

from nameko.rpc import rpc, RpcProxy
from nameko.events import EventDispatcher, event_handler

class UserSerivce:
    name = "user_service":
    @rpc
    def get_user(self, user_id):
      return {"id": user_id, "name": "Sherman", "mobile": ""}

class GoodService:
    name = "good_service":
    @rpc
    def get_good(self, good_id):
      return {"id": good_id, "name", "name": "Milk", "price": 5.99}

class MessageService:
    name = "message_service":
    @event_handler("order_service", "order_created")
    def handle_order_create(self, payload):
        print("message service received:", payload)
        # send sms to user


class OrderService:
    name = "order_service"

    user = RpcProxy("user_service")
    good = RpcProxy("good_service")
    dispatch = EventDispatcher()

    @rpc
    def create_order(self, user_id, good_id):
        user = self.user.get_user(user_id)
        mobile = user["mobile"]
        good = self.good.get_good(good_id)

        order = {
            "id": 1,
            "user_id": user_id,
            "status": "created",
            "mobile": user["mobile"],
            "good_id": good_id,
            "good_name": good["name"],
            "good_price": good["price"]
        }
        self.dispatch("order_created", order)
        return order

AMQP

AMQP stands for Advanced Message Queuing Protocol. It’s a messaging protocol that allows services to communicate by sending messages to a broker (like RabbitMQ) instead of talking directly to each other. The broker delivers messages to the correct consumers, making services decoupled and asynchronous.

How Nameko Works

In a typical setup, one service publishes a message while one or more consumers listen. But how does the consumer return a response to the publisher? For example, in the code above, how does order-service get user info from user-service?

RPC Mode in RabbitMQ

RabbitMQ supports a RPC pattern, which looks likt this:

1. Client sends a message to a request queue

• Includes two properties in the message:
  • reply_to: a temporary queue where the response should be sent
  • correlation_id: a unique identifier for the request

2. Server consumers messages from the request queue

• Processes the request
• Sends the response back to the queue specified in reply_to with the same correlation_id

3. Clients listens on the reply_to queue

• When a message with a matching correlation_id arrives, it knows this is the response

How Nameko Implements RPC

Nameko wraps the RPC pattern of RabbitMQ. When order_service calls user = self.user.get_user(user_id), a temporary_queue is created to wait for the response. The message sent to user_service includes the reply_to queue name and a correlation_id. When user_service receives the message, it calls the appropriate method and sends the result back. order_service listens on the temporary queue, and once the response arrives, it stores the result in the user variable.

But how does user_service know which method to call? For the code:

class UserSerivce:
    name = "user_service":
    @rpc
    def get_user(self, user_id):
      return {"id": user_id, "name": "Sherman", "mobile": ""}

When user_service starts, Nameko registers all methods decorated with @rpc in a lookup table:

{
  "get_user": UserService.get_user,
}

The message sent by order_service includes the service name, method name, and arguments. Nameko unpacks the payload, finds get_user in the lookup table, and calls it with the provided arguments.

Internal Details

From a source-code perspective, Nameko works like this:

1. @rpc decorator

• Adds a special attribute (ENTRYPOINT_EXTENSIONS_ATTR) to mark the method as an RPC entrypoint

2. Service discovery

• When nameko run starts, Nameko inspects all service classes.
• Using inspect, it finds methods with ENTRYPOINT_EXTENSIONS_ATTR into the entrypoints set

3. RpcConsumer setup

• Each Rpc entrypoint is registered with an RpcConsumer
• The RpcConsumer maintains a provider mapping: {"get_user": UserService.get_user}

4. Starting consumers

• Each entrypoint’s start method is called
• This sets up a RabbitMQ consumer (via Kombu) and registers a callback function
• The consumer runs in a green thread, continuously listening for new messages

5. Handling a message

• When RabbitMQ delivers a message, Kombu runs the callback function to handle it
• The callback uses the routing_key (service + method name) to find the right Rpc object
• It calls Rpc.handle_message, which pushes the request into an eventlet GreenPool
• A worker green thread runs the actual method

6. Sending the response

• The return value is sent back with the same correlation_id to the client’s temporary queue

Namekoman

Debugging Nameko services can be tricky because you normally have to call services from the terminal, which is cumbersome for large payloads:

$ nameko shell
>>> n.rpc.order_service.update_user(user_id=1, first_name="", last_name="", address={})

To simplify this, I created Namekoman a GUI test tool for testing Nameko rpc calls. It allows you to input parameters in JSON, view logs from the service, and save calls for later reference.

Here's a screenshot of Namekoman: