DEV Community: Eric Cheatham

Learning from over-engineering

Eric Cheatham — Tue, 31 Oct 2023 19:58:13 +0000

We can gain incredible insights into from our success as well as our failures in software engineering. Yet it is often overlooked how sharing our failures can help us understand why our specific solution failed and help others avoid the same pitfalls.

I’d like to take a moment to explore a time where I over-engineered myself in to a solution that, while novel, ended up being far too difficult to maintain in comparison to the value the code itself delivered.

The Challenge

I was tasked with creating a python package that would serve as the middleware between a client application and an existing REST API. The package was expected to, at the bare minimum, be able to do the following:

Make async REST requests against a given API endpoint
Unmarshal the responses from the API and be able to process that the response
Return the response according to the expectations of the client application
Handle any errors that may occur

Proposed Solution

I knew from the very beginning that my solution needed to be as DRY as possible. In this case I knew that the endpoints of the API I was working with were all constructed very similar to one another; for example:

.../resource/
.../users/
.../blogs/
...

The same can be said for the responses, again, for example:

{
    "uuid": "94cccdc9-3564-434f-9b07-a02dc1edb353",
    "name": "object_name",
    "tags": ["python", "code_example"],
    "meta": {"state": "IA", "details": ""},
}

I also wanted my solution to produce responses that could be introspected. This meant that would need to create a unique Class for reach response that I could receive so that consumers would be able to know or find out members of those classes.

As you might imagine that could be difficult to do with N number of possible classes and only one function making the requests. Instead of opting for conditionals or flags I chose to create a generic Python Decorator that would decorate member methods of a given Class and make the requests on that class’s behalf.

The Classes ended up taking the following structure.

class ExampleResponse(ProcessResponse):
    def __init__(
        self, 
        id: str,
        some_field: str,
    ) -> None:
        self.id = id
        self.some_field = some_field

class Example:
    def __init__(self, session) -> None:
        # async request setup stuff

    @process(ExampleResponse)
    async def get(self, id: str):
        async with self.session.get(...) as resp:
            return await resp.text()

Each class would have to have its own response class it could unmarshal into and each member method was decorated with @process() that takes the destination object as its only parameter. Each endpoint needed its own class and member methods all doing very similar things.

That’s not very DRY.

The DRYness of my code was achieved in the decorator itself. All the unmarshalling and processing happened here.

class ProcessResponse(object):
    def __init__(self, *args, **kwargs):
        ...

def parse_error_message(error):
    ... # Read the error string in and turn it into a python error obj

def process(return_type: Type[ProcessResponse]):
    def inner(func):
        async def wrapper(*args, **kwargs):
            res = await func(*args, **kwargs)
            try:
                parsed = json.loads(res)
                if isinstance(parsed, list):
                    return None, [return_type(**par) for par in parsed]
                return None, return_type(**parsed)
            except (JSONDecodeError, TypeError):
                return parse_error_message(res), None
        return wrapper
    return inner

The decorator module defined a parent class that all responses had to inherit from and returned either an object of type return_type or a Python error.

Challenges Faced

This approach had two very clear weaknesses that became challenges in their own right:

It was overly complicated
It was far too rigid to be maintainable

First, the excessive complexity. It was hard to trace errors though this code when something broke. In just these few lines of code we:

Decode a JSON string
Attempt to unpack that decoded string (twice!)
Assume that if the decoded string could not be unpacked that meant it was an error and that we should try to force it into an error of some form

Not to mention that decorators are not the most inherently easy to understand structures by themselves. So I mixed in a little async for fun. While it made sense to me, I found it difficult to explain to others if they didn’t already have a strong understanding of decorators and other Python behavioral traits.

Secondly, what happens when the response from the API changes? For example, our ExampleResponse class expects a JSON response that looks like

{
    "id": "123",
    "some_field": "some cool string",
}

If we added, renamed, or otherwise modified this response without also modifying our ExampleResponse class we would face JSON parsing errors as we attempt to unpack the decoded response into our classes.

This was a large challenge for this library as it was being developed against an API that would change with some high level of frequency.

Potential Improvements

The initial design, while well-intentioned, fell short of its potential due to its complexity and rigidity. To prevent falling into similar traps in future projects, the following considerations are essential:

Decorators:
- Carefully assess the necessity of using decorators. While they can be powerful, their excessive use can lead to convoluted code.
Generalization vs. Clarity:
- Strive for generalized logic, but not at the expense of code clarity. Code should be easy to understand and troubleshoot. Keep complexity in check.
Leverage Python's Standard Library:
- Explore the capabilities of Python's Standard Library thoroughly before diving into complex solutions. Often, you'll find that Python provides built-in tools to simplify tasks.
  - For example, Python offers SimpleNamespace, a versatile data structure that, when combined with json.loads, allows you to load JSON into an object that supports dot notation, making it easy to access fields like example.some_field.

