DynamoDB locally with Docker

Author: Lucas Belo

README.md

@@ -6,4 +6,23 @@ The easiest way to manage servers, subnets, load balancers, managed databases, e
 
 ## Services
 
-Anything that is long-running or talks to our end users is considered a service. It can be monolithic Django application, Vue UI, or a couple of microservice APIs -- all will reside in this folder.
+Anything that is long-running or talks to our end users is considered a service. It can be monolithic Django application, Vue UI, or a couple of microservice APIs -- all will reside in this folder.
+
+## Running code-quality locally
+
+Run the code-quality tools before commiting code. There's a validation in the CI/CD that might brake.
+
+```bash
+$ poetry run black .
+$ poetry run isort . --profile black
+$ poetry run flake8 .
+```
+
+## Running tests locally
+
+You need to set AWS_DEFAULT_REGION environment variable since Boto3 requires it to be set. Then, run the tests to see the new one fail:
+
+``` bash
+$ export AWS_DEFAULT_REGION=eu-west-1
+$ poetry run pytest tests.py
+```

services/tasks_api/docker-compose.yml

@@ -0,0 +1,12 @@
+version: '3.8'
+
+services:
+  dynamodb-local:
+    command: "-jar DynamoDBLocal.jar -sharedDb -dbPath ./data"
+    image: "amazon/dynamodb-local:latest"
+    container_name: dynamodb-local
+    ports:
+      - "9999:8000"
+    volumes:
+      - "./docker/dynamodb:/home/dynamodblocal/data"
+    working_dir: /home/dynamodblocal

services/tasks_api/store.py

@@ -8,11 +8,12 @@
 
 
 class TaskStore:
-    def __init__(self, table_name):
+    def __init__(self, table_name, dynamodb_url=None):
         self.table_name = table_name
+        self.dynamodb_url = dynamodb_url
 
     def add(self, task):
-        dynamodb = boto3.resource("dynamodb")
+        dynamodb = boto3.resource("dynamodb", endpoint_url=self.dynamodb_url)
         table = dynamodb.Table(self.table_name)
         table.put_item(
             Item={
@@ -28,7 +29,7 @@ def add(self, task):
         )
 
     def get_by_id(self, task_id, owner):
-        dynamodb = boto3.resource("dynamodb")
+        dynamodb = boto3.resource("dynamodb", endpoint_url=self.dynamodb_url)
         table = dynamodb.Table(self.table_name)
         record = table.get_item(
             Key={
@@ -50,7 +51,7 @@ def list_closed(self, owner):
         return self._list_by_status(owner, TaskStatus.CLOSED)
 
     def _list_by_status(self, owner, status):
-        dynamodb = boto3.resource("dynamodb")
+        dynamodb = boto3.resource("dynamodb", endpoint_url=self.dynamodb_url)
         table = dynamodb.Table(self.table_name)
         last_key = None
         query_kwargs = {