Base class for importing employees between fyle and bamboohr

admin_settings/tests/settings.py

@@ -50,7 +50,8 @@
     'apps.orgs',
     'apps.travelperk',
     'apps.gusto',
-    'apps.integrations'
+    'apps.integrations',
+    'apps.fyle_hrms_mappings',
 ]
 
 MIDDLEWARE = [

apps/fyle_hrms_mappings/migrations/0003_auto_20231221_1815.py

@@ -0,0 +1,30 @@
+# Generated by Django 3.1.14 on 2023-12-21 18:15
+
+from django.db import migrations
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('orgs', '0004_auto_20230627_1133'),
+        ('fyle_hrms_mappings', '0002_auto_20231221_1515'),
+    ]
+
+    operations = [
+        migrations.AlterUniqueTogether(
+            name='destinationattribute',
+            unique_together={('destination_id', 'attribute_type', 'org')},
+        ),
+        migrations.AlterUniqueTogether(
+            name='expenseattribute',
+            unique_together={('value', 'attribute_type', 'org')},
+        ),
+        migrations.AlterModelTable(
+            name='destinationattribute',
+            table='destination_attributes',
+        ),
+        migrations.AlterModelTable(
+            name='expenseattribute',
+            table='expense_attributes',
+        ),
+    ]

apps/fyle_hrms_mappings/models.py

@@ -22,6 +22,10 @@ class DestinationAttribute(models.Model):
     auto_created = models.BooleanField(default=False,
                                         help_text='Indicates whether the field is auto created by the integration')
 
+    class Meta:
+        db_table = 'destination_attributes'
+        unique_together = ('destination_id', 'attribute_type', 'org')
+
     @staticmethod
     def create_or_update_destination_attribute(attribute: Dict, org_id):
         """
@@ -33,7 +37,6 @@ def create_or_update_destination_attribute(attribute: Dict, org_id):
             org_id=org_id,
             defaults={
                 'active': attribute['active'] if 'active' in attribute else None,
-                'display_name': attribute['display_name'],
                 'value': attribute['value'],
                 'detail': attribute['detail'] if 'detail' in attribute else None
             }
@@ -42,14 +45,13 @@ 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, display_name: str = None):
+            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
         :param attribute_type: Attribute type
         :param attributes: attributes = [{
             'attribute_type': Type of attribute,
-            'display_name': Display_name of attribute_field,
             'value': Value of attribute,
             'destination_id': Destination Id of the attribute,
             'detail': Extra Details of the attribute
@@ -64,11 +66,9 @@ def bulk_create_or_update_destination_attributes(
             'attribute_type': attribute_type,
             'org_id': org_id
         }
-        if display_name:
-            filters['display_name'] = display_name
 
         existing_attributes = DestinationAttribute.objects.filter(**filters)\
-            .values('id', 'destination_id', 'detail', 'active')
+            .values('id', 'destination_id', 'value', 'detail', 'active')
 
         existing_attribute_destination_ids = []
 
@@ -94,7 +94,6 @@ def bulk_create_or_update_destination_attributes(
                 attributes_to_be_created.append(
                     DestinationAttribute(
                         attribute_type=attribute_type,
-                        display_name=attribute['display_name'],
                         value=attribute['value'],
                         destination_id=attribute['destination_id'],
                         detail=attribute['detail'] if 'detail' in attribute else None,
@@ -142,6 +141,9 @@ class ExpenseAttribute(models.Model):
     auto_created = models.BooleanField(default=False,
                                     help_text='Indicates whether the field is auto created by the integration')
 
+    class Meta:
+        db_table = 'expense_attributes'
+        unique_together = ('value', 'attribute_type', 'org')
 
     @staticmethod
     def create_or_update_expense_attribute(attribute: Dict, org_id):
@@ -155,7 +157,6 @@ def create_or_update_expense_attribute(attribute: Dict, org_id):
             defaults={
                 'active': attribute['active'] if 'active' in attribute else None,
                 'source_id': attribute['source_id'],
-                'display_name': attribute['display_name'],
                 'detail': attribute['detail'] if 'detail' in attribute else None
             }
         )
@@ -170,7 +171,6 @@ def bulk_create_or_update_expense_attributes(
         :param attribute_type: Attribute type
         :param attributes: attributes = [{
             'attribute_type': Type of attribute,
-            'display_name': Display_name of attribute_field,
             'value': Value of attribute,
             'source_id': Fyle Id of the attribute,
             'detail': Extra Details of the attribute
@@ -182,7 +182,7 @@ def bulk_create_or_update_expense_attributes(
 
         existing_attributes = ExpenseAttribute.objects.filter(
             value__in=attribute_value_list, attribute_type=attribute_type,
-            org_id=org_id).values('id', 'detail', 'active')
+            org_id=org_id).values('id', 'detail', 'value', 'active')
 
         existing_attribute_values = []
 
@@ -206,7 +206,6 @@ def bulk_create_or_update_expense_attributes(
                 attributes_to_be_created.append(
                     ExpenseAttribute(
                         attribute_type=attribute_type,
-                        display_name=attribute['display_name'],
                         value=attribute['value'],
                         source_id=attribute['source_id'],
                         detail=attribute['detail'] if 'detail' in attribute else None,

apps/users/helpers.py

@@ -1,8 +1,9 @@
 from django.conf import settings
 import json
-from typing import Dict
+from typing import Dict, List
 import requests
 from fyle.platform import Platform
+from apps.fyle_hrms_mappings.models import ExpenseAttribute
 
 
 class PlatformConnector:
@@ -20,8 +21,54 @@ def __init__(self, refresh_token: str, cluster_domain: str):
             client_secret=settings.FYLE_CLIENT_SECRET,
             refresh_token=refresh_token
         )
+
+    def bulk_post_employees(self, employees_payload):
+        self.connection.v1beta.admin.employees.invite_bulk({'data': employees_payload})
 
+    def get_department_generator(self, query_params):
+        departments = self.connection.v1beta.admin.departments.list_all(query_params={
+            'order': 'id.desc'
+        })
+        return departments
 
+    def post_department(self, department):
+        self.connection.v1beta.admin.departments.post({"data": department})
+
+    def bulk_create_or_update_expense_attributes(self, attributes: List[dict], attribute_type, org_id, update_existing: bool = False) -> None:
+        """
+        Bulk creates or updates expense attributes.
+        :param attributes: List of expense attributes.
+        :param update_existing: If True, updates/creates the existing expense attributes.
+        """
+        ExpenseAttribute.bulk_create_or_update_expense_attributes(
+            attributes, attribute_type, org_id, update_existing
+        )
+
+    def sync_employees(self, org_id):
+        query_params = {'is_enabled': 'eq.true','order': 'updated_at.desc'}
+        attribute_type = 'EMPLOYEE'
+        generator = self.connection.v1beta.admin.employees.list_all(query_params)
+        for items in generator:
+            employee_attributes = []
+            for employee in items['data']:
+                employee_attributes.append({
+                        'attribute_type': attribute_type,
+                        'display_name': attribute_type.replace('_', ' ').title(),
+                        'value': employee['user']['email'],
+                        'source_id': employee['id'],
+                        'active': True,
+                        'detail': {
+                            'user_id': employee['user_id'],
+                            'employee_code': employee['code'],
+                            'full_name': employee['user']['full_name'],
+                            'location': employee['location'],
+                            'department': employee['department']['name'] if employee['department'] else None,
+                            'department_id': employee['department_id'],
+                            'department_code': employee['department']['code'] if employee['department'] else None
+                        }
+                    })
+
+        self.bulk_create_or_update_expense_attributes(employee_attributes, attribute_type, org_id, True)
 
 def post_request(url: str, body: Dict, api_headers: Dict) -> Dict:
     """

