Webhook Workato like setup (#143)

* Webhook workato like setup

* comment resolved

* added private key for future security

apps/bamboohr/migrations/0008_bamboohr_private_key.py

@@ -0,0 +1,18 @@
+# Generated by Django 3.1.14 on 2024-01-18 09:29
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('bamboohr', '0007_bamboohr_webhook_id'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='bamboohr',
+            name='private_key',
+            field=models.CharField(help_text='Private key to verify webhook calls', max_length=255, null=True),
+        ),
+    ]

apps/bamboohr/models.py

@@ -20,7 +20,8 @@ class BambooHr(models.Model):
     employee_exported_at = models.DateTimeField(auto_now_add=True, help_text='Employee exported to Fyle at datetime')
     is_credentials_expired = models.BooleanField(default=False, help_text='BambooHr Credential Status')
     webhook_id = models.IntegerField(null=True, help_text='ID of the webhook created by BambooHr')
-
+    private_key = models.CharField(max_length=255, null=True, help_text='Private key to verify webhook calls')
+
     class Meta:
         db_table = 'bamboohr'
 

apps/bamboohr/signals.py

@@ -18,7 +18,7 @@ def run_post_save_bamboohr_triggers(sender, instance: BambooHr, **kwargs):
             'department': 'department',
             'workEmail': 'workEmail',
             'status': 'status',
-            'reportingTo': 'reportingTo'
+            'supervisorEId': 'supervisorEId'
         },
         'name': instance.org.name,
         'monitorFields': ['firstName', 'lastName', 'department', 'workEmail', 'status', 'reportingTo'],
@@ -27,4 +27,5 @@ def run_post_save_bamboohr_triggers(sender, instance: BambooHr, **kwargs):
     }
 
     response = bamboohrsdk.webhook.post(payload=webhook_payload)
-    BambooHr.objects.filter(id=instance.id).update(webhook_id=int(response['id']))
+    private_key = response['privateKey']
+    BambooHr.objects.filter(id=instance.id).update(webhook_id=int(response['id']), private_key=private_key)

apps/bamboohr/tasks.py

@@ -3,6 +3,7 @@
 from apps.orgs.models import FyleCredential, Org
 from apps.users.helpers import PlatformConnector
 from fyle_employee_imports.bamboo_hr import BambooHrEmployeeImport
+from bamboosdk.bamboohrsdk import BambooHrSDK
 
 def refresh_employees(org_id: int):
     """
@@ -16,28 +17,14 @@ def update_employee(org_id: int, payload: dict):
     """
         Update employee in fyle when employee in Bamboohr is added or updated
     """
-    bamboohr = BambooHr.objects.get(org_id__in=[org_id], is_credentials_expired=False)
-    bamboohr_importer = BambooHrEmployeeImport(org_id=org_id)
-    employee_payload = {'employees': []}
-    payload = payload['employees'][0]
     employee = {}
+    payload = payload['employees'][0]
     employee['id'] = payload['id']
-    employee['firstName'] = payload['fields']['firstName']['value']
-    employee['lastName'] = payload['fields']['lastName']['value']
-    for field in payload['changedFields']:  
-        employee[field] = payload['fields'][field]['value']
-
-    if not employee.get('status', None):
-        employee['status'] = 'Active'
-
-    employee_payload['employees'].append(employee)
-
-    bamboohr_importer.upsert_employees(employees=employee_payload, webhook_update=True)
-
-    hrms_employees = DestinationAttribute.objects.filter(
-            attribute_type='EMPLOYEE',
-            org_id=org_id,
-            updated_at__gte=bamboohr.employee_exported_at,
-        ).order_by('value', 'id')
-    bamboohr_importer.import_departments(hrms_employees=hrms_employees)
-    bamboohr_importer.fyle_employee_import(hrms_employees=hrms_employees)
+    payload = payload['fields']
+    for field in payload.keys():
+        employee[field] = payload[field]['value']
+
+    employee['status'] = True if employee.get('status', None) == 'Active' else False
+
+    bamboohr_importer = BambooHrEmployeeImport(org_id=org_id)
+    bamboohr_importer.sync_with_webhook(employee=employee)

apps/fyle_hrms_mappings/models.py

@@ -45,7 +45,7 @@ def create_or_update_destination_attribute(attribute: Dict, org_id):
 
     @staticmethod
     def bulk_create_or_update_destination_attributes(
-            attributes: List[Dict], attribute_type: str, org_id: int, update: bool = False, webhook_update: bool = False):
+            attributes: List[Dict], attribute_type: str, org_id: int, update: bool = False):
         """
         Create Destination Attributes in bulk
         :param update: Update Pre-existing records or not
@@ -109,26 +109,6 @@ def bulk_create_or_update_destination_attributes(
                         or
                         ('active' in attribute and attribute['active'] != primary_key_map[attribute['destination_id']]['active'])
                     ):
