Using XML-RPC with Magento

Sure, we could use a cushy SOAP library to communicate with Magento, but maybe you’d want to capitalize on the cacheability of XML-RPC, instead. Sure, we could use an XML-RPC library, but that would be less fun and, as engineers, we like knowing how stuff works. Magento is not for the faint of heart and knowing how to communicate with it at the low-level might be useful at some point.

import os
import json

import xml.etree.ElementTree as ET

import requests

# http://xmlrpc.scripting.com/spec.html

_HOSTNAME = os.environ['MAGENTO_HOSTNAME']
_USERNAME = os.environ['MAGENTO_USERNAME']
_PASSWORD = os.environ['MAGENTO_PASSWORD']

_URL = "http://" + _HOSTNAME + "/api/xmlrpc"

_HEADERS = {
    'Content-Type': 'text/xml',
}

def _pretty_print(results):
    print(json.dumps(
            results, 
            sort_keys=True,
            indent=4, 
            separators=(',', ': ')))

def _send_request(payload):
    r = requests.post(_URL, data=payload, headers=_HEADERS)
    r.raise_for_status()

    root = ET.fromstring(r.text)
    return root

def _send_array(session_id, method_name, args):

    data_parts = []
    for (type_name, value) in args:
        data_parts.append('<value><' + type_name + '>' + str(value) + '</' + type_name + '></value>')

    payload = """\
<?xml version='1.0'?>
<methodCall>
    <methodName>call</methodName>
    <params>
        <param>
            <value><string>""" + session_id + """\
</string></value>
        </param>
        <param>
            <value><string>""" + method_name + """\
</string></value>
        </param>
        <param>
            <value>
                <array>
                    <data>
                        """ + ''.join(data_parts) + """
                    </data>
                </array>
            </value>
        </param>
    </params>
</methodCall>
"""

    return _send_request(payload)

def _send_struct(session_id, method_name, args):
    struct_parts = []

    for (type_name, argument_name, argument_value) in args:
        struct_parts.append("<member><name>" + argument_name + "</name><value><" + type_name + ">" + str(argument_value) + "</" + type_name + "></value></member>")

    payload = """\
<?xml version='1.0'?>
<methodCall>
    <methodName>call</methodName>
    <params>
        <param>
            <value><string>""" + session_id + """\
</string></value>
        </param>
        <param>
            <value><string>""" + method_name + """\
</string></value>
        </param>
        <param>
            <value>
                <struct>
                    """ + ''.join(struct_parts) + """
                </struct>
            </value>
        </param>
    </params>
</methodCall>
"""

    return _send_request(payload)

def _send_login(args):
    param_parts = []
    for (type_name, value) in args:
        param_parts.append('<param><value><' + type_name + '>' + value + '</' + type_name + '></value></param>')

    payload = """\
<?xml version="1.0"?>
<methodCall>
    <methodName>login</methodName>
    <params>""" + ''.join(param_parts) + """\
</params>
</methodCall>
"""

    return _send_request(payload)


class XmlRpcFaultError(Exception):
    pass

def _distill(value_node):
    type_node = value_node[0]
    type_name = type_node.tag

    if type_name == 'nil':
        return None
    elif type_name in ('int', 'i4'):
        return int(type_node.text)
    elif type_name == 'boolean':
        return bool(type_node.text)
    elif type_name == 'double':
        return float(type_node.text)
    elif type_name == 'struct':
        values = {}
        for member_node in type_node:
            key = member_node.find('name').text

            value_node = member_node.find('value')
            value = _distill(value_node)

            values[key] = value

        return values
    elif type_name == 'array':
        flat = []
        for i, child_value_node in enumerate(type_node.findall('data/value')):
            flat.append(_distill(child_value_node))

        return flat
    elif type_name in ('string', 'dateTime.iso8601', 'base64'):
        return type_node.text
    else:
        raise ValueError("Invalid type: [{0}] [{1}]".format(type_name, type_node))

def _parse_response(root):
    if root.find('fault') is not None:
        for e in root.findall('fault/value/struct/member'):
            if e.find('name').text == 'faultString':
                message = e.find('value/string').text
                raise XmlRpcFaultError(message)

        raise ValueError("Malformed fault response")

    value_node = root.find('params/param/value')
    result = _distill(value_node)

    return result

