refactor: project structure and oop

Author: srgchrksv

create-synth-data/create_collection.py

@@ -5,7 +5,7 @@
 import os 
 from dotenv import load_dotenv
 
-load_dotenv()
+load_dotenv('../.env')
 # get database name
 FIRESTORE_DATABASE = os.environ.get('FIRESTORE_DATABASE')
 
@@ -49,7 +49,7 @@ def populate_firestore(rows):
     for _ in range(rows):
         product = generate_random_product()
         collection_ref.document(product["_id"]).set(product)
-    print("Inserted {rows} random products into Firestore.")
+    print(f"Inserted {rows} random products into Firestore.")
 
 
 populate_firestore(100)

create-synth-data/create_dynamo_table.py

@@ -20,7 +20,7 @@
 dynamodb = session.resource('dynamodb')
 
 # Define the table schema
-table_name = 'Products'
+table_name = os.environ.get('DYNAMODB_TABLE')
 attribute_definitions = [
     {'AttributeName': '_id', 'AttributeType': 'S'},  # Partition key
     {'AttributeName': 'category', 'AttributeType': 'S'}  # Sort key (optional)

dynamodb_firestore.py

@@ -0,0 +1,44 @@
+from migration import source, destination, migrate
+from google.cloud import firestore
+import boto3
+from dotenv import load_dotenv
+import os
+import logging
+import sys
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
+    handlers=[
+        logging.StreamHandler(sys.stdout)  # Specify sys.stdout for standard output
+    ],
+)
+
+load_dotenv()
+
+if __name__ == "__main__":
+
+    BATCH_SIZE = os.environ.get("BATCH_SIZE")
+    # Set up Firestore and Dynamodb configurations
+    FIRESTORE_DATABASE = os.environ.get("FIRESTORE_DATABASE")
+    FIRESTORE_COLLECTION = os.environ.get("FIRESTORE_COLLECTION")
+    DYNAMODB_TABLE = os.environ.get("DYNAMODB_TABLE")
+
+    # Initialize Firestore
+    firestore_client = firestore.Client(database=FIRESTORE_DATABASE)
+    firestore_collection = firestore_client.collection(FIRESTORE_COLLECTION)
+
+    # Initialize DynamoDB
+    session = boto3.Session(
+        aws_access_key_id=os.environ.get("AWS_ACCESS_KEY_ID"),
+        aws_secret_access_key=os.environ.get("AWS_SECRET_ACCESS_KEY"),
+        region_name=os.environ.get("AWS_REGION"),
+    )
+    dynamodb = session.resource("dynamodb")
+    table = dynamodb.Table(DYNAMODB_TABLE)
+
+    source_dynamodb = source.DynamoDB(table, BATCH_SIZE)
+    destination_firebase = destination.Firebase(firestore_client, firestore_collection)
+
+    migrate = migrate.DatabaseMigration(source_dynamodb, destination_firebase)
+    migrate.run()

dynamodb_to_firebase.py

@@ -0,0 +1,44 @@
+from migration import source, destination, migrate
+from google.cloud import firestore
+import boto3
+from dotenv import load_dotenv
+import os
+import logging
+import sys
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
+    handlers=[
+        logging.StreamHandler(sys.stdout)  # Specify sys.stdout for standard output
+    ],
+)
+
+load_dotenv()
+
+if __name__ == "__main__":
+
+    BATCH_SIZE = os.environ.get("BATCH_SIZE")
+    # Set up Firestore and Dynamodb configurations
+    FIRESTORE_DATABASE = os.environ.get("FIRESTORE_DATABASE")
+    FIRESTORE_COLLECTION = os.environ.get("FIRESTORE_COLLECTION")
+    DYNAMODB_TABLE = os.environ.get("DYNAMODB_TABLE")
+
+    # Initialize Firestore
+    firestore_client = firestore.Client(database=FIRESTORE_DATABASE)
+    firestore_collection = firestore_client.collection(FIRESTORE_COLLECTION)
+
+    # Initialize DynamoDB
+    session = boto3.Session(
+        aws_access_key_id=os.environ.get("AWS_ACCESS_KEY_ID"),
+        aws_secret_access_key=os.environ.get("AWS_SECRET_ACCESS_KEY"),
+        region_name=os.environ.get("AWS_REGION"),
+    )
+    dynamodb = session.resource("dynamodb")
+    table = dynamodb.Table(DYNAMODB_TABLE)
+
+    source_firebase = source.Firebase(firestore_collection, BATCH_SIZE)
+    destination_dynamodb = destination.DynamoDB(table)
+
+    migrate = migrate.DatabaseMigration(source_firebase, destination_dynamodb)
+    migrate.run()

firebase_to_dynamodb.py

@@ -0,0 +1,47 @@
+from abc import ABC, abstractmethod
+import logging
+from migration.utils.convert import floats_to_decimals, decimals_to_floats
+
+logger = logging.getLogger(__name__)
+
+
+class Loader(ABC):
+    @abstractmethod
+    def load(self, data):
+        pass
+
+
+class Firebase(Loader):
+    def __init__(self, client, collection) -> None:
+        self.client = client
+        self.collection = collection
+
+    def load(self, data):
+        """Load data into Firestore in batches"""
+        logger.info(f"Loading to Firestore")
+
+        batch = self.client.batch()
+        data = decimals_to_floats(data)
+
+        for item in data:
+            doc_ref = self.colletion.document(item.get("_id"))
+            batch.set(doc_ref, item)
+
+        batch.commit()
+        logger.info(f"Loaded {len(data)} documents")
+
+
+class DynamoDB(Loader):
+    def __init__(self, table) -> None:
+        self.table = table
+
+
+    def load(self, data):
+        """Load data into DynamoDB in batches"""
+        logger.info(f"Loading to dynamodb")
+
+        data = floats_to_decimals(data)
+        with self.table.batch_writer() as batch:
+            for item in data:
+                batch.put_item(Item=item)
+        logger.info(f"Loaded {len(data)} documents")

firestore_dynamodb.py

@@ -0,0 +1,27 @@
+import logging
+import traceback
+
+from  migration.destination import Loader
+from migration.source import Extractor
+
+
+class DatabaseMigration:
+    def __init__(self, extractor: Extractor, loader: Loader) -> None:
+        self.extractor = extractor
+        self.loader = loader
+        self.logger = logging.getLogger(__name__)
+
+    def run(self):
+        last_document = False
+        try:
+            while True:
+                    data, last_document = self.extractor.extract(last_document)
+                    if data:
+                        self.loader.load(data)
+                    else:
+                        break
+                    self.logger.info("Migration complete")
+        except Exception as e:
+                stack_trace = traceback.format_exc()
+                logging.error(stack_trace)
+                logging.error(f"Error when migrating data: {e}")