feat(migrate): add migration by type (csv, json, excel)

hubmigrate/classes/migration.py

@@ -1,5 +1,7 @@
 import requests
 from .auth import Auth
+import pandas as pd
+import json
 
 
 class Migrate():
@@ -34,19 +36,37 @@ def get_data(self, params):
             print(f"Error retrieving data from {self.path} ❌")
         return data
 
-    
-    def post_data(self, data, path):
+
+    def post_data(self, data, path, data_type='json'):
         """ Post data to path 
         
         Arguments:
-            data {dict} -- Data to post to path
+            data {str or dict} -- Data to post to path (file path for Excel/CSV or dict for JSON)
+            path {str} -- HubSpot API path
+            data_type {str} -- Type of data ('json', 'excel', 'csv')
         """
-        url = f"{self.base_path}/{path}"
-        response = requests.post(url, json=data, headers=self.headers)
-        if response.status_code in self.status_codes:
-            print(f"Successfully posted data to {url} 🎉")
+        if data_type == 'json':
+            if isinstance(data, str):
+                data_list = json.loads(data)
+            else:
+                data_list = data
+        elif data_type == 'excel':
+            df = pd.read_excel(data)
+            data_list = df.to_dict(orient='records')
+        elif data_type == 'csv':
+            df = pd.read_csv(data)
+            data_list = df.to_dict(orient='records')
         else:
-            print(f"Error posting data to {url}: {response.status_code} - {response.text} ❌")  
+            raise ValueError("Unsupported data type. Use 'json', 'excel', or 'csv'.")
+
+        # Post each record to HubSpot
+        for record in data_list:
+            url = f"{self.base_path}/{path}"
+            response = requests.post(url, json=record, headers=self.headers)
+            if response.status_code in self.status_codes:
+                print(f"Successfully posted data to {url} 🎉")
+            else:
+                print(f"Error posting data to {url}: {response.status_code} - {response.text} ❌")
         return response
 
 

hubmigrate/client.py

@@ -35,13 +35,13 @@ def delete_contact(self, contact_id):
 
 
     # Migrate companies
-    def migrate_company(self, data):
+    def migrate_company(self, data, data_type):
         """ Migrate company to HubSpot 
         
         Arguments:
             data {dict} -- Data to migrate to HubSpot
         """
-        return self.migrate.post_data(data, 'companies')
+        return self.migrate.post_data(data, 'companies', data_type)
 
     def update_company(self, data, company_id):
         """ Update company in HubSpot 

hubmigrate/requirements.txt

@@ -1,4 +1,5 @@
 pytest
 coverage
 python-decouple
-requests
+requests
+pandas

hubmigrate/sample_company.xlsx

@@ -0,0 +1,2 @@
+name,country,city,address,zip,website,phone
+Politie Utrecht,Netherland,Utrecht,Marco Pololaan 215,3526GB,https://www.politie.nl/,0900-8844

hubmigrate/tests/migrate_company_excel.py

@@ -0,0 +1,33 @@
+import unittest
+from unittest.mock import Mock, patch
+from hubmigrate.client import MigrationClient
+from ..classes.auth import Auth
+import pandas as pd
+import os
+
+
+class TestMigrateCompanyExcel(unittest.TestCase):
+    @patch('hubmigrate.client.MigrationClient.migrate_company', return_value=Mock(status_code=200))
+    @patch('hubmigrate.classes.auth.Auth.get_token')
+    def test_migrate_company(self, mock_auth, mock_post):
+        # Create a test company
+        # Get the path to the current directory of your test script
+        current_dir = os.path.dirname(os.path.realpath(__file__))
+
+        # Define the relative path to your JSON file from the current directory
+        relative_path = '../sample_company.xlsx'
+
+        # Construct the full file path
+        excel_file_path = os.path.join(current_dir, relative_path)
+        path = excel_file_path
+        with open(path) as f:
+            test_company = pd.read_excel(f)
+
+        # Create an instance of the MigrationClient class
+        client = MigrationClient('config', 'hubspot')
+
+        # Call the migrate_company method
+        result = client.migrate_company({'properties': test_company}, 'excel')
+
+        # Assertions
+        self.assertEqual(result.status_code, 200)