def _main():
    args = [
        ('string', _USERNAME),
        ('string', _PASSWORD),
    ]

    root = _send_login(args)
    session_id = _parse_response(root)

    resource_name = 'catalog_product.info'

    args = [
        ('int', 'productId', '314'),
    ]

    root = _send_struct(session_id, resource_name, args)
    result = _parse_response(root)
    _pretty_print(result)

if __name__ == '__main__':
    _main()

Output:

{
    "apparel_type": "33",
    "categories": [
        "13"
    ],
    "category_ids": [
        "13"
    ],
    "color": "27",
    "country_of_manufacture": null,
    "created_at": "2013-03-05T00:48:15-05:00",
    "custom_design": null,
    "custom_design_from": null,
    "custom_design_to": null,
    "custom_layout_update": null,
    "description": "Two sash, convertible neckline with front ruffle detail. Unhemmed, visisble seams. Hidden side zipper. Unlined. Wool/elastane. Hand wash.",
    "fit": null,
    "gender": "94",
    "gift_message_available": null,
    "gift_wrapping_available": null,
    "gift_wrapping_price": null,
    "group_price": [],
    "has_options": "0",
    "image_label": null,
    "is_recurring": "0",
    "length": "82",
    "meta_description": null,
    "meta_keyword": null,
    "meta_title": null,
    "minimal_price": null,
    "msrp": null,
    "msrp_display_actual_price_type": "4",
    "msrp_enabled": "2",
    "name": "Convertible Dress",
    "news_from_date": "2013-03-01 00:00:00",
    "news_to_date": null,
    "occasion": "29",
    "old_id": null,
    "options_container": "container1",
    "page_layout": "one_column",
    "price": "340.0000",
    "product_id": "314",
    "recurring_profile": null,
    "required_options": "0",
    "set": "13",
    "short_description": "This all day dress has a flattering silhouette and a convertible neckline to suit your mood. Wear tied and tucked in a sailor knot, or reverse it for a high tied feminine bow.",
    "size": "72",
    "sku": "wsd017",
    "sleeve_length": "45",
    "small_image_label": null,
    "special_from_date": null,
    "special_price": null,
    "special_to_date": null,
    "status": "1",
    "tax_class_id": "2",
    "thumbnail_label": null,
    "tier_price": [],
    "type": "simple",
    "type_id": "simple",
    "updated_at": "2014-03-08 08:31:20",
    "url_key": "convertible-dress",
    "url_path": "convertible-dress-418.html",
    "visibility": "1",
    "websites": [
        "1"
    ],
    "weight": "1.0000"
}

You may download the code here.

Advertisements

Accessing Sales History in Magento From Python With XML-RPC

Install the python-magento package and simply pass in database table columns with some criteria. I had difficulty figuring out the structure of the filters because there’s so little documentation/examples anywhere online and, where there are, they all differ format/version.

So, here’s an example:

import magento
import json

_USERNAME = "apiuser"
_PASSWORD = "password"
_HOSTNAME = "dev.bugatchi.com"
_PORT = 80

m = magento.MagentoAPI(_HOSTNAME, _PORT, _USERNAME, _PASSWORD)

filters = {
    'created_at': {
        'from': '2013-05-29 12:38:43',
        'to': '2013-05-29 12:55:33',
    },
}

l = m.sales_order_invoice.list(filters)
print(json.dumps(l))

Output:

[
    {
        "created_at": "2013-05-29 12:38:43",
        "grand_total": "447.1400",
        "increment_id": "100000036",
        "invoice_id": "38",
        "order_currency_code": "USD",
        "order_id": "186",
        "state": "2"
    },
    {
        "created_at": "2013-05-29 12:52:44",
        "grand_total": "333.2100",
        "increment_id": "100000037",
        "invoice_id": "39",
        "order_currency_code": "USD",
        "order_id": "185",
        "state": "2"
    },
    {
        "created_at": "2013-05-29 12:55:33",
        "grand_total": "432.2600",
        "increment_id": "100000038",
        "invoice_id": "40",
        "order_currency_code": "USD",
        "order_id": "184",
        "state": "2"
    }
]

Note that debugging is fairly simple since a failure will return the failed query (unless the server is configured not to). So, you can use that to feel out many of the column names and comparison operators.