Django implementation for communicating with Nmbrs' SOAP API

nmbrs.py

import datetime

from django.conf import settings

from tenant_schemas.utils import tenant_context
from zeep import Client

from apps.company import models
from apps.member.models import Tenant


class NmbrsBase:
    client = None
    header = None
    ns = None
    WSDL = None

    def _connect(self):
        self.client = Client(self.WSDL)

    def _create_header(self):
        el = self._get_element('AuthHeader')

        self.header = el(Username=settings.NMBRS_USERNAME, Token=settings.NMBRS_TOKEN)

    def _get_element(self, element):
        return self.client.get_element('{%s}%s' % (self.ns, element))


class NmbrsEmployee(NmbrsBase):
    company_code = None
    ns = None
    WSDL = None
    company_id = None
    test_mode = True

    def __init__(self, company_code, test_mode=True):
        self.WSDL = settings.NMBRS_EMPLOYEE_WSDL
        self.ns = settings.NMBRS_EMPLOYEE_NS

        self.company_code = company_code

        nmbrs_company = NmbrsCompany(test_mode=test_mode)
        nmbrs_company.set_company_id()

        self.company_id = nmbrs_company.company_id

    def get_users(self):
        self._connect()
        self._create_header()

        request_data = {
            'CompanyId': self.company_id,
            'active': 'all'
        }

        users = self.client.service.List_GetByCompany(
            **request_data, _soapheaders=[self.header])

        return users

    def get_employees(self):
        self._connect()
        self._create_header()

        request_data = {
            'CompanyID': self.company_id,
        }

        employees = self.client.service.Function_GetAll_AllEmployeesByCompany_V2(
            **request_data, _soapheaders=[self.header])

        return employees

    def create_user(self, user_pk):
        self._connect()
        self._create_header()

        tenant = Tenant.objects.filter(member__companycode=self.company_code).first()

        if not tenant:
            print('Help, no tenant found')
            return

        with tenant_context(tenant):
            student = models.StudentUser.objects.get(user__pk=user_pk)

            employee_id = self._do_create_user(student)
            self._do_create_address(employee_id, student)

    def _do_create_user(self, student):
        request_data = {
            'FirstName': student.user.first_name,
            'LastName': student.user.last_name,
            'CompanyId': self.company_id,
            'UnprotectedMode': False,
            'StartDate': datetime.datetime.now()
        }

        import ipdb
        ipdb.set_trace()
        result = self.client.service.Employee_Insert(
            **request_data, _soapheaders=[self.header])

        employee_id = result

        # save id
        student.external_id = employee_id
        student.save()

        return employee_id

    def _do_create_address(self, employee_id, student):
        request_data = {
            'EmployeeId': employee_id,
            'Address': {
                'Id': student.user.id,
                'Default': True,
                'Street': student.address,
                'HouseNumber': '',
                'HouseNumberAddition': False,
                'PostalCode': student.postal,
                'City': student.city,
                'StateProvince': student.state,
                'CountryISOCode': 'NL',
                'Type': 'HomeAddress',
            },
            'Period': 1,
            'Year': datetime.datetime.now().year,
            'UnprotectedMode': False,
        }

        self.client.service.Address_Insert(
            **request_data, _soapheaders=[self.header])

    def _do_update_personal_info(self, employee_id, student):
        request_data = {
            'EmployeeId': employee_id,
            'PersonalInfo': {
                'BSN': '',
                'Title': '',
                'FirstName': '',
                'Initials': '',
                'Prefix': '',
                'LastName': '',
                'Gender': '',
                'EmailPrivate': '',
            },
            'Period': 1,
            'Year': datetime.datetime.now().year,
        }

        self.client.service.PersonalInfo_Update(
            **request_data, _soapheaders=[self.header])


class NmbrsCompany(NmbrsBase):
    company_code = None
    companies = None
    ns = None
    WSDL = None
    company_id = None
    test_mode = True

    def __init__(self, test_mode=True):
        self.WSDL = settings.NMBRS_COMPANY_WSDL
        self.ns = settings.NMBRS_COMPANY_NS

    def get_companies(self):
        self._connect()
        self._create_header()

        self.companies = self.client.service.List_GetAll(_soapheaders=[self.header])

    def set_company_id(self):
        company_name = settings.NMBRS_TEST_COMPANY_NAME

        if not self.test_mode:
            company_name = settings.NMBRS_LIVE_COMPANY_NAME

        if not self.companies:
            self.get_companies()

        for company in self.companies:
            if company['Name'] == company_name:
                self.company_id = company['ID']