fyle_employee_imports/bamboo_hr.py

@@ -0,0 +1,43 @@
+from typing import Dict
+from apps.users.models import User
+from apps.fyle_hrms_mappings.models import DestinationAttribute
+from .base import FyleEmployeeImport
+from bamboosdk.bamboohrsdk import BambooHrSDK
+from apps.bamboohr.models import BambooHr
+
+
+class BambooHrEmployeeImport(FyleEmployeeImport):
+
+    def __init__(self, org_id: int, user: User):
+        super().__init__(org_id, user)
+        bamboo_hr = BambooHr.objects.get(org_id__in=org_id)
+        self.bamboohr_sdk = BambooHrSDK(api_token=bamboo_hr.api_token, sub_domain=bamboo_hr.sub_domain)
+
+    def sync_hrms_employees(self):
+        employees = self.bamboohr_sdk.employees.get_all()
+        self.upsert_employees(employees)
+
+    def upsert_employees(self, employees: Dict):
+        attributes = []
+        for employee in employees['employees']:
+            supervisor = [employee['supervisorEmail']]
+            active_status = True if employee['status'] == 'Active' else False
+            detail = {
+                'email': employee['workEmail'] if employee['workEmail'] else None,
+                'department_name': employee['department'] if employee['department'] else None,
+                'full_name': employee['displayName'] if employee['displayName'] else None,
+                'approver_emails': supervisor,
+            }
+
+            attributes.append({
+                'attribute_type': 'EMPLOYEE',
+                'value': employee['displayName'],
+                'destination_id': employee['id'],
+                'detail': detail,
+                'active': active_status
+                })
+
+        DestinationAttribute.bulk_create_or_update_destination_attributes(
+            attributes=attributes, attribute_type='EMPLOYEE', org_id=self.org_id, update=True)
+
+        return []

fyle_employee_imports/base.py

