Automating Invoice Updates in Zuora with Python

Effective subscription management depends on synchronizing your invoicing system with your accounting tools. Sometimes, invoices can get stuck or produce errors during the transfer to your accounting system, causing discrepancies and manual workload. In this article, we'll explore how to use a Python script to automate the process of identifying specific invoices in Zuora and updating their "Transferred to Accounting" field to "Ignore". This ensures they won't be picked up by ERP integrations that rely on this field as a filter.

Note: This script is designed to handle a large volume of invoices efficiently without overloading your server or crashing your computer.

1. Replace Credentials:

   • Replace 'your_client_id' with your actual Zuora API CLIENT_ID.

   • Replace 'your_client_secret' with your actual Zuora API CLIENT_SECRET.

   • Replace '2c92c0f86d4df9d3016d5556e8a34567' with your actual RATE_PLAN_ID.

2. Configure Logging (Optional):

   The script uses Python's logging module to write logs to a file named invoice_update.log. You can change the filename or logging level as needed.

3. Run the Script:

   Save the script to a file, e.g., update_invoices.py, and run it from your terminal:

   python update_invoices.py

• Imports and Logging Configuration:

  import requests
  import time
  import logging
  
  logging.basicConfig(level=logging.INFO, filename='invoice_update.log', filemode='a',
                      format='%(asctime)s - %(levelname)s - %(message)s')

  • requests: For making HTTP requests to Zuora's API.

  • time: For adding delays to prevent hitting rate limits.

  • logging: For logging information and errors.

• Global Variables:

  CLIENT_ID = 'your_client_id'
  CLIENT_SECRET = 'your_client_secret'
  RATE_PLAN_ID = '2c92c0f86d4df9d3016d5556e8a34567'

• Authentication Function:

  def get_access_token():
      # Function code...

  • Authenticates with Zuora's API to obtain an access token.

  • Handles exceptions and logs errors.

• Query Invoices Function:

  def query_invoices(rate_plan_id, page_size=200):
      # Function code...

  • Queries invoices matching the specified criteria.

  • Uses pagination to handle large volumes efficiently.

  • Handles exceptions and logs errors.

• Update Invoices Function:

  def update_invoices_in_batches(invoices, batch_size=100):
      # Function code...

  • Updates invoices in batches to prevent overloading the system.

  • Implements retry logic and rate limiting.

  • Logs successes and errors.

• Main Function:

  def main(): 
     # Function code...

  • Orchestrates the authentication, querying, and updating processes.

  • Handles any exceptions that may cause the script to terminate.

• Execution Entry Point:

  if __name__ == '__main__': 
    main()

  • Ensures that the main() function is called when the script is executed.

• Error Handling:

  The script includes try-except blocks to catch and log exceptions without crashing.

• Rate Limiting:

  • The script sleeps for a short duration between batches to respect Zuora's API rate limits.

  • If a rate limit error (HTTP 429) is encountered, the script waits longer before retrying.

• Logging:

  • Successes and errors are logged with timestamps.

  • The log file invoice_update.log will be created in the same directory as the script.

• Testing:

  • Always test the script in a sandbox environment before running it in production.

  • Verify that the credentials and Rate Plan ID are correct.

• Data Safety:

  Be cautious when running scripts that modify data. Ensure you have backups or can revert changes if necessary.

• Permissions:

  Make sure your Zuora API credentials have the necessary permissions to perform the queries and updates.

import requests
import time
import logging

# Configure logging
logging.basicConfig(level=logging.INFO, filename='invoice_update.log', filemode='a',
                    format='%(asctime)s - %(levelname)s - %(message)s')

# Replace with your actual credentials
CLIENT_ID = 'your_client_id'          # Replace with your Zuora API client ID
CLIENT_SECRET = 'your_client_secret'  # Replace with your Zuora API client secret
RATE_PLAN_ID = '2c92c0f86d4df9d3016d5556e8a34567'  # Replace with your Rate Plan ID

