Consuming AMQP/RabbitMQ messages in Python is simple if you use Sparkplug. I read “Building RabbitMQ apps using Python“, and there’s just too much code in the monitor method and the separation of concerns aren’t very clear.

With Sparkplug, you create a configuration file, then define Consumer class to handle each message. As messages are pushed into the queue, Sparkplug will call your class and you decide how you want handle the message in the __call__ method. The great thing about this is that all the connection logic is removed from your code letting you concentrate on how you want to handle each message.

Now you can run your app using sparkplug --daemon consumer.ini. Configuration is boring, so I’ve left the consumer.ini out of the blog post and put it up on github instead.

Jesse has written a great blog post on using test builders to improve the expressiveness of unit tests. He has convinced me that builders are useful and that they provide an easy way to construct objects that are in a valid state. Testing can be hard; I find defining a method for each of the object’s properties to be tedious and repetitive. This is especially problematic if you’re working with an object with dozens of properties.

To balance expressive with conciseness, I’ve refactored the builder to use Python dictionaries

I find this code to be just as expressive, easier to use and most importantly not repetitive.

My previous discussion on Inversion of Control raised some questions so I want to take a step back and discuss Dependency Injection. Dependency Injection is a pattern where software components (classes, methods or functions) are given their dependencies as parameters and these software components do not instantiate their dependencies directly.

Dependency Injection is an important pattern to use when you wan to create classes that are easier to reuse and unit test. Since the dependencies are external, they can be configured and maintained outside class or function and there’s no need to change the code in order to reuse it. Also, when dependencies are injected into a class or a function it is possible to substitute a mock implementation of the dependency. In unit tests, mock objects are used as replacement for the real implementation, this helps isolate the functionality being tested.

A simple example

The following function has it’s dependencies, a list of numbers, passed to it, and since calculate_average only talks to a list of numbers it is easy to reuse.

def calculate_average(list_of_numbers):
  x = sum(list_of_numbers)
  return x/len(list_of_numbers)

Imagine a similar function which uses a database connection, retrieves a list of numbers then calculates the average.

def calculate_average():
  sql_query = "select numbers from table"
  list_of_numbers = db.query(sql_query)
  x = sum(list_of_numbers)
  return x/len(list_of_numbers)

This function is difficult to reuse because it depends on the database, future users of this function will have to either modify the function or create and maintain a database just to use it. More importantly it’s difficult to test this function. A test would require a database connection and a fixture to populate the database with data.

You should be thinking, but Shey, I would never do this with a simple function like this. You’re right, you wouldn’t do this with a simple function, but as soon as calculate_average turns into a more complicated calculation such as calculate_value_at_risk we start instantiating or dependencies.

A slightly less contrived example

The BillingService class is responsible for charging your customers a fee.

class BillingService
  def initialize(credit_card_processor)
    self.processor = processor

  def charge(account, amount):
    if account.has_balance:
      result = processor.charge(account.cc_number, amount)
    if result == "success":
      account.deduct(amount)

Lets assume that Authorize.net is offering better rates than PayPal. If the class accepts an implementation of a credit card processor as a dependency it is possible to switch to the authorize.net implementation of the credit card processor class without changing the BillingService class. Similarly during unit testing, a mock implementation of a credit card processor can be used to alwasy return “success” and isolate the charge functionality of the BillingService.

class mock_credit_card_processor:
  def charge(cc_number, amount):
    return "success"

    def test_charge():
      bs = BillingService(mock_credit_card_processor)
      a = Account(cc_number="411111111111111", balance=300.00)
      bs.charge(a, 300.00)
      assert(a.balance == 0)