@@ -0,0 +1,136 @@
+from typing import Dict, List
+from apps.fyle_hrms_mappings.models import DestinationAttribute, ExpenseAttribute
+from apps.orgs.models import Org
+from apps.users.helpers import PlatformConnector
+from fyle_rest_auth.models import AuthToken
+
+class FyleEmployeeImport():
+
+    def __init__(self, org_id: int, user):
+        self.org_id = org_id
+        self.user = user
+        refresh_token = AuthToken.objects.get(user__user_id=self.user).refresh_token
+        cluster_domain = Org.objects.get(user__user_id=self.user).cluster_domain
+        self.platform_connection = PlatformConnector(refresh_token, cluster_domain)
+
+    def sync_fyle_employees(self):
+        self.platform_connection.sync_employees(org_id=self.org_id)
+
+    def get_existing_departments_from_fyle(self):
+        existing_departments: Dict = {}
+        query_params={
+            'order': 'id.desc'
+        }
+        departments_generator = self.platform_connection.get_department_generator(query_params=query_params)
+        for response in departments_generator:
+            for department in response['data']:
+                existing_departments[department['display_name']] = {
+                    'id': department['id'],
+                    'is_enabled': department['is_enabled']
+                }
+        return existing_departments
+
+    def create_fyle_department_payload(self, existing_departments, new_departments):
+        departments_payload = []
+
+        for department in new_departments:
+            if department in existing_departments.keys():
+                if not existing_departments[department]['is_enabled']:
+                    departments_payload.append({
+                    'name': department,
+                    'id': existing_departments[department]['id'],
+                    'is_enabled': True,
+                    'display_name': department
+                })
+            else:
+                departments_payload.append({
+                'name': department,
+                'display_name': department
+            })
+
+        return departments_payload
+
+    def departments_to_be_imported(self, hrms_employees):
+        new_departments = []
+
+        for employee in hrms_employees:
+            if employee.detail['department_name']:
+                new_departments.append(employee.detail['department_name'])
+
+        return list(set(new_departments))
+
+    def post_department(self, departments_payload):
+        for department in departments_payload:
+            self.platform_connection.post_department(department)
+
+    def import_departments(self, hrms_employees):
+        existing_departments = self.get_existing_departments_from_fyle()
+        new_departments = self.departments_to_be_imported(hrms_employees)
+        departments_payload = self.create_fyle_department_payload(existing_departments, new_departments)
+        self.post_department(departments_payload)
+
+    def get_employee_and_approver_payload(self, hrms_employees):
+        employee_payload: List[Dict] = []
+        employee_emails: List[str] = []
+        approver_emails: List[str] = []
+        employee_approver_payload: List[Dict] = []
+
+        for employee in hrms_employees:
+            if employee.detail['email']:
+                update_create_employee_payload = {
+                    'user_email': employee.detail['email'],
+                    'user_full_name': employee.detail['full_name'],
+                    'code': employee.destination_id,
+                    'department_name': employee.detail['department_name'] if employee.detail['department_name'] else '',
+                    'is_enabled': employee.active
+                }
+                employee_payload.append(update_create_employee_payload)
+                employee_emails.append(employee.detail['email'])
+
+                if employee.detail['approver_emails']:
+                    employee_approver_payload.append({
+                        'user_email': employee.detail['email'],
+                        'approver_emails': employee.detail['approver_emails']
+                    })
+                    approver_emails.extend(employee.detail['approver_emails'])
+
+        existing_approver_emails = ExpenseAttribute.objects.filter(
+            org_id=self.org_id, attribute_type='EMPLOYEE', value__in=approver_emails
+        ).values_list('value', flat=True)
+
+        employee_approver_payload = list(filter(
+            lambda employee_approver: set(
+                employee_approver['approver_emails']
+            ).issubset(employee_emails) or set(
+                employee_approver['approver_emails']
+            ).issubset(existing_approver_emails),
+            employee_approver_payload
+        ))
+
+        return employee_payload, employee_approver_payload
+
+    def fyle_employee_import(self, hrms_employees):
+        fyle_employee_payload, employee_approver_payload = self.get_employee_and_approver_payload(hrms_employees)
+
+        if fyle_employee_payload:
+            self.platform_connection.bulk_post_employees(employees_payload=fyle_employee_payload)
+
+        if employee_approver_payload:
+            self.platform_connection.bulk_post_employees(employees_payload=employee_approver_payload)
+
+        self.platform_connection.sync_employees(org_id=self.org_id)
+
+    def sync_hrms_employees(self):
+        raise NotImplementedError('Implement sync_hrms_employees() in the child class')
+
+    def sync_employees(self):
+        self.sync_fyle_employees()
+        self.sync_hrms_employees()
+
+        hrms_employees = DestinationAttribute.objects.filter(
+            attribute_type='EMPLOYEE',
+            org_id=self.org_id
+        ).order_by('value', 'id')
+
+        self.import_departments(hrms_employees)
+        self.fyle_employee_import(hrms_employees)

requirements.txt

@@ -12,7 +12,7 @@ django-filter==21.1
 dj-database-url==0.5.0
 
 # Platform SDK
-fyle==v0.29.0
+fyle==v0.35.0
 
 # DjangoQ for running async tasks
 django-q==1.3.4