After years of ruby and javascript I’ve started doing some work with python recently after joining the team at Agero. One of the first things I noticed was that a lot of folks in the python world pass around a lot of named arguments as a convention. My first example of this is the boto3 library used to interact with AWS in python (for proof, click here)

In an effort to reduce some of this parameter passing folks might be tempted to make their own class that reduces the number of parameters one needs to pass around. I offer a simpler functional solution taken from some patterns I’ve seen used in the functional javascript community.

Enter partial function application with a builder pattern. The following code highlights how one might create reusable functions without creating a domain specific class hierarchy.

Take the boto3 library. More specifically, take the scan function for a dyanmodb table. The following is an example of how one could use this concept to produce a convenient way to iterate through all items in a dynamo collection.

# ./function_builder.py
import functools

class FunctionBuilder:

    def __init__(self, func):
       self.func = func

    def having(self, *args, **kwargs):
        self.func = functools.partial(self.func, *args, **kwargs)
        return self

    def get(self):
        return self.func

# ./dynamo_iterator.py
class DynamoIterator:
    
    def __init__(self, func):
        """Given dynamodb function that returns items and a LastEvaluatedKey iterate
    
        Keyword arguments:
        func -- function that takes LastEvaluatedKey and returns { items:  [] }
        """
        self.func = func
        self.last_evaluated_key = None
        self.first_fetch = True

    def _get_items(self, db_result):
        return db_result.get('Items', [])

    def _has_more_results(self):
        return self.last_evaluated_key and len(self.data) == 0

    def __iter__(self):
        """Required to implement the Iterator interface in python"""
        return self

    def __next__(self):
        """Required to implement the Iterator interface in python"""
        if self.first_fetch or self._has_more_results():
            self.first_fetch = False
            result = self.func(LastEvaluatedKey=self.last_evaluated_key)
            self.last_evaluated_key = result.get('LastEvaluatedKey', None)
            self.data = self._get_items(result)

        if len(self.data) <= 0:
            raise StopIteration

        return self.data.pop()

    """Python 2.7 requireds next vs 3.7 that requires __next__ hence we alias the function"""
    next=__next__

Now that we have these two simple base classes we can iterate through a collection of dynamodb objects using the following code.

# ./main.py
import boto3
from .dynamo_iterator import DynamoIterator
from .function_builder import FunctionBuilder

dynamodb = boto3.resource('dynamodb')
table = self.dynamodb.Table('my-table-name')

partial_scan_all = FunctionBuilder(table.scan).having(Limit=100, Select='ALL_ATTRIBUTES').get() 

# this will iterate through every item in a dynamodb table.  Obviously not meant for quickly returning values to a user interface.
for item in DynamoIterator(partial_scan_all):
    print(item)

Hopefully you found this code example useful. If you have any questions please leave a comment and I’ll try to help with an answer.