Scribd
Upload
4 views6 pages

Econet API Integration with Flask

This document outlines a Flask application that interfaces with the Econet XML-RPC API to perform various operations such as loading airtime, bundles, deducting value, checking account balances, validating MSISDNs, and transferring accounts. It includes error handling for missing parameters and unexpected API responses, and utilizes environment variables for sensitive information like API credentials. The application defines multiple API endpoints for different functionalities, ensuring structured JSON responses for success and error cases.

Uploaded by

emeliamakore
Copyright
© All Rights Reserved
We take content rights seriously. If you suspect this is your content, claim it here.
Available Formats
Download as TXT, PDF, TXT or read online on Scribd
4 views6 pages

Econet API Integration with Flask

This document outlines a Flask application that interfaces with the Econet XML-RPC API to perform various operations such as loading airtime, bundles, deducting value, checking account balances, validating MSISDNs, and transferring accounts. It includes error handling for missing parameters and unexpected API responses, and utilizes environment variables for sensitive information like API credentials. The application defines multiple API endpoints for different functionalities, ensuring structured JSON responses for success and error cases.

Uploaded by

emeliamakore
Copyright
© All Rights Reserved
We take content rights seriously. If you suspect this is your content, claim it here.
Available Formats
Download as TXT, PDF, TXT or read online on Scribd

from flask import Flask, request, jsonify

import [Link]
import [Link] as ET
import os

app = Flask(__name__)

