bento_meta.mdb

This module contains MDB, with machinery for efficiently querying a Neo4j instance of a Metamodel Database.

Submodules

Classes

`MDB`	A class representing a Metamodel Database.
`SearchableMDB`	`bento_meta.mdb.MDB` subclass for searching fulltext indices on an MDB.
`WriteableMDB`	`bento_meta.mdb.MDB` subclass for writing to an MDB.

Functions

`load_mdf`(→ None)	Load an MDF object into an MDB instance.
`load_model`(→ None)	Load a model object into an MDB instance.
`load_model_statements`(...)	Create Cypher statements from a model to load it de novo into an MDB instance.
`make_nanoid`(→ str)	Create a random nanoid and return it as a string.
`read_txn`(→ collections.abc.Callable[Concatenate[MDB, ...)	Decorate a query function to run a read transaction based on its query.
`read_txn_data`(...)	Decorate a query function to run a read transaction based on its query.
`read_txn_value`(...)	Decorate a query function to run a read transaction based on its query.

Package Contents

bento_meta.mdb.load_mdf(mdf: MDFProtocol, mdb: bento_meta.mdb.writeable.WriteableMDB, _commit: str | None = None) → None[source]: Load an MDF object into an MDB instance.

bento_meta.mdb.load_model(model: bento_meta.model.Model, mdb: bento_meta.mdb.writeable.WriteableMDB, _commit: str | None = None) → None[source]: Load a model object into an MDB instance.

bento_meta.mdb.load_model_statements(model: bento_meta.model.Model, _commit: str | None = None) → list[minicypher.statement.Statement][source]

Create Cypher statements from a model to load it de novo into an MDB instance.

:param mdb.Model model: Model instance for loading :param str _commit: ‘Commit string’ for marking entities in DB. If set,

this will override _commit attributes already existing on Model entities.

class bento_meta.mdb.MDB(uri: str | None = None, user: str | None = None, password: str | None = None)[source]

A class representing a Metamodel Database.

uri

user

password

driver: neo4j.Driver | None = None

models: dict[str, list[str]]

latest_version: dict[str, str | None]

_txfns

close() → None[source]: Close the driver connection.

register_txfn(name: str, fn: collections.abc.Callable) → None[source]

Register a transaction function with the class for later use.

See https://neo4j.com/docs/api/python-driver/current/api.html#managed-transactions-transaction-functions

Parameters:

name – Name to register the function under.
fn – The transaction function to register.

get_model_info() → tuple[str, None, str][source]: Get models, versions, and latest versions from MDB Model nodes.

get_model_handles() → list[str][source]

Return a simple list of model handles available.

Queries Model nodes (not model properties in Entity nodes).

Returns:: List of model handle strings.

get_model_versions(model: str) → list[str] | None[source]

Get list of version strings present in database for a given model.

Parameters:: model – Model handle to get versions for.
Returns:: List of version strings, or None if model not found.

get_latest_version(model: str) → str | None[source]

Get the version string from Model node marked is_latest:True for a model handle.

Parameters:: model – Model handle to get latest version for.
Returns:: Version string, or None if model not found.

get_model_nodes(model: str | None = None) → list[dict[str, Any]] | None[source]

Return a list of dicts representing Model nodes.

Returns all versions.

Parameters:: model – Optional model handle to filter by.
Returns:: List of dicts representing Model nodes, or None if not found.

get_nodes_by_model(model: str | None = None, version: str | None = None) → list[dict[str, Any]] | None[source]

Get all nodes for a given model.

Parameters:

model – Model handle to get nodes for. If None, get all nodes in database.
version – Version to filter by. If None, get nodes from model version marked is_latest:true. If ‘*’, get nodes from all model versions.

Returns:

List of node dicts.

get_model_nodes_edges(model: str, version: str | None = None) → list[dict[str, Any]] | None[source]

Get all node-relationship-node paths for a given model and version.

Parameters:

model – Model handle to get paths for.
version – Version to filter by. If None, use version marked is_latest:true for model. If ‘*’, retrieve from all versions.

Returns:

List of path dicts.

get_node_edges_by_node_id(nanoid: str) → list[dict[str, str]][source]

Get incoming and outgoing relationship information for a node from its nanoid.

Parameters:: nanoid – The nanoid of the node to get edges for.
Returns:: List of dicts with id, handle, model, version, near_type, far_type, rln, far_node.

get_node_and_props_by_node_id(nanoid: str) → list[dict[str, Any]] | None[source]

Get a node and its properties, given the node nanoid.

Parameters:: nanoid – The nanoid of the node to get.
Returns:: List with dict containing id, handle, model, version, node, props[].

get_nodes_and_props_by_model(model: str | None = None, version: str | None = None) → list[dict[str, Any]] | None[source]

Get all nodes with associated properties given a model handle.

Parameters:

model – Model handle to get nodes for. If None, get all nodes with their properties.
version – Version to filter by. If None, get nodes and props from model version marked is_latest:true. If ‘*’, get nodes and props from all model versions.

Returns:

List of dicts with id, handle, model, version, props[].

get_prop_node_and_domain_by_prop_id(nanoid: str) → list[dict[str, Any]] | None[source]

Get a property, its node, and its value domain or value set of terms by nanoid.

Parameters:: nanoid – The nanoid of the property to get.
Returns:: List with dict containing id, handle, model, version, value_domain, prop, node, value_set, terms[].

get_valueset_by_id(nanoid: str) → list[dict[str, Any]] | None[source]

Get a valueset with the properties that use it and the terms that constitute it.

Parameters:: nanoid – The nanoid of the valueset to get.
Returns:: List with dict containing id, handle, url, terms[], props[].

get_valuesets_by_model(model: str | None = None, version: str | None = None) → list[dict[str, Any]] | None[source]

Get all valuesets that are used by properties in the given model and version.

Gets all valuesets if model is None. Also returns list of properties using each valueset.

Parameters:

model – Model handle to get valuesets for. If None, get all valuesets.
version – Version to filter by. If None, get value sets associated with latest model version. If ‘*’, get those associated with all versions.

Returns:

List of dicts with value_set, props[].

get_term_by_id(nanoid: str) → list[dict[str, Any]] | None[source]

Get a term having the given nanoid, with its origin.

Parameters:: nanoid – The nanoid of the term to get.
Returns:: Dict with term, origin.

get_props_and_terms_by_model(model: str | None = None, version: str | None = None) → list[dict[str, Any]] | None[source]

Get terms from valuesets associated with properties in a model and version.

Gets all terms if model is None.

Parameters:

model – Model handle to get props and terms for. If None, get all terms.
version – Version to filter by. If None, get props and terms from the latest model version. If ‘*’, get those from all versions.

Returns:

List of dicts with prop, terms[].

get_origins() → list[dict[str, Any]] | None[source]

Get all origins.

Returns:: List of origin dicts.

get_origin_by_id(oid: str) → list[dict[str, Any]] | None[source]: Get an origin by nanoid.

get_tags_for_entity_by_id(nanoid: str) → list[dict[str, Any]] | None[source]

Get all tags attached to an entity, given the entity’s nanoid.

Parameters:: nanoid – The nanoid of the entity to get tags for.
Returns:: List with dict containing model(str), tags[].

get_tags_and_values(key: str | None = None) → list[dict[str, Any]] | None[source]

Get all tag key/value pairs that are present in database.

Parameters:: key – Optional key to filter by.
Returns:: values[].
Return type:: List of dicts with key(str)

get_entities_by_tag(key: str, value: str | None = None) → list[dict[str, Any]] | None[source]

Get all entities, tagged with a given key or key:value pair.

Parameters:

key – Tag key to filter by.
value – Optional tag value to filter by.

Returns:

List of dicts with tag_key(str), tag_value(str), entity(str - label), entities[].

get_with_statement(qry: str, parms: dict[str, Any] | None = None) → list[dict[str, Any]] | None[source]: Run an arbitrary read statement and return data.

bento_meta.mdb.make_nanoid(alphabet: str = 'abcdefghijkmnopqrstuvwxyzABCDEFGHJKMNPQRSTUVWXYZ0123456789', size: int = 6) → str[source]: Create a random nanoid and return it as a string.

bento_meta.mdb.read_txn(func: collections.abc.Callable[Concatenate[MDB, P], tuple[str, dict[str, Any] | None]]) → collections.abc.Callable[Concatenate[MDB, P], list[neo4j.Record]][source]

Decorate a query function to run a read transaction based on its query.

Query function should return a tuple (qry_string, param_dict).

Parameters:: func – The query function to decorate.
Returns:: Decorated function that returns list of driver Records.

bento_meta.mdb.read_txn_data(func: collections.abc.Callable[Concatenate[MDB, P], tuple[str, dict[str, Any] | None]]) → collections.abc.Callable[Concatenate[MDB, P], list[dict[str, Any]] | None][source]

Decorate a query function to run a read transaction based on its query.

Query function should return a tuple (qry_string, param_dict).

Parameters:: func – The query function to decorate.
Returns:: Decorated function that returns records as a list of simple dicts.

bento_meta.mdb.read_txn_value(func: collections.abc.Callable[Concatenate[MDB, P], tuple[str, dict[str, Any] | None, str]]) → collections.abc.Callable[Concatenate[MDB, P], list[Any]][source]

Decorate a query function to run a read transaction based on its query.

Query function should return a tuple (qry_string, param_dict, values_key).

Parameters:: func – The query function to decorate.
Returns:: Decorated function that returns list of values for key specified by query function.

class bento_meta.mdb.SearchableMDB(uri: str | None = None, user: str | None = None, password: str | None = None)[source]

Bases: bento_meta.mdb.MDB

bento_meta.mdb.MDB subclass for searching fulltext indices on an MDB.

ftindexes

available_indexes() → dict[str, dict[str, list[str]]][source]

Fulltext indexes present in database.

Returns:: Dict mapping index_name to dict with entity_type (NODE|RELATIONSHIP), entities ([labels]), properties ([[props]]).

query_index(index: str, qstring: str, skip: str | None = None, limit: str | None = None) → list[dict[str, Any]] | None[source]

Query a named fulltext index of nodes or relationships.

Parameters:

index – Name of the fulltext index to query.
qstring – Lucene query string.
skip – Number of results to skip.
limit – Maximum number of results to return.

Returns:

List of dicts with ent (entity dict), label, score (lucene score).

search_entity_handles(qstring: str) → dict[str, list[dict[str, Any]]] | None[source]

Fulltext search of qstring over node, relationship, and property handles.

Parameters:: qstring – Lucene query string.
Returns:: Dict with nodes, relationships, properties, each containing list of dicts with ent (entity dict) and score (lucene score).

search_terms(qstring: str, *, search_values: bool = True, search_definitions: bool = True) → list[dict[str, Any]] | None[source]

Fulltext search for qstring over terms, by value, definition, or both (default).

Parameters:

qstring – Lucene query string.
search_values – If True, search term values.
search_definitions – If True, search term definitions.

Returns:

List of dicts with ent (term dict) and score (lucene score).

class bento_meta.mdb.WriteableMDB(uri: str | None = None, user: str | None = None, password: str | None = None)[source]

Bases: bento_meta.mdb.mdb.MDB

bento_meta.mdb.MDB subclass for writing to an MDB.

put_with_statement(qry: str, parms: dict[str, Any] | None = None) → list[neo4j.Record][source]: Run an arbitrary write statement.

put_term_with_origin(term: bento_meta.objects.Term, commit: str = '', _from: int = 1) → list[neo4j.Record][source]

Merge a bento-meta Term object, that has an Origin object set into an MDB.

If a new term is created, assign a random 6-char nanoid to it. The Origin must already be represented in the database.

Parameters:

term – Term object to merge.
commit – GitHub commit SHA1 associated with the term (if any).
_from – Source identifier.

Returns:

List of Records from the transaction.