-                    if webhook_update:
-                        # update the fields whose keys are present in the payload, if they are none in the payload
-                        # update with the old values
-                        existing_employee_attribute = primary_key_map[attribute['destination_id']]['detail']
-                        new_employee_attributes = attribute['detail']
-                        updated_employee_attributes_detail = {key: new_value if new_value is not None else existing_employee_attribute[key] \
-                                                            for key, new_value in new_employee_attributes.items()}
-
-                        # for approver email we keep the existing value as we don't get this in the webhook call
-                        updated_employee_attributes_detail['approver_emails'] = existing_employee_attribute['approver_emails']
-                        attributes_to_be_updated.append(
-                            DestinationAttribute(
-                                id=primary_key_map[attribute['destination_id']]['id'],
-                                value=attribute['value'],
-                                detail=updated_employee_attributes_detail,
-                                active=attribute['active'] if 'active' in attribute else None,
-                                updated_at=datetime.now()
-                            )
-                        )
-                    else:
                         attributes_to_be_updated.append(
                             DestinationAttribute(
                                 id=primary_key_map[attribute['destination_id']]['id'],

apps/users/helpers.py

@@ -22,6 +22,17 @@ def __init__(self, refresh_token: str, cluster_domain: str):
             refresh_token=refresh_token
         )
 
+    def get_employee_by_email(self, email: str):
+        """
+        Get employee by email
+        """
+        return self.connection.v1beta.admin.employees.list({
+            'user->email': 'eq.{}'.format(email),
+            'offset': 0,
+            'limit': 1,
+            'order': 'updated_at.desc'
+        })['data']
+
     def bulk_post_employees(self, employees_payload):
         self.connection.v1beta.admin.employees.invite_bulk({'data': employees_payload})
 

bamboosdk/api/employee.py

@@ -4,6 +4,7 @@
 class Employee(ApiBase):
 
     GET_EMPLOYEE_REPORT = '/v1/reports/custom?format=JSON&onlyCurrent=false'
+    GET_EMPLOYEE = '/v1/employees/{}/?fields=workEmail&onlyCurrent=false'
     payload = { "fields": ["displayName", "firstName", "lastName", "department", "workEmail", "supervisorEmail", "status"] }
 
     def get_all(self):
@@ -12,3 +13,6 @@ def get_all(self):
             List with dicts in Employee schema.
         """
         return self._post_request(self.GET_EMPLOYEE_REPORT, payload=self.payload)
+
+    def get(self, id):
+        return self._get_request(self.GET_EMPLOYEE.format(id))

fyle_employee_imports/bamboo_hr.py

@@ -1,4 +1,5 @@
 from typing import Dict
+from apps.bamboohr.email import send_failure_notification_email
 from apps.users.models import User
 from apps.fyle_hrms_mappings.models import DestinationAttribute
 from .base import FyleEmployeeImport
@@ -31,8 +32,48 @@ def get_employee_exported_at(self):
     def sync_hrms_employees(self):
         employees = self.bamboohr_sdk.employees.get_all()
         self.upsert_employees(employees)
+
+    def sync_with_webhook(self, employee):
+        employee_payload = []
+        employee_approver_payload = []
+
+        full_name = '{} {}'.format(employee['firstName'], employee['lastName'])
+
+        if not employee.get('workEmail'):
+            admin_email = self.get_admin_email()
+            incomplete_employee = {'name': full_name, 'id': employee['id']}
+            send_failure_notification_email(employees=[incomplete_employee], number_of_employees=1, admin_email=admin_email)
+            return
+        if employee.get('supervisorEId'):
+            response = self.bamboohr_sdk.employees.get(employee['supervisorEId'])
+            email = response['workEmail']
+            fyle_employee = self.platform_connection.get_employee_by_email(email=email)
+            if len(fyle_employee):
+                employee_approver_payload.append({
+                    'user_email': employee['workEmail'],
+                    'approver_emails': [email]
+                    }
+                )
+        update_create_employee_payload = {
+            'user_email': employee['workEmail'],
+            'user_full_name': full_name,
+            'code': employee['id'],
+            'department_name': employee['department'] if employee['department'] else '',
+            'is_enabled': employee['status']
+        }
+        employee_payload.append(update_create_employee_payload)
+        if employee['department']:
+            existing_departments = self.get_existing_departments_from_fyle()
+            department_payload = self.create_fyle_department_payload(existing_departments, [employee['department']])
+            if department_payload:
+                self.platform_connection.post_department(department_payload[0])
+
+        self.platform_connection.bulk_post_employees(employees_payload=employee_payload)
+        if employee_approver_payload:
+            self.platform_connection.bulk_post_employees(employees_payload=employee_approver_payload)
+
 
-    def upsert_employees(self, employees: Dict, webhook_update: bool = False):
+    def upsert_employees(self, employees: Dict):
         attributes = []
         for employee in employees['employees']:
 
@@ -59,6 +100,6 @@ def upsert_employees(self, employees: Dict, webhook_update: bool = False):
             })
 
         DestinationAttribute.bulk_create_or_update_destination_attributes(
-            attributes=attributes, attribute_type='EMPLOYEE', org_id=self.org_id, update=True, webhook_update=webhook_update)
+            attributes=attributes, attribute_type='EMPLOYEE', org_id=self.org_id, update=True)
 
         return []