Conclusion

While this solution initially served its purpose, it soon became apparent that it wasn't a sustainable, long-term answer. In retrospect, a more effective strategy would have involved thorough pre-planning and research, leading to the discovery of potentially simpler and more maintainable solutions. This approach would have spared me the need for frequent rewrites and the complexities of debugging sessions.

Learning from over-engineering

Eric Cheatham — Tue, 31 Oct 2023 19:58:13 +0000

The Challenge

Make async REST requests against a given API endpoint
Unmarshal the responses from the API and be able to process that the response
Return the response according to the expectations of the client application
Handle any errors that may occur

Proposed Solution

.../resource/
.../users/
.../blogs/
...

The same can be said for the responses, again, for example:

{
    "uuid": "94cccdc9-3564-434f-9b07-a02dc1edb353",
    "name": "object_name",
    "tags": ["python", "code_example"],
    "meta": {"state": "IA", "details": ""},
}

The Classes ended up taking the following structure.

class ExampleResponse(ProcessResponse):
    def __init__(
        self, 
        id: str,
        some_field: str,
    ) -> None:
        self.id = id
        self.some_field = some_field

class Example:
    def __init__(self, session) -> None:
        # async request setup stuff

    @process(ExampleResponse)
    async def get(self, id: str):
        async with self.session.get(...) as resp:
            return await resp.text()

That’s not very DRY.

The DRYness of my code was achieved in the decorator itself. All the unmarshalling and processing happened here.

class ProcessResponse(object):
    def __init__(self, *args, **kwargs):
        ...

def parse_error_message(error):
    ... # Read the error string in and turn it into a python error obj

def process(return_type: Type[ProcessResponse]):
    def inner(func):
        async def wrapper(*args, **kwargs):
            res = await func(*args, **kwargs)
            try:
                parsed = json.loads(res)
                if isinstance(parsed, list):
                    return None, [return_type(**par) for par in parsed]
                return None, return_type(**parsed)
            except (JSONDecodeError, TypeError):
                return parse_error_message(res), None
        return wrapper
    return inner

The decorator module defined a parent class that all responses had to inherit from and returned either an object of type return_type or a Python error.

Challenges Faced

This approach had two very clear weaknesses that became challenges in their own right:

It was overly complicated
It was far too rigid to be maintainable

First, the excessive complexity. It was hard to trace errors though this code when something broke. In just these few lines of code we:

Decode a JSON string
Attempt to unpack that decoded string (twice!)
Assume that if the decoded string could not be unpacked that meant it was an error and that we should try to force it into an error of some form

Secondly, what happens when the response from the API changes? For example, our ExampleResponse class expects a JSON response that looks like

{
    "id": "123",
    "some_field": "some cool string",
}

This was a large challenge for this library as it was being developed against an API that would change with some high level of frequency.

Potential Improvements

Decorators:
- Carefully assess the necessity of using decorators. While they can be powerful, their excessive use can lead to convoluted code.
Generalization vs. Clarity:
- Strive for generalized logic, but not at the expense of code clarity. Code should be easy to understand and troubleshoot. Keep complexity in check.
Leverage Python's Standard Library:
- Explore the capabilities of Python's Standard Library thoroughly before diving into complex solutions. Often, you'll find that Python provides built-in tools to simplify tasks.
  - For example, Python offers SimpleNamespace, a versatile data structure that, when combined with json.loads, allows you to load JSON into an object that supports dot notation, making it easy to access fields like example.some_field.

Conclusion

Consuming AMQP Messages using Alpakka

Eric Cheatham — Tue, 03 Dec 2019 03:08:54 +0000

Alpakka is a library written in both Scala and Java that provides a way to implement stream-aware pipelines for streaming data from one source to another.

One such source that Alpakka supports is from AMQP (Advanced Message Queuing Protocol) servers. Alpakka allows a consumer to treat an AMQP server as either a source (origin) or a sink (destination).

We will be exploring an example of using an AMQP server as a source as well as exploring another of Alpakka's features, using a relational database as the sink.

Overview

At a high level our example will be doing the following steps:

Declaring and attaching to an AMQP queue
Creating a consumer to consume messages off previously declared queue
Transforming the data into something that can be further manipulated
Writing our data to our database

In pictures, our steps would look a little something like this:

This may seem fairly complicated at first, however, Alpakka handles a lot of the heavy lifting for us and makes setting all this up a breeze!

Declaring a Queue and Creating a Consumer

Our first two steps go hand-in-hand; declaring our queue and creating our consumer.

