Skip to content

Commit

Permalink
refactor get_aliases to have one exit point and to always return the …
Browse files Browse the repository at this point in the history
…input 'seed' identifier as its own alias; avoids need for gateway code in caller
  • Loading branch information
RichardBruskiewich committed Jun 23, 2023
1 parent 0042e65 commit 19b28d0
Show file tree
Hide file tree
Showing 2 changed files with 45 additions and 50 deletions.
8 changes: 0 additions & 8 deletions reasoner_validator/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -725,20 +725,12 @@ def case_input_found_in_response(
nodes: Dict = knowledge_graph["nodes"]
subject_id = case["subject_id"] if "subject_id" in case else case["subject"]
subject_aliases = get_aliases(subject_id)
if not subject_aliases:
# in the off chance that node normalization doesn't know
# about the given subject id, then simply use it directly
subject_aliases.append(subject_id)
if not self.case_node_found("subject", subject_aliases, case, nodes):
# 'subject' node identifier not found?
return False

object_id = case["object_id"] if "object_id" in case else case["object"]
object_aliases = get_aliases(object_id)
if not object_aliases:
# in the off chance that node normalization doesn't know
# about the given subject id, then simply use it directly
object_aliases.append(object_id)
if not self.case_node_found("object", object_aliases, case, nodes):
# 'object' node identifier not found?
return False
Expand Down
87 changes: 45 additions & 42 deletions reasoner_validator/sri/util.py
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,7 @@ def get_aliases(identifier: str) -> List[str]:
if not identifier:
raise RuntimeError("get_aliases(): empty input identifier?")

aliases: Optional[List[str]] = None
#
# TODO: maybe check for IRI's here and attempt to translate
# (Q: now do we access the prefix map from BMT to do this?)
Expand All @@ -49,51 +50,53 @@ def get_aliases(identifier: str) -> List[str]:
# We won't raise a RuntimeError for other various
# erroneous runtime conditions but simply report warnings
if not is_curie(identifier):

logging.warning(f"get_aliases(): identifier '{identifier}' is not a CURIE thus cannot resolve its aliases?")
return list()

# Use the Translator Node Normalizer service to resolve the identifier clique
response = requests.get(NODE_NORMALIZER_URL, params={'curie': identifier})

if response.status_code != 200:
logging.warning(f"get_aliases(): unsuccessful Node Normalizer HTTP call, status code: {response.status_code}")
return list()
else:
try:
result: Dict = response.json()
except (JSONDecodeError, UnicodeDecodeError) as je:
logging.warning(f"get_aliases(): Node Normalizer response JSON could not be decoded: {str(je)}?")
return list()

if result and identifier not in result.keys():
logging.warning(f"get_aliases(): Node Normalizer didn't return the identifier '{identifier}' clique?")
return list()

clique = result[identifier]

aliases: List[str] = list()
if clique:
if "id" in clique.keys():
# TODO: Don't need the canonical identifier for method
# but when you do, this is how you'll get it?
# clique_id = clique["id"]
# preferred_curie = preferred_id["identifier"]
# preferred_name = preferred_id["label"]
if "equivalent_identifiers" in clique.keys():
aliases: List[str] = [entry["identifier"] for entry in clique["equivalent_identifiers"]]
#
# Decided that it was a bad idea to remove
# the original identifier from the aliases...
# aliases.remove(identifier)
#
# print(dumps(aliases, indent=2))
else:
logging.warning(f"get_aliases(): missing the 'equivalent identifiers' for the '{identifier}' clique?")
# Use the Translator Node Normalizer service to resolve the identifier clique
response = requests.get(NODE_NORMALIZER_URL, params={'curie': identifier})

result: Optional[Dict] = None
if response.status_code != 200:
logging.warning(
f"get_aliases(): unsuccessful Node Normalizer HTTP call, status code: {response.status_code}"
)
else:
logging.warning(f"get_aliases(): missing the preferred 'id' for the '{identifier}' clique?")
else:
logging.warning(f"get_aliases(): '{identifier}' is a singleton in its clique thus has no aliases...")
try:
result = response.json()
except (JSONDecodeError, UnicodeDecodeError) as je:
logging.warning(f"get_aliases(): Node Normalizer response JSON could not be decoded: {str(je)}?")

if result:
if identifier not in result.keys():
logging.warning(f"get_aliases(): Node Normalizer didn't return the identifier '{identifier}' clique?")
else:
clique = result[identifier]
if clique:
if "id" in clique.keys():
# TODO: Don't need the canonical identifier for method
# but when you do, this is how you'll get it?
# clique_id = clique["id"]
# preferred_curie = preferred_id["identifier"]
# preferred_name = preferred_id["label"]
if "equivalent_identifiers" in clique.keys():
aliases = [entry["identifier"] for entry in clique["equivalent_identifiers"]]
#
# Decided that it was a bad idea to remove
# the original identifier from the aliases...
# aliases.remove(identifier)
#
# print(dumps(aliases, indent=2))
else:
logging.warning(f"get_aliases(): missing the 'equivalent identifiers' for the '{identifier}' clique?")
else:
logging.warning(f"get_aliases(): missing the preferred 'id' for the '{identifier}' clique?")
else:
logging.warning(f"get_aliases(): '{identifier}' is a singleton in its clique thus has no aliases...")

if not aliases:
# Logging various errors but always
# return the identifier as its own alias
aliases = [identifier]

return aliases

Expand Down

0 comments on commit 19b28d0

Please sign in to comment.