ECONET_USERNAME = [Link]("ECONET_USERNAME")
ECONET_PASSWORD = [Link]("ECONET_PASSWORD")
ECONET_API_URL = [Link]("ECONET_API_URL",
"[Link]

if not ECONET_USERNAME or not ECONET_PASSWORD:


raise ValueError("Econet username and password must be set as environment
variables.")

def call_econet_api(method_name, params):


"""Calls the Econet XML-RPC API."""
try:
server = [Link](ECONET_API_URL)
api_params = [ECONET_USERNAME, ECONET_PASSWORD]
api_params.append(params)
result = getattr(server, method_name)(*api_params)

if isinstance(result, str):
root = [Link](result)
response_data = xml_to_dict(root)
return response_data
else:
print(f"Unexpected response from Econet API: {result}")
return None

except [Link] as e:
print(f"XML-RPC Fault: {e}")
return None
except Exception as e:
print(f"An error occurred: {e}")
return None

def xml_to_dict(element):
"""Recursively converts XML to a Python dictionary."""
data = {}
for child in element:
if [Link] == 'params':
for param in child:
for value in param:
if [Link] == 'value': # Handle direct values
for child2 in value:
if [Link] == 'struct':
for member in child2:
name = None
value_element = None
for item in member:
if [Link] == 'name':
name = [Link]
elif [Link] == 'value':
value_element = item
if name and value_element:
value = None
for v in value_element:
if [Link] is not None:
value = [Link]
try:
value = int(value)
except ValueError:
try:
value = float(value)
except ValueError:
pass
break # take first non-empty value
data[name] = value
elif [Link] is not None: # Handling direct values
in response
data = [Link]
try:
data = int(data)
except ValueError:
try:
data = float(data)
except ValueError:
pass
break
return data

@[Link]("/api/airtime/load", methods=["POST"])
def load_airtime():
data = request.get_json()
msisdn = [Link]("msisdn")
amount = [Link]("amount")
reference = [Link]("reference")
currency = [Link]("currency", 840) # Default to Zimbabwe Dollar if not
provided

if not msisdn or not amount or not reference:


return jsonify({"error": "Missing msisdn, amount, or reference"}), 400

api_params = {
"MSISDN": str(msisdn),
"Amount": int(amount),
"Reference": str(reference),
"Currency": int(currency) # Ensure Currency is an integer
}

response = call_econet_api("load_value", api_params)

if response:
status = [Link]("Status")
status_code = [Link]("StatusCode")
description = [Link]("Description")

if status == 1:
return jsonify({
"status": "success",
"statusCode": status_code,
"description": description,
"serial": [Link]("Serial"),
"providerReference": [Link]("ProviderReference"),
"rate": [Link]("Rate")
}), 200
else:
return jsonify({
"status": "error",
"statusCode": status_code,
"description": description
}), 400
else:
return jsonify({"error": "Failed to call Econet API"}), 500

@[Link]("/api/bundle/load", methods=["POST"])
def load_bundle():
data = request.get_json()
msisdn = [Link]("msisdn")
provider_code = [Link]("provider_code")
amount = [Link]("amount")
currency = [Link]("currency")
account_type = [Link]("account_type")
quantity = [Link]("quantity")
reference = [Link]("reference")
product_code = [Link]("product_code") # Ensure Product code is available

if not msisdn or not provider_code or not amount or not currency or not


account_type or not quantity or not reference or not product_code:
return jsonify({"error": "Missing required parameters for load_bundle"}),
400

api_params = {
"MSISDN": str(msisdn),
"ProviderCode": int(provider_code),
"Amount": int(amount),
"Currency": int(currency),
"AccountType": int(account_type),
"Quantity": int(quantity),
"Reference": str(reference),
"ProductCode": str(product_code)
}

response = call_econet_api("load_bundle", api_params)

if response:
status = [Link]("Status")
status_code = [Link]("StatusCode")
description = [Link]("Description")

if status == 1:
return jsonify({
"status": "success",
"statusCode": status_code,
"description": description,
"serial": [Link]("Serial"),
"providerReference": [Link]("ProviderReference")
}), 200
else:
return jsonify({
"status": "error",
"statusCode": status_code,
"description": description
}), 400
else:
return jsonify({"error": "Failed to call Econet API"}), 500

@[Link]("/api/value/deduct", methods=["POST"])
def deduct_value():
data = request.get_json()
msisdn = [Link]("msisdn")
provider_code = [Link]("provider_code")
amount = [Link]("amount")
reference = [Link]("reference")

if not msisdn or not provider_code or not amount or not reference:


return jsonify({"error": "Missing required parameters for deduct_value"}),
400

api_params = {
"MSISDN": str(msisdn),
"ProviderCode": int(provider_code),
"Amount": int(amount),
"Reference": str(reference)
}

response = call_econet_api("deduct_value", api_params)

if response:
status = [Link]("Status")
status_code = [Link]("StatusCode")
description = [Link]("Description")

if status == 1:
return jsonify({
"status": "success",
"statusCode": status_code,
"description": description,
"serial": [Link]("Serial"),
"providerReference": [Link]("ProviderReference")
}), 200
else:
return jsonify({
"status": "error",
"statusCode": status_code,
"description": description
}), 400
else:
return jsonify({"error": "Failed to call Econet API"}), 500

@[Link]("/api/balance", methods=["POST"])
def account_balance_enquiry():
data = request.get_json()
msisdn = [Link]("msisdn")
provider_code = [Link]("provider_code")

if not msisdn or not provider_code:


return jsonify({"error": "Missing msisdn and provider_code"}), 400

api_params = {
"MSISDN": str(msisdn),
"ProviderCode": int(provider_code)
}

response = call_econet_api("account_balance_enquiry", api_params)


if response:
# Modified to handle direct array response
if isinstance(response, list):
balance_data = response[0]
return jsonify(balance_data), 200
else:
return jsonify({"error": "Unexpected response structure for
account_balance_enquiry"}), 500
else:
return jsonify({"error": "Failed to call Econet API"}), 500

@[Link]("/api/msisdn/validate", methods=["POST"])
def validate_msisdn():
data = request.get_json()
msisdn = [Link]("msisdn")
provider_code = [Link]("provider_code")

if not msisdn or not provider_code:


return jsonify({"error": "Missing msisdn and provider_code"}), 400

api_params = {
"MSISDN": str(msisdn),
"ProviderCode": int(provider_code)
}
response = call_econet_api("validate_msisdn", api_params)
if response:
status = [Link]("Status")
status_code = [Link]("StatusCode")
description = [Link]("Description")

return jsonify({
"status": status,
"statusCode": status_code,
"description": description
}), 200
else:
return jsonify({"error": "Failed to call Econet API"}), 500

@[Link]("/api/transaction/status", methods=["POST"])
def get_transaction_status():
data = request.get_json()
reference = [Link]("reference")

if not reference:
return jsonify({"error": "Missing reference"}), 400

api_params = {
"Reference": str(reference)
}
response = call_econet_api("get_transaction_status", api_params)

if response:
status = [Link]("Status")
status_code = [Link]("StatusCode")
description = [Link]("Description")
return jsonify({
"status": status,
"statusCode": status_code,
"description": description
}), 200
else:
return jsonify({"error": "Failed to call Econet API"}), 500

@[Link]("/api/account/transfer", methods=["POST"])
def account_transfer():
data = request.get_json()
from_company_id = [Link]("from_company_id")
to_company_id = [Link]("to_company_id")
currency = [Link]("currency")
amount = [Link]("amount")
reference = [Link]("reference") #Added Reference field

if not from_company_id or not to_company_id or not currency or not amount:


return jsonify({"error": "Missing required parameters for
account_transfer"}), 400

api_params = {
"FromCompanyID": int(from_company_id),
"ToCompanyID": int(to_company_id),
"Currency": int(currency),
"Amount": int(amount),
"Reference": str(reference) #Make sure you have a reference in the api
parameters for account transfer
}

response = call_econet_api("account_transfer", api_params)


if response:
status = [Link]("Status")
status_code = [Link]("StatusCode")
description = [Link]("Description")
return jsonify({
"status": status,
"statusCode": status_code,
"description": description
}), 200
else:
return jsonify({"error": "Failed to call Econet API"}), 500

if __name__ == "__main__":
[Link](debug=True)

You might also like