Skip to content

chore: refactor remaining if/isinstance chains to match/case #318

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.

Sign up for GitHub

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

Open
phanak-sap wants to merge 1 commit into
base: master
Choose a base branch
from
Open

chore: refactor remaining if/isinstance chains to match/case #318

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
41 changes: 21 additions & 20 deletions pyodata/v2/model.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2442,15 +2442,15 @@ def from_etree(target, annotation_node):
search_supported = False
params_node = None
for prop_value in annotation_node.xpath('edm:Record/edm:PropertyValue', namespaces=ANNOTATION_NAMESPACES):
rprop = prop_value.get('Property')
if rprop == 'Label':
label = prop_value.get('String')
elif rprop == 'CollectionPath':
collection_path = prop_value.get('String')
elif rprop == 'SearchSupported':
search_supported = prop_value.get('Bool')
elif rprop == 'Parameters':
params_node = prop_value
match prop_value.get('Property'):
case 'Label':
label = prop_value.get('String')
case 'CollectionPath':
collection_path = prop_value.get('String')
case 'SearchSupported':
search_supported = prop_value.get('Bool')
case 'Parameters':
params_node = prop_value

value_helper = ValueHelper(target, collection_path, label, search_supported)

Expand Down Expand Up @@ -2536,11 +2536,11 @@ def from_etree(value_help_parameter_node):
local_prop_name = None
list_prop_name = None
for pval in value_help_parameter_node.xpath('edm:PropertyValue', namespaces=ANNOTATION_NAMESPACES):
pv_name = pval.get('Property')
if pv_name == 'LocalDataProperty':
local_prop_name = pval.get('PropertyPath')
elif pv_name == 'ValueListProperty':
list_prop_name = pval.get('String')
match pval.get('Property'):
case 'LocalDataProperty':
local_prop_name = pval.get('PropertyPath')
case 'ValueListProperty':
list_prop_name = pval.get('String')

return ValueHelperParameter(direction, local_prop_name, list_prop_name)

Expand Down Expand Up @@ -2707,12 +2707,13 @@ def config(self):
def build(self):
""" Build model from the XML metadata"""

if isinstance(self._xml, str):
mdf = io.StringIO(self._xml)
elif isinstance(self._xml, bytes):
mdf = io.BytesIO(self._xml)
else:
raise TypeError(f'Expected bytes or str type on metadata_xml, got : {type(self._xml)}')
match self._xml:
case str():
mdf = io.StringIO(self._xml)
case bytes():
mdf = io.BytesIO(self._xml)
case _:
raise TypeError(f'Expected bytes or str type on metadata_xml, got : {type(self._xml)}')

namespaces = self._config.namespaces

Expand Down
93 changes: 40 additions & 53 deletions pyodata/v2/service.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -861,33 +861,23 @@ def __init__(self, service, entity_set, entity_type, proprties=None, entity_key=
# entity type of navigation property
prop_etype = prop.to_role.entity_type

# cache value according to multiplicity
if prop.to_role.multiplicity in \
[model.EndRole.MULTIPLICITY_ONE,
model.EndRole.MULTIPLICITY_ZERO_OR_ONE]:

# cache None in case we receive nothing (null) instead of entity data
if proprties[prop.name] is None:
self._cache[prop.name] = None
else:
self._cache[prop.name] = EntityProxy(service, None, prop_etype, proprties[prop.name])

elif prop.to_role.multiplicity == model.EndRole.MULTIPLICITY_ZERO_OR_MORE:
# default value is empty array
self._cache[prop.name] = []

# if there are no entities available, received data consists of
# metadata properties only.
if 'results' in proprties[prop.name]:
# available entities are serialized in results array
for entity in proprties[prop.name]['results']:
self._cache[prop.name].append(EntityProxy(service, None, prop_etype, entity))
elif isinstance(proprties[prop.name], list):
for entity in proprties[prop.name]:
self._cache[prop.name].append(EntityProxy(service, None, prop_etype, entity))
else:
raise PyODataException('Unknown multiplicity {0} of association role {1}'
.format(prop.to_role.multiplicity, prop.to_role.name))
match prop.to_role.multiplicity:
case model.EndRole.MULTIPLICITY_ONE | model.EndRole.MULTIPLICITY_ZERO_OR_ONE:
if proprties[prop.name] is None:
self._cache[prop.name] = None
else:
self._cache[prop.name] = EntityProxy(service, None, prop_etype, proprties[prop.name])
case model.EndRole.MULTIPLICITY_ZERO_OR_MORE:
self._cache[prop.name] = []
if 'results' in proprties[prop.name]:
for entity in proprties[prop.name]['results']:
self._cache[prop.name].append(EntityProxy(service, None, prop_etype, entity))
elif isinstance(proprties[prop.name], list):
for entity in proprties[prop.name]:
self._cache[prop.name].append(EntityProxy(service, None, prop_etype, entity))
case _:
raise PyODataException('Unknown multiplicity {0} of association role {1}'
.format(prop.to_role.multiplicity, prop.to_role.name))

# build entity key if not provided
if self._entity_key is None:
Expand Down Expand Up @@ -1702,32 +1692,29 @@ def function_import_handler(fimport, response):

response_data = response.json()['d']

# 1. if return type is an entity type or collection, resolve the entity set once
if isinstance(fimport.return_type, model.EntityType | model.Collection):
entity_set = self._service.schema.entity_set(fimport.entity_set_name)

if isinstance(fimport.return_type, model.EntityType):
return EntityProxy(self._service, entity_set, fimport.return_type, response_data)

if isinstance(fimport.return_type, model.Collection):
total_count = None
next_url = None
if '__count' in response_data:
total_count = int(response_data['__count'])
if '__next' in response_data:
next_url = response_data['__next']
results = response_data.get('results')
if results is None:
raise PyODataException(
f'Function import {fimport.name} returned a Collection response without a "results" key')
collection = ListWithTotalCount(total_count, next_url)
collection_item_type = fimport.return_type.item_type
for entity in results:
collection.append(EntityProxy(self._service, entity_set, collection_item_type, entity))
return collection

# 2. return raw data for all other return types (primitives, complex types encoded in dicts, etc.)
return response_data
match fimport.return_type:
case model.EntityType():
entity_set = self._service.schema.entity_set(fimport.entity_set_name)
return EntityProxy(self._service, entity_set, fimport.return_type, response_data)
case model.Collection():
entity_set = self._service.schema.entity_set(fimport.entity_set_name)
total_count = None
next_url = None
if '__count' in response_data:
total_count = int(response_data['__count'])
if '__next' in response_data:
next_url = response_data['__next']
results = response_data.get('results')
if results is None:
raise PyODataException(
f'Function import {fimport.name} returned a Collection response without a "results" key')
collection = ListWithTotalCount(total_count, next_url)
collection_item_type = fimport.return_type.item_type
for entity in results:
collection.append(EntityProxy(self._service, entity_set, collection_item_type, entity))
return collection
case _:
return response_data

return FunctionRequest(self._service.url, self._service.connection,
partial(function_import_handler, fimport), fimport,
Expand Down
Loading