Distributed Lock Manager using DynamoDB and Python

dynamo_lock.py

"""
Distributed Lock Manager using DynamoDB and Python
Inspired by: https://aws.amazon.com/blogs/database/building-distributed-locks-with-the-dynamodb-lock-client/
Written for Python 3.6 on macOS. Your milage may vary.
Requirements (pip install):
    pynamdodb
    pynamdodb[signals]
    boto3
"""
import os
import unittest
import inspect
from datetime import datetime, timezone, timedelta
from pynamodb.models import Model
from pynamodb.attributes import (
    UnicodeAttribute, NumberAttribute, UnicodeSetAttribute, UTCDateTimeAttribute, BinaryAttribute
)

# Set the required AWS credentials 
os.environ['AWS_ACCESS_KEY_ID'] = 'AWS_ACCESS_KEY_ID'
os.environ['AWS_SECRET_ACCESS_KEY'] = 'AWS_SECRET_ACCESS_KEY'
os.environ['AWS_DEFAULT_REGION'] = 'us-east-1'


class LockTable(Model):

    lock_name = UnicodeAttribute(hash_key=True)
    creation_time = UTCDateTimeAttribute(range_key=True, default=lambda: datetime.now(timezone.utc))
    expiration_time = UTCDateTimeAttribute()
    version_number = NumberAttribute(default=1)
    is_released = BinaryAttribute(default=False)

    @staticmethod
    def get_future_datetime(minutes_into_future):
        current_datetime = datetime.now()
        time_delta_mintues = timedelta(minutes=minutes_into_future)
        return current_datetime + time_delta_mintues

    class Meta:
        table_name = 'Lock'
        region = 'us-east-1'
        host = 'http://localhost:8000'
        write_capacity_units = 1
        read_capacity_units = 1


class LockManager(object):
    
    def __init__(self):
        if not LockTable.exists():
            LockTable.create_table(wait=True)
        self.lock = LockTable()
    
    def describe(self, lock_name):
        pass

    def acquire(self, lock_name, wait=False):
        pass

    def renew(self, lock_name):
        pass

    def release(self, lock_name):
        pass

    
class LockManagerTests(unittest.TestCase):

    def setUp(self):
        self.lock_manager = LockManager()

    def tearDown(self):
        pass

    def test_LockTable_class_exists(self):
        self.assertTrue(inspect.isclass(LockTable))
    
    def test_LockTable_class_properties_exists(self):
        self.assertTrue(isinstance(LockTable.lock_name, UnicodeAttribute))
        self.assertTrue(isinstance(LockTable.creation_time, UTCDateTimeAttribute))
        self.assertTrue(isinstance(LockTable.expiration_time, UTCDateTimeAttribute))
        self.assertTrue(isinstance(LockTable.version_number, NumberAttribute))
        self.assertTrue(isinstance(LockTable.is_released, BinaryAttribute))

    def test_LockTable_class_methods_exists(self):
        self.assertTrue(inspect.ismethod(LockTable.get_future_datetime))

    def test_LockManager_class_exists(self):
        self.assertTrue(inspect.isclass(LockManager))
        #self.assertTrue(inspect.ismethod(LockManager.describe))
        #self.assertTrue(inspect.ismethod(LockManager.acquire))
        #self.assertTrue(inspect.ismethod(LockManager.renew))
        #self.assertTrue(inspect.ismethod(LockManager.release))


if __name__ == '__main__':
    unittest.main()