introduced persistence objects

freqtrade/aws/service/Persistence.py

@@ -0,0 +1,53 @@
+import boto3
+import simplejson as json
+
+
+class Persistence:
+    """
+        simplistic persistence framework
+    """
+
+    def __init__(self, table):
+        """
+            creates a new object with the associated table
+        :param table:
+        """
+
+        self.table = table
+        self.db = boto3.resource('dynamodb')
+
+    def load(self, sample):
+        """
+            loads a given object from the database storage
+        :param sample:
+        :return:
+        """
+
+        table = self.db.Table(self.table)
+        result = table.get_item(
+            Key={
+                'id': sample
+            }
+        )
+
+        if 'Item' in result:
+            return result['Item']
+        else:
+            return None
+
+    def save(self, object):
+        """
+
+        saves and object to the database storage with the specific key
+
+        :param object:
+        :return:
+        """
+
+        table = self.db.Table(self.table)
+
+        # force serialization to deal with decimal number tag
+        data = json.dumps(object, use_decimal=True)
+        data = json.loads(data, use_decimal=True)
+        print(data)
+        return table.put_item(Item=data)

freqtrade/aws/service/Queue.py

@@ -0,0 +1,47 @@
+import simplejson as json
+import os
+import boto3
+
+
+class Queue:
+    """
+    abstraction of the underlaying queuing system to schedule a message to the backend for processing
+    """
+
+    def submit(self, object, routingKey):
+        """
+            submits the given object to the queue associated with the
+            routing key.
+            The routing lambda function will than make sure it will be delivered to the right destination
+
+        :param object:
+        :param routingKey:
+        :return:
+        """
+
+        # get topic refrence
+        client = boto3.client('sns')
+
+        # if topic exists, we just reuse it
+        topic_arn = client.create_topic(Name=os.environ['topic'])['TopicArn']
+
+        serialized = json.dumps(object, use_decimal=True)
+        # submit item to queue for routing to the correct persistence
+
+        result = client.publish(
+            TopicArn=topic_arn,
+            Message=json.dumps({'default': serialized}),
+            Subject="route:" + routingKey,
+            MessageStructure='json',
+            MessageAttributes={
+                'route': {
+                    'DataType': 'String',
+                    'StringValue': routingKey
+                }
+            },
+        )
+
+        return {
+            "statusCode": result['ResponseMetadata']['HTTPStatusCode'],
+            "body": serialized
+        }