def get_access_token():
    """
    Authenticate with Zuora's API and obtain an access token.
    """
    url = 'https://rest.zuora.com/oauth/token'  # Use 'https://rest.zuora.com' for Production
    payload = {
        'client_id': CLIENT_ID,
        'client_secret': CLIENT_SECRET,
        'grant_type': 'client_credentials'
    }
    try:
        response = requests.post(url, data=payload)
        response.raise_for_status()
        access_token = response.json()['access_token']
        logging.info("Successfully obtained access token.")
        return access_token
    except requests.exceptions.HTTPError as e:
        logging.error(f"HTTPError during authentication: {e}")
        raise
    except Exception as e:
        logging.error(f"Error during authentication: {e}")
        raise

def query_invoices(rate_plan_id, page_size=200):
    """
    Query invoices that meet the specified criteria using pagination.

    :param rate_plan_id: The ID of the rate plan to filter invoices.
    :param page_size: Number of records per page.
    :return: A list of invoice records.
    """
    invoices = []
    query = f"""
    SELECT Id
    FROM Invoice
    WHERE TransferredToAccounting != 'Ignore'
    AND TransferredToAccounting != 'Yes'
    AND Id IN (
        SELECT InvoiceId
        FROM InvoiceItem
        WHERE RatePlanId = '{rate_plan_id}'
    )
    """
    url = 'https://rest.apisandbox.zuora.com/v1/action/query'
    payload = {
        'queryString': query,
        'pageSize': page_size
    }

    try:
        response = requests.post(url, headers=headers, json=payload)
        response.raise_for_status()
        result = response.json()
        invoices.extend(result.get('records', []))
        query_locator = result.get('queryLocator')

        while query_locator:
            payload = {'queryLocator': query_locator}
            response = requests.post(f'{url}More', headers=headers, json=payload)
            response.raise_for_status()
            result = response.json()
            invoices.extend(result.get('records', []))
            query_locator = result.get('queryLocator')

        logging.info(f"Retrieved {len(invoices)} invoices to update.")
        return invoices
    except requests.exceptions.HTTPError as e:
        logging.error(f"HTTPError during invoice query: {e}")
        raise
    except Exception as e:
        logging.error(f"Error during invoice query: {e}")
        raise

def update_invoices_in_batches(invoices, batch_size=100):
    """
    Update the 'TransferredToAccounting' field of invoices in batches.

    :param invoices: List of invoice records to update.
    :param batch_size: Number of invoices to update per batch.
    """
    url = 'https://rest.apisandbox.zuora.com/v1/action/update'
    total_invoices = len(invoices)

    for i in range(0, total_invoices, batch_size):
        batch = invoices[i:i+batch_size]
        objects = [{
            'Type': 'Invoice',
            'Id': invoice['Id'],
            'TransferredToAccounting': 'Ignore'
        } for invoice in batch]

        payload = {'objects': objects}

        for attempt in range(3):  # Retry up to 3 times
            try:
                response = requests.post(url, headers=headers, json=payload)
                response.raise_for_status()
                break  # Exit retry loop on success
            except requests.exceptions.HTTPError as e:
                if response.status_code == 429:
                    logging.warning("Rate limit exceeded. Sleeping for 5 seconds.")
                    time.sleep(5)
                else:
                    logging.error(f"HTTP Error: {e}")
                    time.sleep(1)
            except Exception as e:
                logging.error(f"Unexpected error: {e}")
                time.sleep(1)
        else:
            logging.error("Failed to update batch after 3 attempts.")
            continue  # Proceed to next batch

        update_results = response.json().get('results', [])
        for result in update_results:
            if result.get('Success'):
                logging.info(f"Invoice {result['Id']} updated successfully.")
            else:
                logging.error(f"Failed to update Invoice {result['Id']}: {result.get('Errors')}")

        # Sleep to prevent hitting rate limits
        time.sleep(1)

def main():
    """
    Main function to execute the script.
    """
    try:
        access_token = get_access_token()
        global headers
        headers = {
            'Authorization': f'Bearer {access_token}',
            'Content-Type': 'application/json'
        }
        invoices = query_invoices(RATE_PLAN_ID)
        if invoices:
            update_invoices_in_batches(invoices)
        else:
            logging.info("No invoices found to update.")
    except Exception as e:
        logging.error(f"Script terminated due to an error: {e}")

if __name__ == '__main__':
    main()