KGX Utils¶

Utility methods that are reused across the codebase.

kgx.utils.kgx_utils¶

class kgx.utils.kgx_utils.GraphEntityType[source]¶

Bases: enum.Enum

An enumeration.

kgx.utils.kgx_utils.apply_edge_filters(graph: kgx.graph.base_graph.BaseGraph, edge_filters: Dict[str, Union[str, Set]]) → None[source]¶

Apply filters to graph and remove edges that do not pass given filters.

Parameters

graph (kgx.graph.base_graph.BaseGraph) – The graph
edge_filters (Dict[str, Union[str, Set]]) – Edge filters

kgx.utils.kgx_utils.apply_filters(graph: kgx.graph.base_graph.BaseGraph, node_filters: Dict[str, Union[str, Set]], edge_filters: Dict[str, Union[str, Set]]) → None[source]¶

Apply filters to graph and remove nodes and edges that do not pass given filters.

Parameters

graph (kgx.graph.base_graph.BaseGraph) – The graph
node_filters (Dict[str, Union[str, Set]]) – Node filters
edge_filters (Dict[str, Union[str, Set]]) – Edge filters

kgx.utils.kgx_utils.apply_graph_operations(graph: kgx.graph.base_graph.BaseGraph, operations: List) → None[source]¶

Apply graph operations to a given graph.

Parameters

graph (kgx.graph.base_graph.BaseGraph) – An instance of BaseGraph
operations (List) – A list of graph operations with configuration

kgx.utils.kgx_utils.apply_node_filters(graph: kgx.graph.base_graph.BaseGraph, node_filters: Dict[str, Union[str, Set]]) → None[source]¶

Apply filters to graph and remove nodes that do not pass given filters.

Parameters

graph (kgx.graph.base_graph.BaseGraph) – The graph
node_filters (Dict[str, Union[str, Set]]) – Node filters

kgx.utils.kgx_utils.camelcase_to_sentencecase(s: str) → str[source]¶

Convert CamelCase to sentence case.

Parameters: s (str) – Input string in CamelCase
Returns: string in sentence case form
Return type: str

kgx.utils.kgx_utils.contract(uri: str, prefix_maps: Optional[List[Dict]] = None, fallback: bool = True) → str[source]¶

Contract a given URI to a CURIE, based on mappings from prefix_maps. If no prefix map is provided then will use defaults from prefixcommons-py.

This method will return the URI as the CURIE if there is no mapping found.

Parameters

uri (str) – A URI
prefix_maps (Optional[List[Dict]]) – A list of prefix maps to use for mapping
fallback (bool) – Determines whether to fallback to default prefix mappings, as determined by prefixcommons.curie_util, when URI prefix is not found in prefix_maps.

Returns

A CURIE corresponding to the URI

Return type

str

kgx.utils.kgx_utils.current_time_in_millis()[source]¶

Get current time in milliseconds.

Returns: Time in milliseconds
Return type: int

kgx.utils.kgx_utils.expand(curie: str, prefix_maps: Optional[List[dict]] = None, fallback: bool = True) → str[source]¶

Expand a given CURIE to an URI, based on mappings from prefix_map.

This method will return the CURIE as the IRI if there is no mapping found.

Parameters

curie (str) – A CURIE
prefix_maps (Optional[List[dict]]) – A list of prefix maps to use for mapping
fallback (bool) – Determines whether to fallback to default prefix mappings, as determined by prefixcommons.curie_util, when CURIE prefix is not found in prefix_maps.

Returns

A URI corresponding to the CURIE

Return type

str

kgx.utils.kgx_utils.format_biolink_category(s: str) → str[source]¶

Convert a sentence case Biolink category name to a proper Biolink CURIE with the category itself in CamelCase form.

Parameters: s (str) – Input string in sentence case
Returns: a proper Biolink CURIE
Return type: str

kgx.utils.kgx_utils.generate_edge_identifiers(graph: kgx.graph.base_graph.BaseGraph)[source]¶

Generate unique identifiers for edges in a graph that do not have an id field.

Parameters: graph (kgx.graph.base_graph.BaseGraph) –

kgx.utils.kgx_utils.generate_edge_key(s: str, edge_predicate: str, o: str) → str[source]¶

Generates an edge key based on a given subject, predicate, and object.

Parameters

s (str) – Subject
edge_predicate (str) – Edge label
o (str) – Object

Returns

Edge key as a string

Return type

str

kgx.utils.kgx_utils.generate_uuid()[source]¶

Generates a UUID.

Returns: A UUID
Return type: str

kgx.utils.kgx_utils.get_biolink_ancestors(name: str)[source]¶

Get ancestors for a given Biolink class.

