From 385f3831b298a84d1554bde1b1e95d5fd8bda3b3 Mon Sep 17 00:00:00 2001
From: Vadim Redkozubov <>
Date: Sun, 21 Jan 2018 21:25:53 +0200
Subject: [PATCH 1/2] .gitignore
---
.gitignore | 1 +
1 file changed, 1 insertion(+)
diff --git a/.gitignore b/.gitignore
index 5191521..1eb011e 100644
--- a/.gitignore
+++ b/.gitignore
@@ -5,3 +5,4 @@
dist
build/
doc/_build/
+venv/
From 0c16bd188b0962360914eacf20483f7703843144 Mon Sep 17 00:00:00 2001
From: Vadim Redkozubov <>
Date: Sun, 21 Jan 2018 23:03:16 +0200
Subject: [PATCH 2/2] ComplexType support
---
odata/metadata.py | 59 ++++++++++++++++++++++-----------
odata/tests/__init__.py | 9 +++++
odata/tests/demo_metadata.xml | 5 +++
odata/tests/test_complextype.py | 58 ++++++++++++++++++++++++++++++++
4 files changed, 112 insertions(+), 19 deletions(-)
create mode 100644 odata/tests/test_complextype.py
diff --git a/odata/metadata.py b/odata/metadata.py
index cc37b03..50ec0d7 100644
--- a/odata/metadata.py
+++ b/odata/metadata.py
@@ -330,6 +330,29 @@ def _parse_function(self, xmlq, function_element, schema_name):
function['return_type'] = type_name
return function
+ def _parse_property(self, xmlq, entity_pks, property_):
+
+ p_name = property_.attrib['Name']
+ p_type = property_.attrib['Type']
+
+ is_collection, p_type = self._type_is_collection(p_type)
+ is_computed_value = False
+
+ for annotation in xmlq(property_, 'edm:Annotation'):
+ annotation_term = annotation.attrib.get('Term', '')
+ annotation_bool = annotation.attrib.get('Bool') == 'true'
+ if annotation_term == self._annotation_term_computed:
+ is_computed_value = annotation_bool
+
+ return {
+ 'name': p_name,
+ 'type': p_type,
+ 'is_primary_key': p_name in entity_pks,
+ 'is_collection': is_collection,
+ 'is_computed_value': is_computed_value,
+ }
+
+
def _parse_entity(self, xmlq, entity_element, schema_name, schema_alias):
entity_name = entity_element.attrib['Name']
@@ -356,25 +379,7 @@ def _parse_entity(self, xmlq, entity_element, schema_name, schema_alias):
entity_pks[pk_property_name] = 0
for entity_property in xmlq(entity_element, 'edm:Property'):
- p_name = entity_property.attrib['Name']
- p_type = entity_property.attrib['Type']
-
- is_collection, p_type = self._type_is_collection(p_type)
- is_computed_value = False
-
- for annotation in xmlq(entity_property, 'edm:Annotation'):
- annotation_term = annotation.attrib.get('Term', '')
- annotation_bool = annotation.attrib.get('Bool') == 'true'
- if annotation_term == self._annotation_term_computed:
- is_computed_value = annotation_bool
-
- entity['properties'].append({
- 'name': p_name,
- 'type': p_type,
- 'is_primary_key': p_name in entity_pks,
- 'is_collection': is_collection,
- 'is_computed_value': is_computed_value,
- })
+ entity['properties'].append(self._parse_property(xmlq, entity_pks, entity_property))
for nav_property in xmlq(entity_element, 'edm:NavigationProperty'):
p_name = nav_property.attrib['Name']
@@ -409,6 +414,18 @@ def _parse_enumtype(self, xmlq, enumtype_element, schema_name):
})
return enum
+ def _parse_complextype(self, xmlq, complextype_element, schema_name):
+ complex_name = complextype_element.attrib['Name']
+ complex_ = {
+ 'name': complex_name,
+ 'fully_qualified_name': '.'.join([schema_name, complex_name]),
+ 'properties': []
+ }
+ for complex_property in xmlq(complextype_element, 'edm:Property'):
+ complex_['properties'].append(self._parse_property(xmlq, {}, complex_property))
+ return complex_
+
+
def parse_document(self, doc):
schemas = []
container_sets = {}
@@ -442,6 +459,10 @@ def xmlq(node, xpath):
entity = self._parse_entity(xmlq, entity_type, schema_name, schema_alias)
schema_dict['entities'].append(entity)
+ for complex_type in xmlq(schema, 'edm:ComplexType'):
+ complex_ = self._parse_complextype(xmlq, complex_type, schema_name)
+ schema_dict['complex_types'].append(complex_)
+
schemas.append(schema_dict)
for schema in xmlq(doc, 'edmx:DataServices/edm:Schema'):
diff --git a/odata/tests/__init__.py b/odata/tests/__init__.py
index 4a1f2db..b4bca98 100644
--- a/odata/tests/__init__.py
+++ b/odata/tests/__init__.py
@@ -4,6 +4,7 @@
from odata.property import StringProperty, IntegerProperty, DecimalProperty, \
NavigationProperty, DatetimeProperty
from odata.enumtype import EnumType, EnumTypeProperty
+from odata.complextype import ComplexType, ComplexTypeProperty
url = 'http://unittest.server.local/odata/'
Service = ODataService(url)
@@ -49,6 +50,13 @@ class ColorSelection(EnumType):
Green = 3
+class Dimensions(ComplexType):
+ properties = dict(Height=DecimalProperty,
+ Weight=StringProperty,
+ Length=DecimalProperty,
+ )
+
+
class Product(Service.Entity):
__odata_type__ = 'ODataTest.Objects.Product'
__odata_collection__ = 'ProductParts'
@@ -59,6 +67,7 @@ class Product(Service.Entity):
price = DecimalProperty('Price')
color_selection = EnumTypeProperty('ColorSelection',
enum_class=ColorSelection)
+ dimensions = ComplexTypeProperty('Dimensions', type_class=Dimensions)
DemoAction = DemoAction()
DemoCollectionAction = DemoCollectionAction()
diff --git a/odata/tests/demo_metadata.xml b/odata/tests/demo_metadata.xml
index cb98786..9802910 100644
--- a/odata/tests/demo_metadata.xml
+++ b/odata/tests/demo_metadata.xml
@@ -34,6 +34,11 @@
+
+
+
+
+
diff --git a/odata/tests/test_complextype.py b/odata/tests/test_complextype.py
new file mode 100644
index 0000000..3f19d45
--- /dev/null
+++ b/odata/tests/test_complextype.py
@@ -0,0 +1,58 @@
+# -*- coding: utf-8 -*-
+
+import json
+from decimal import Decimal
+from unittest import TestCase
+
+import requests
+import responses
+
+from odata.tests import Service, Product, Dimensions
+
+
+class TestComplextype(TestCase):
+
+ def test_insert_value(self):
+ def request_callback(request):
+ content = json.loads(request.body)
+ content['ProductID'] = 1
+ self.assertIn('Dimensions', content)
+ self.assertEqual(content.get('Dimensions'), {'Height': 10.1})
+ headers = {}
+ return requests.codes.ok, headers, json.dumps(content)
+
+ with responses.RequestsMock() as rsps:
+ rsps.add_callback(rsps.POST,
+ Product.__odata_url__(),
+ callback=request_callback,
+ content_type='application/json')
+
+ new_product = Product()
+ new_product.name = 'Test Product'
+ new_product.dimensions = Dimensions({'Height': 10.1})
+ Service.save(new_product)
+
+ def test_read_value(self):
+ expected_height = Decimal('11')
+ expected_weight = u'110 kg.'
+ expected_length = Decimal('8')
+ test_product_values = dict(
+ ProductID=1,
+ ProductName='Test Product',
+ Category='',
+ ColorSelection='Red',
+ Dimensions={'Height': float(expected_height),
+ 'Weight': expected_weight,
+ 'Length': float(expected_length)},
+ Price=0.0,
+ )
+ with responses.RequestsMock() as rsps:
+ rsps.add(rsps.GET, Product.__odata_url__(),
+ content_type='application/json',
+ json=dict(value=[test_product_values]))
+
+ product = Service.query(Product).get(1)
+ self.assertIsInstance(product.dimensions, Dimensions)
+ assert product.dimensions['Height'] == expected_height
+ assert product.dimensions['Weight'] == expected_weight
+ assert product.dimensions['Length'] == expected_length