Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Fixes #339 and add support for object specific filtering #341

Merged
merged 8 commits into from
Sep 30, 2023
74 changes: 66 additions & 8 deletions src/xero/basemanager.py
Original file line number Diff line number Diff line change
@@ -1,8 +1,9 @@
import io
import json
import requests
from datetime import datetime
from datetime import date, datetime
from urllib.parse import parse_qs
from uuid import UUID
from xml.etree.ElementTree import Element, SubElement, tostring
from xml.parsers.expat import ExpatError

Expand Down Expand Up @@ -40,6 +41,35 @@ class BaseManager:
"Invoices": ["email", "online_invoice"],
"Organisations": ["actions"],
}
OBJECT_FILTER_FIELDS = {
"Invoices": {
"createdByMyApp": bool,
"summaryOnly": bool,
"IDs": list,
"InvoiceNumbers": list,
"ContactIDs": list,
"Statuses": list,
},
"PurchaseOrders": {"DateFrom": date, "DateTo": date, "Status": str},
"Quotes": {
"ContactID": UUID,
"ExpiryDateFrom": date,
"ExpiryDateTo": date,
"DateFrom": date,
"DateTo": date,
"Status": str,
"QuoteNumber": str,
},
"Journals": {"paymentsOnly": bool},
"Budgets": {"DateFrom": date, "DateTo": date},
"Contacts": {
"IDs": list,
"includeArchived": bool,
"summaryOnly": bool,
"searchTerm": str,
},
"TrackingCategories": {"includeArchived": bool},
}
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Document valid filtering fields for Objects along with some basic typing

DATETIME_FIELDS = (
"UpdatedDateUTC",
"Updated",
Expand Down Expand Up @@ -397,10 +427,19 @@ def _filter(self, **kwargs):
headers = self.prepare_filtering_date(val)
del kwargs["since"]

# Accept IDs parameter for Invoices and Contacts endpoints
if "IDs" in kwargs:
params["IDs"] = ",".join(kwargs["IDs"])
del kwargs["IDs"]
def get_filter_value(key, value, value_type=None):
if key in self.BOOLEAN_FIELDS or value_type == bool:
return "true" if value else "false"
elif key in self.DATE_FIELDS or value_type == date:
return f"{value.year}-{value.month}-{value.day}"
elif key in self.DATETIME_FIELDS or value_type == datetime:
return value.isoformat()
elif key.endswith("ID") or value_type == UUID:
return "%s" % (
value.hex if type(value) == UUID else UUID(value).hex
)
else:
return value
Comment on lines +441 to +453
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Helper function to convert filter fields to the format expected by xero


def get_filter_params(key, value):
last_key = key.split("_")[-1]
Expand Down Expand Up @@ -440,11 +479,30 @@ def generate_param(key, value):
field = field.replace("_", ".")
return fmt % (field, get_filter_params(key, value))

KNOWN_PARAMETERS = ["order", "offset", "page"]
object_params = self.OBJECT_FILTER_FIELDS.get(self.name, {})
LIST_PARAMETERS = list(
filter(lambda x: object_params[x] == list, object_params)
)
EXTRA_PARAMETERS = list(
filter(lambda x: object_params[x] != list, object_params)
)

Comment on lines +493 to +501
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Differentiated processing for basic fields and "list" fields

# Move any known parameter names to the query string
KNOWN_PARAMETERS = ["order", "offset", "page", "includeArchived"]
for param in KNOWN_PARAMETERS:
for param in KNOWN_PARAMETERS + EXTRA_PARAMETERS:
if param in kwargs:
params[param] = get_filter_value(
param, kwargs.pop(param), object_params.get(param, None)
)
Comment on lines +503 to +507
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Processing of simple filter fields, with some basic typed conversion of input data

# Support xero optimised list filtering; validate IDs we send but may need other validation
for param in LIST_PARAMETERS:
if param in kwargs:
params[param] = kwargs.pop(param)
if param.endswith("IDs"):
params[param] = ",".join(
map(lambda x: UUID(x).hex, kwargs.pop(param))
)
else:
params[param] = ",".join(kwargs.pop(param))
Comment on lines +509 to +516
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Processing of list filter fields, with no input verification except for IDs that are interpreted as UUIDs.


filter_params = []

Expand Down
14 changes: 10 additions & 4 deletions src/xero/exceptions.py
Original file line number Diff line number Diff line change
Expand Up @@ -50,10 +50,16 @@ def __init__(self, response):

elif response.headers["content-type"].startswith("text/html"):
payload = parse_qs(response.text)
self.errors = [payload["oauth_problem"][0]]
self.problem = self.errors[0]
super().__init__(response, payload["oauth_problem_advice"][0])

if payload:
Copy link
Contributor Author

@bpotard bpotard Sep 29, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Handles the undocumented format of error messages from xero when sending malformed UUIDs for a list filter request.

self.errors = [payload["oauth_problem"][0]]
self.problem = self.errors[0]
super().__init__(response, payload["oauth_problem_advice"][0])
else:
# Sometimes xero returns the error message as pure text
# Not sure how to validate this is always the case
self.errors = [response.text]
self.problem = self.errors[0]
super().__init__(response, response.text)
else:
# Extract the messages from the text.
# parseString takes byte content, not unicode.
Expand Down
2 changes: 1 addition & 1 deletion src/xero/manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,7 @@ def __init__(self, name, credentials, unit_price_4dps=False, user_agent=None):
from xero import __version__ as VERSION # noqa

self.credentials = credentials
self.name = name
self.name = name.capitalize() if name.islower() else name
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I had added this as a hack to make the tests behave "as expected". But unfortunately that breaks the behaviour for all objects that have more than one upper case letter, e.g. "PurchaseOrders", as it enforce only the first letter is capitalized.

Now that I think about it, it should really be the tests that should be updated to use the CapitalizedNames of the objects rather than a lower case version...

Copy link
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Agreed. The problem is with the test, not the manager.

self.base_url = credentials.base_url + XERO_API_URL
self.extra_params = {"unitdp": 4} if unit_price_4dps else {}
self.singular = singular(name)
Expand Down
9 changes: 7 additions & 2 deletions tests/test_manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -226,13 +226,18 @@ def test_filter_ids(self):
manager = Manager("contacts", credentials)

uri, params, method, body, headers, singleobject = manager._filter(
IDs=["1", "2", "3", "4", "5"]
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

These are actually invalid IDs, as only UUIDs are accepted. Sending this as input to the xero API results in a rather strange error message in an undocumented format. I suggest to run the tests on valid input, i.e. UUIDs.

IDs=[
"3e776c4b-ea9e-4bb1-96be-6b0c7a71a37f",
"12345678901234567890123456789012",
]
)

self.assertEqual(method, "get")
self.assertFalse(singleobject)

expected_params = {"IDs": "1,2,3,4,5"}
expected_params = {
"IDs": "3e776c4bea9e4bb196be6b0c7a71a37f,12345678901234567890123456789012"
}
self.assertEqual(params, expected_params)

def test_rawfilter(self):
Expand Down