Parameters: name (str) –
Returns: A list of ancestors
Return type: List

kgx.utils.kgx_utils.get_biolink_element(name) → Optional[linkml_runtime.linkml_model.meta.Element][source]¶

Get Biolink element for a given name, where name can be a class, slot, or relation.

Parameters: name (str) – The name
Returns: An instance of linkml_model.meta.Element
Return type: Optional[linkml_model.meta.Element]

kgx.utils.kgx_utils.get_biolink_property_types() → Dict[source]¶

Get all Biolink property types. This includes both node and edges properties.

Returns: A dict containing all Biolink property and their types
Return type: Dict

kgx.utils.kgx_utils.get_cache(maxsize=10000)[source]¶

Get an instance of cachetools.cache

Parameters: maxsize (int) – The max size for the cache (10000, by default)
Returns: An instance of cachetools.cache
Return type: cachetools.cache

kgx.utils.kgx_utils.get_curie_lookup_service()[source]¶

Get an instance of kgx.curie_lookup_service.CurieLookupService

Returns: An instance of CurieLookupService
Return type: kgx.curie_lookup_service.CurieLookupService

kgx.utils.kgx_utils.get_prefix_prioritization_map() → Dict[str, List][source]¶

Get prefix prioritization map as defined in Biolink Model.

Returns
Return type: Dict[str, List]

kgx.utils.kgx_utils.get_toolkit(biolink_release: Optional[str] = None) → bmt.toolkit.Toolkit[source]¶

Get an instance of bmt.Toolkit If there no instance defined, then one is instantiated and returned.

Parameters: biolink_release (Optional[str]) – URL to (Biolink) Model Schema to be used for validated (default: None, use default Biolink Model Toolkit schema)

kgx.utils.kgx_utils.get_type_for_property(p: str) → str[source]¶

Get type for a property.

TODO: Move this to biolink-model-default_toolkit

Parameters: p (str) –
Returns: The type for a given property
Return type: str

kgx.utils.kgx_utils.is_null(item: Any) → bool[source]¶

Checks if a given item is null or correspond to null.

This method checks for: None, numpy.nan, pandas.NA, pandas.NaT, and ` `

Parameters: item (Any) – The item to check
Returns: Whether the given item is null or not
Return type: bool

kgx.utils.kgx_utils.prepare_data_dict(d1: Dict, d2: Dict, preserve: bool = True) → Dict[source]¶

Given two dict objects, make a new dict object that is the intersection of the two.

If a key is known to be multivalued then it’s value is converted to a list. If a key is already multivalued then it is updated with new values. If a key is single valued, and a new unique value is found then the existing value is converted to a list and the new value is appended to this list.

Parameters

d1 (Dict) – Dict object
d2 (Dict) – Dict object
preserve (bool) – Whether or not to preserve values for conflicting keys

Returns

The intersection of d1 and d2

Return type

Dict

kgx.utils.kgx_utils.remove_null(input: Any) → Any[source]¶

Remove any null values from input.

Parameters: input (Any) – Can be a str, list or dict
Returns: The input without any null values
Return type: Any

kgx.utils.kgx_utils.sanitize_import(data: Dict) → Dict[source]¶

Sanitize key-value pairs in dictionary.

Parameters: data (Dict) – A dictionary containing key-value pairs
Returns: A dictionary containing processed key-value pairs
Return type: Dict

kgx.utils.kgx_utils.sentencecase_to_camelcase(s: str) → str[source]¶

Convert sentence case to CamelCase.

Parameters: s (str) – Input string in sentence case
Returns: string in CamelCase form
Return type: str

kgx.utils.kgx_utils.sentencecase_to_snakecase(s: str) → str[source]¶

Convert sentence case to snake_case.

Parameters: s (str) – Input string in sentence case
Returns: string in snake_case form
Return type: str

kgx.utils.kgx_utils.snakecase_to_sentencecase(s: str) → str[source]¶

Convert snake_case to sentence case.

Parameters: s (str) – Input string in snake_case
Returns: string in sentence case form
Return type: str

kgx.utils.kgx_utils.validate_edge(edge: Dict) → Dict[source]¶

Given an edge as a dictionary, check for required properties. This method will return the edge dictionary with default assumptions applied, if any.

Parameters: edge (Dict) – An edge represented as a dict
Returns: An edge represented as a dict, with default assumptions applied.
Return type: Dict

kgx.utils.kgx_utils.validate_node(node: Dict) → Dict[source]¶

Given a node as a dictionary, check for required properties. This method will return the node dictionary with default assumptions applied, if any.

Parameters: node (Dict) – A node represented as a dict
Returns: A node represented as a dict, with default assumptions applied.
Return type: Dict