We will begin by choosing a name for our queue and a connection provider for our queue. For this example our queue will be named amqp-test-queue-example and we will be going with AmqpLocalConnectionProvider. Our choice of connection provider will handle connecting to a local instance of an AMQP server for us. Alpakka provides several other connection providers to handle connecting to non-local servers.

val queueName = "amqp-test-queue-example"
val queueDeclaration = QueueDeclaration(queueName)
val connectionProvider = AmqpLocalConnectionProvider

object MessageQueue {
  def createQueue(connectionProvider: AmqpConnectionProvider,
                  queueName: String,
                  queueDeclaration: Declaration,
                  bufferSize: Int = 10): Source[ReadResult, NotUsed] =
    AmqpSource.atMostOnceSource(
      NamedQueueSourceSettings(connectionProvider, queueName)
        .withDeclaration(queueDeclaration)
        .withAckRequired(true),
      bufferSize = bufferSize
    )
}

It is also worth noting in our above code that we have limited our queue's buffer size to 10 messages. This will limit the number of messages to pre-fetch from the AMQP server.

We are also telling our consumer to ACK all messages that are consumed by using the withAckRequired() method on the NamedQueueSourceSettings class. More complex models allow for custom ACKing/NAKing policies. To learn more about that visit the Alkpakka article about stream graphs

Transforming our Data

Now that we have a consumer that will take consume messages from an AMQP server and ACK those messages we can begin to use the contents of those messages. One hang up though: all our messages will be coming in the form of byte arrays! No worries, we can make use of JSON4s to unmarshall or convert our message from the form that was transmitted to a form we can use.

import org.json4s.native.Serialization.read

case class CustomerCreated(name: String, email: String)

val result =
    MessageQueue.createQueue(connectionProvider, queueName, queueDeclaration, bufferSize)
        .take(bufferSize)
        .map({ x => read[CustomerCreated](x.bytes.utf8String) })
        .log("Received message", x ⇒ println(x))

In our example we are:

Creating our consumer using MessageQueue.createQueue()
Telling our consumer to take up to our buffer size of 10 messages
Iterating over the messages that we get and letting JSON4s convert them into our CustomerCreated case class
Logging out the result of each message

Writing to a Sink

At this point JSON4s has done us a huge favor and converted our message into something we can do some work with. For our purposes we will be writing our data to a PostgreSQL database. A look into our database structure is seen below.

 Column |          Type          | Collation | Nullable |                Default
--------+------------------------+-----------+----------+---------------------------------------
 id     | integer                |           | not null | nextval('customers_id_seq'::regclass)
 name   | character varying(100) |           |          |
 email  | character varying(100) |           |          |

To write to a SQL based database we will have to make use of another of Alpakka's many connectors: Slick. Slick handles connecting to and writing to a database of our choosing.

It's honestly pretty slick.

Slick requires you to provide a configuration file that describes how to connect to a database. For our example we'll be using the following:

slick-postgres {
  profile = "slick.jdbc.PostgresProfile$"

  db = {
    password = "password"
    user = "example"
    url = "jdbc:postgresql://127.0.0.1/exampledb"
  }
}

Going back to our created consumer, we can now tell our consumer where to write our consumed and transformed data. We will do so by telling slick to use our configuration file and writing each item to our sink as sql inserts.

import akka.stream.alpakka.slick.javadsl.SlickSession
import akka.stream.alpakka.slick.scaladsl.Slick

case class CustomerCreated(name: String, email: String)

MessageQueue.createQueue(connectionProvider, queueName, queueDeclaration, bufferSize)
    .take(bufferSize)
    .map({ x => read[CustomerCreated](x.bytes.utf8String) })
    .log("Received message", x ⇒ println(x))
    .runWith(
        Slick.sink({ x ⇒ sqlu"INSERT INTO customers (name, email) VALUES(${x.name}, ${x.email})" })
    )

Wrapping our database sink in a runWith() allows us to materialize our flow. Until now we have merely been describing what we intend to do. Materializing our consumer tells Alpakka to allocate all the necessary resources in order to run our flow. To further explore stream materialization visit the Alpakka article on Stream Materailzation

Seeing this all work

All the code that we have constructed above can be found on my github repository. The README.md will walk you through the process of starting up a local RabbitMq instance (our AMQP server) as well as a Postgres database (our relational database sink).

ericcheatham / event-streaming-example

Scala Event Streaming App Example

In conclusion we explored one of many ways to use Alpakka to consume asynchronous messages from an AMQP source. Alpakka provides connectors for a vast many more connectors and functions aside from what we've explored here. No matter the situation you are approaching, Alpakka has it covered in their incredible documenation

If you ever want to talk Scala or have really spicy takes about food, send me a message over on Twitter or LinkedIn