Update customer in Merit from Python with Merit API

merit-api-update-customer-email-django.py

import hmac
import datetime
import json
from base64 import b64encode
from urllib.parse import urlencode

import requests

from customers.models import Customer

API_KEY = b'...'
API_ID = '...'
API_BASE_URL = 'https://aktiva.merit.ee/api/v1/'
API_GET_CUSTOMERS_URL = API_BASE_URL + 'getcustomers'
API_UPDATE_CUSTOMER_URL = API_BASE_URL + 'updatecustomer'

def calculate_signature(timestamp):
    data_string = API_ID + timestamp
    hash_obj = hmac.new(key=API_KEY, msg=data_string.encode('ascii'), digestmod='sha256')
    return b64encode(hash_obj.digest())

def get_query_params():
    timestamp = datetime.datetime.now().strftime('%Y%m%d%H%M%S')
    signature = calculate_signature(timestamp)
    return {
        'ApiId': API_ID,
        'timestamp': timestamp,
        'signature': signature,
    }

# Alternatively could simply get all customers, the RegNo API filter seems to have no effect
def get_customer_from_merit_by_reg_no(reg_no):
    query_params = get_query_params()
    query_params['RegNo'] = reg_no
    customers = requests.get(API_GET_CUSTOMERS_URL + '?' + urlencode(query_params)).text
    customers = json.loads(json.loads(customers))
    customer = list(filter(lambda c: c['RegNo'] == reg_no, customers))
    return customer[0] if customer else None

def update_customer_email_in_merit(merit_customer, new_email):
    query_params = get_query_params()
    post_data = {
        'Id': merit_customer['CustomerId'],
        'Email': new_email,
    }
    result = requests.post(API_UPDATE_CUSTOMER_URL + '?' + urlencode(query_params), data=json.dumps(post_data)).text
    return result == '"Updated"'

def update_customer_emails():
    customers = Customer.objects.all()
    for django_customer in customers:
        merit_customer = get_customer_from_merit_by_reg_no(django_customer.registration_code)
        email = django_customer.emails.split(',')[0]
        if update_customer_email_in_merit(merit_customer, email):
            print('Successfully updated', django_customer, 'email to', email)
        else:
            print('Failed to update', django_customer)

if __name__ == '__main__':
    update_customer_emails()