mex.backend.graph package¶

Submodules¶

mex.backend.graph.connector module¶

class mex.backend.graph.connector.GraphConnector¶

Bases: BaseConnector

Connector to handle authentication and transactions with the graph database.

__init__() → None¶: Create a new graph database connection.

_check_connectivity_and_authentication() → Result¶

Check the connectivity and authentication to the graph.

Raises:: IncompatibleVersionError – If the connected neo4j is older than MIN_NEO4J_VERSION.

_check_merge_preconditions_tx(tx: Transaction, goner: MergedAccessPlatform | MergedActivity | MergedBibliographicResource | MergedConsent | MergedContactPoint | MergedDistribution | MergedLocation | MergedOrganization | MergedOrganizationalUnit | MergedPerson | MergedPrimarySource | MergedResource | MergedResourceSeries | MergedVariable | MergedVariableGroup, keeper: MergedAccessPlatform | MergedActivity | MergedBibliographicResource | MergedConsent | MergedContactPoint | MergedDistribution | MergedLocation | MergedOrganization | MergedOrganizationalUnit | MergedPerson | MergedPrimarySource | MergedResource | MergedResourceSeries | MergedVariable | MergedVariableGroup) → None¶: Raise an error when the preconditions for performing a merge aren’t met.

_fetch_extracted_or_rule_items(query_string: str | None, identifier: str | None, entity_type: Sequence[__annotationlib_name_1__], reference_filters: Sequence[ReferenceFilter] | None, skip: int, limit: int) → Result¶

Query the graph for extracted or rule items.

Parameters:

query_string – Optional full text search query term
identifier – Optional identifier filter
entity_type – List of allowed entity types
reference_filters – Optional reference field filters
skip – How many items to skip for pagination
limit – How many items to return at most

Returns:

Graph result instance

_init_driver() → Driver¶: Initialize and return a database driver.

Take exclusive write locks on the merged items taking part in a merge.

Reads in neo4j’s read-committed isolation take no locks, so the precondition check has to run behind these locks to not act on a stale result.

_merge_items_tx(tx: Transaction, goner: MergedAccessPlatform | MergedActivity | MergedBibliographicResource | MergedConsent | MergedContactPoint | MergedDistribution | MergedLocation | MergedOrganization | MergedOrganizationalUnit | MergedPerson | MergedPrimarySource | MergedResource | MergedResourceSeries | MergedVariable | MergedVariableGroup, keeper: MergedAccessPlatform | MergedActivity | MergedBibliographicResource | MergedConsent | MergedContactPoint | MergedDistribution | MergedLocation | MergedOrganization | MergedOrganizationalUnit | MergedPerson | MergedPrimarySource | MergedResource | MergedResourceSeries | MergedVariable | MergedVariableGroup) → None¶: Run all required merging steps in a single transaction.

_run(query: Query, /, tx: Transaction | None = None, access_mode: str = 'READ', **parameters: Any) → Result¶

Run a query in the given transaction, or in a fresh session when none given.

Reads that are part of a bigger write operation need to see that operation’s uncommitted state, so they have to run in its transaction instead of opening a session of their own.

Parameters:

query – Query instance to render and run
tx – Optional open transaction to run the query in
access_mode – Access mode to use when no transaction is given
parameters – Keyword parameters for the query

Returns:

Graph result instance

_run_ingest_in_transaction(tx: Transaction, model: ExtractedAccessPlatform | ExtractedActivity | ExtractedBibliographicResource | ExtractedConsent | ExtractedContactPoint | ExtractedDistribution | ExtractedLocation | ExtractedOrganization | ExtractedOrganizationalUnit | ExtractedPerson | ExtractedPrimarySource | ExtractedResource | ExtractedResourceSeries | ExtractedVariable | ExtractedVariableGroup | AccessPlatformRuleSetResponse | ActivityRuleSetResponse | BibliographicResourceRuleSetResponse | ConsentRuleSetResponse | ContactPointRuleSetResponse | DistributionRuleSetResponse | LocationRuleSetResponse | OrganizationRuleSetResponse | OrganizationalUnitRuleSetResponse | PersonRuleSetResponse | PrimarySourceRuleSetResponse | ResourceRuleSetResponse | ResourceSeriesRuleSetResponse | VariableRuleSetResponse | VariableGroupRuleSetResponse | ExtractedPrimarySourceWithHardcodedIdentifiers) → None¶: Ingest a single item in a database transaction.

_seed_constraints() → None¶: Ensure property constraints are created for all entity types.

_seed_data() → None¶: Ensure the primary source mex is seeded and linked to itself.

_seed_indices() → Result¶: Ensure there is a full text search index for all searchable fields.

close() → None¶: Close the connector’s underlying requests session.

commit(query: Query, /, access_mode: str = 'READ', **parameters: Any) → Result¶

Send and commit a single graph transaction with retry configuration.

Parameters:

query – The query string or Query object to execute
access_mode – Whether to run the query with read or write access
**parameters – Query parameters to substitute in the Cypher query

Returns:

Result object containing query execution results and metadata

delete_item(identifier: str) → Result¶: Delete a merged item including all extracted items and rule-sets.

delete_rule_set(stable_target_id: str) → Result¶

Delete a rule-set by stableTargetId.

Deletes all additive, subtractive, preventive, and workflow rules connected to the given stableTargetId, along with their nested items and outbound connections.

exists_item(identifier: Identifier, entity_types: Sequence[__annotationlib_name_1__]) → bool¶

Validate whether an item with the given identifier and entity type exists.

Parameters:

identifier – Identifier of the to-be-checked item
entity_types – Allowed entity types of the to-be-checked item

Returns:

Boolean representing the existence of the requested item

fetch_extracted_items(query_string: str | None, identifier: str | None, entity_type: Sequence[__annotationlib_name_1__] | None, reference_filters: Sequence[ReferenceFilter] | None, skip: int, limit: int) → Result¶

Query the graph for extracted items.

Parameters:

query_string – Optional full text search query term
identifier – Optional identifier filter
entity_type – Optional entity type filter
reference_filters – Optional reference field filters
skip – How many items to skip for pagination
limit – How many items to return at most

Returns:

Graph result instance

fetch_identities(had_primary_source: Identifier | None = None, identifier_in_primary_source: str | None = None, stable_target_id: Identifier | None = None, limit: int = 1000) → Result¶

Search the graph for nodes matching the given ID combination.

Identity queries can be filtered by stable_target_id, had_primary_source or identifier_in_primary_source.

Parameters:

had_primary_source – The stableTargetId of a connected PrimarySource
identifier_in_primary_source – The id the item had in its primary source
stable_target_id – The stableTargetId of an item
limit – How many results to return, defaults to 1000

Returns:

A graph result set containing identities

Query the graph for merged items.

Parameters:

query_string – Optional full text search query term
identifier – Optional merged item identifier filter
entity_type – Optional merged entity type filter
reference_filters – Optional reference field filters
skip – How many items to skip for pagination
limit – How many items to return at most
tx – Optional open transaction to run the query in

Returns:

Graph result instance

fetch_rule_items(query_string: str | None, identifier: str | None, entity_type: Sequence[__annotationlib_name_1__] | None, reference_filters: Sequence[ReferenceFilter] | None, skip: int, limit: int) → Result¶

Query the graph for rule items.

Parameters:

query_string – Optional full text search query term
identifier – Optional identifier filter
entity_type – Optional entity type filter
reference_filters – Optional reference field filters
skip – How many items to skip for pagination
limit – How many items to return at most

Returns:

Graph result instance

fetch_rule_set_response(stable_target_id: str, tx: Transaction | None = None) → Result¶

Query the graph for the rule set belonging to one merged item.

Parameters:

stable_target_id – Identifier of the merged item whose rule set to fetch
tx – Optional open transaction to run the query in

Returns:

Graph result instance with a single rule-set-response shaped record (one column per rule_set_field plus stableTargetId), or no records when the merged item has no rule nodes

flush() → None¶

Flush the database by deleting all nodes, constraints and indexes.

This operation only executes when debug mode is enabled in settings. Completely wipes the Neo4j database including all data, constraints, and indexes. Used for testing and development cleanup.

flush_data() → None¶

Flush the database by deleting all nodes, keeping the schema intact.

This operation only executes when debug mode is enabled in settings. Unlike flush, constraints and indexes are left in place, so callers that only need a clean dataset (e.g. tests) can avoid the cost of dropping and recreating the schema on every reset.

ingest_items(models: Iterable[__annotationlib_name_1__]) → Generator[None]¶: Ingest a list of extracted models or rule set responses into the graph.

merge_items(goner: MergedAccessPlatform | MergedActivity | MergedBibliographicResource | MergedConsent | MergedContactPoint | MergedDistribution | MergedLocation | MergedOrganization | MergedOrganizationalUnit | MergedPerson | MergedPrimarySource | MergedResource | MergedResourceSeries | MergedVariable | MergedVariableGroup, keeper: MergedAccessPlatform | MergedActivity | MergedBibliographicResource | MergedConsent | MergedContactPoint | MergedDistribution | MergedLocation | MergedOrganization | MergedOrganizationalUnit | MergedPerson | MergedPrimarySource | MergedResource | MergedResourceSeries | MergedVariable | MergedVariableGroup) → None¶: Merge a goner merged item into a keeper merged item.

static pick_anchor_filter(raw_reference_filters: list[RawReferenceFilter]) → RawReferenceFilter | None¶

Pick the reference filter that can drive the match from an index seek.

Matching every extracted and rule node and then checking each one’s references is O(graph). Starting from one referenced merged item instead is an index seek plus two expands, and any one of the ANDed filters will do, because an item that satisfies all of them satisfies this one too. A filter looking for the absence of a reference has nothing to seek, so it cannot serve as the anchor.

Parameters:: raw_reference_filters – The reference filters the query was asked for
Returns:: The filter to anchor on, or None to fall back to scanning

mex.backend.graph.connector.get_graph_status() → VersionStatus¶

Get the status and version of the graph database.

Returns:: VersionStatus with status “ok” and the neo4j version, status “offline” when the graph cannot be reached, or status “error” when the graph is misconfigured or answers with an error

mex.backend.graph.constants module¶

mex.backend.graph.exceptions module¶

exception mex.backend.graph.exceptions.DeletionFailedError(*args: Any, errors: Sequence[ErrorDetails] = (), retryable: bool = False)¶

Bases: GraphError

Exception raised when the the specified item could not be deleted.

exception mex.backend.graph.exceptions.GraphError(*args: Any, errors: Sequence[ErrorDetails] = (), retryable: bool = False)¶

Bases: BackendError

Graph error that can be constructed with more error metadata.

__init__(*args: Any, errors: Sequence[ErrorDetails] = (), retryable: bool = False) → None¶: Construct a new ingestion failure with underlying details.

errors() → list[ErrorDetails]¶: Details about underlying errors.

is_retryable() → bool¶: Whether the error is retryable.

exception mex.backend.graph.exceptions.IncompatibleVersionError¶

Bases: BackendError

Exception raised when the version of a connected service is too old.

exception mex.backend.graph.exceptions.InconsistentGraphError¶

Bases: BackendError

Exception raised for inconsistencies found in the graph database.

exception mex.backend.graph.exceptions.IngestionError(*args: Any, errors: Sequence[ErrorDetails] = (), retryable: bool = False)¶

Bases: GraphError

Error for ingestion failures with underlying details.

exception mex.backend.graph.exceptions.MergingError(*args: Any, errors: Sequence[ErrorDetails] = (), retryable: bool = False)¶

Bases: GraphError

Exception raised for errors during the merging process.

exception mex.backend.graph.exceptions.MultipleResultsFoundError¶

Bases: BackendError

A single database result was required but more than one were found.

exception mex.backend.graph.exceptions.NoResultFoundError¶

Bases: BackendError

A database result was required but none was found.

mex.backend.graph.models module¶

class mex.backend.graph.models.EdgeExporter¶

Bases: RecordExporter

Transformer class that turns edges into a string of format label {props}.

Full example:: shortName {position: 0}

transform(x: Any) → Any¶: Transform a value, or collection of values.

class mex.backend.graph.models.ExtractedPrimarySourceWithHardcodedIdentifiers(*, hadPrimarySource: MergedPrimarySourceIdentifier, identifierInPrimarySource: Annotated[str, MinLen(min_length=1), MaxLen(max_length=1000), _PydanticGeneralMetadata(pattern='^[^\\n\\r]+$')], version: Annotated[str, FieldInfo(annotation=NoneType, required=True, examples=['v1', '2023-01-16', 'Schema 9'], metadata=[MaxLen(max_length=100)])] | None = None, title: Annotated[list[Text], MinLen(min_length=1)], alternativeTitle: list[Text] = [], contact: list[Annotated[MergedOrganizationalUnitIdentifier | MergedPersonIdentifier | MergedContactPointIdentifier, AfterValidator(func=Identifier)]] = [], contributor: list[MergedPersonIdentifier] = [], description: list[Text] = [], documentation: list[Link] = [], locatedAt: list[Link] = [], unitInCharge: list[MergedOrganizationalUnitIdentifier] = [], entityType: Literal['ExtractedPrimarySource'] = 'ExtractedPrimarySource', identifier: ExtractedPrimarySourceIdentifier, stableTargetId: MergedPrimarySourceIdentifier)¶

Bases: BasePrimarySource, ExtractedData

Static metadata set describing a primary source.

An instance of this class will bypass the IdentityProvider. This way we can ensure the primary source receives static identifiers. We use this for the MEx maintained primary sources, e.g. for the mex and mex-editor primary source.

entityType: Annotated[Literal['ExtractedPrimarySource'], FieldInfo(annotation=NoneType, required=True, alias='$type', alias_priority=2, frozen=True)]¶

identifier: ExtractedPrimarySourceIdentifier¶

model_config = {'extra': 'forbid', 'json_schema_extra': {'closeMatch': ['http://www.w3.org/ns/dcat#Catalog', 'http://www.w3.org/ns/prov#PrimarySource'], 'description': 'A collection of information that is managed and curated by an RKI unit and lists activities and/or resources.'}, 'model_title_generator': <function <lambda>>, 'populate_by_name': True, 'str_max_length': 100000, 'str_min_length': 1, 'str_strip_whitespace': True, 'use_enum_values': True, 'validate_assignment': True, 'validate_default': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

stableTargetId: MergedPrimarySourceIdentifier¶

class mex.backend.graph.models.GraphRel¶

Bases: TypedDict

Type definition for graph relations.

edgeLabel: str¶

edgeProps: dict[str, str | int | float | None | bool | list[str | int | float | None | bool]]¶

nodeLabels: list[str]¶

nodeProps: dict[str, str | int | float | None | bool | list[str | int | float | None | bool]]¶

Bases: BaseModel

Type definition for ingestion data.

createRels: list[GraphRel]¶

entityType: str¶

identifier: str | None¶

linkRels: list[GraphRel]¶

metadata() → dict[str, int | str | None]¶: Return log-able metadata.

model_config = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

nodeProps: dict[str, str | int | float | None | bool | list[str | int | float | None | bool]]¶

classmethod sort_create_rels(v: list[GraphRel]) → list[GraphRel]¶: Sort the rels by edge label and position.

classmethod sort_link_rels(v: list[GraphRel]) → list[GraphRel]¶: Sort the rels by edge label and position.

stableTargetId: str¶

class mex.backend.graph.models.IngestParams(*, merged_label: str, node_label: str, used_referenced_labels: list[str], used_nested_labels: list[str], detach_node_edges: list[str], delete_node_edges: list[str], has_link_rels: bool, has_create_rels: bool)¶

Bases: BaseModel

Type definition for query parameters.

delete_node_edges: list[str]¶

detach_node_edges: list[str]¶

has_create_rels: bool¶

has_link_rels: bool¶

merged_label: str¶

model_config = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

node_label: str¶

used_nested_labels: list[str]¶

used_referenced_labels: list[str]¶

class mex.backend.graph.models.RawReferenceFilter¶

Bases: TypedDict

Reference filter in raw dictionary form to be used as cypher parameter.

field: str¶

identifiers: list[str]¶

class mex.backend.graph.models.Result(result: Result)¶

Bases: object

Represent a set of graph results.

This class wraps neo4j.Result in an interface akin to sqlalchemy.engine.Result. We do this, to reduce vendor tie-in with neo4j and limit the dependency-scope of the neo4j driver library to the mex.backend.graph submodule.

__init__(result: Result) → None¶: Wrap a neo4j result object in a mex-backend result.

all() → list[dict[str, Any]]¶: Return all records as a list.

get_update_counters() → dict[str, int]¶: Return a summary of counters for operations the query triggered.

log_notifications() → None¶: Log neo4j notifications.

one() → dict[str, Any]¶: Return exactly one record or raise an exception.

one_or_none() → dict[str, Any] | None¶

Return at most one result or raise an exception.

Returns None if the result has no records. Raises MultipleResultsFound if multiple records are returned.

mex.backend.graph.query module¶

class mex.backend.graph.query.Query(name: str, template: Template, kwargs: Mapping[__annotationlib_name_1__, __annotationlib_name_2__])¶

Bases: object

Wrapper for queries that can be rendered.

__init__(name: str, template: Template, kwargs: Mapping[__annotationlib_name_1__, __annotationlib_name_2__]) → None¶: Create a new query instance.

render() → str¶: Render the query for database execution.

class mex.backend.graph.query.QueryBuilder¶

Bases: BaseConnector

Wrapper around jinja template loading and rendering.

__init__() → None¶: Create a new jinja environment with template loader, filters and globals.

_get_ingest_query_for_entity_type(entity_type: str) → str¶

Create an ingest query for the given entity type.

Generates a complex Cypher query template for ingesting extracted or rule models into the graph database. The query handles creation of nodes, nested objects (Text, Link), and reference relationships. Results are cached for performance.

Parameters:: entity_type – The entity type name (e.g. “ExtractedPerson”, “AdditivePerson”)
Raises:: KeyError – If the entity type is not found in the model classes
Returns:: Cypher query string template for ingesting this entity type

close() → None¶: Clean up the connector.

mex.backend.graph.query.render_constraints(fields: list[~typing.Annotated[str, ~pydantic.types.StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=None, min_length=None, max_length=None, pattern=^[a-zA-Z]{1,255}$, ascii_only=None)]]) → str¶: Convert a list of field names into cypher node/edge constraints.

mex.backend.graph.transform module¶

class mex.backend.graph.transform._SearchResultReference¶

Bases: TypedDict

Helper class to show the structure of search result references.

label: str¶

position: int¶

value: str | dict[str, str | None]¶

mex.backend.graph.transform.expand_references_in_search_result(refs: list[_SearchResultReference]) → dict[str, list[str | dict[str, str | None]]]¶

Expand the _refs collection in a search result item.

Each item in a search result has a collection of _refs in the form of _SearchResultReference. Before parsing them into pydantic, we need to inline the references back into the item dictionary.

mex.backend.graph.transform.get_error_details_from_neo4j_error(input_: ExtractedAccessPlatform | ExtractedActivity | ExtractedBibliographicResource | ExtractedConsent | ExtractedContactPoint | ExtractedDistribution | ExtractedLocation | ExtractedOrganization | ExtractedOrganizationalUnit | ExtractedPerson | ExtractedPrimarySource | ExtractedResource | ExtractedResourceSeries | ExtractedVariable | ExtractedVariableGroup | AccessPlatformRuleSetResponse | ActivityRuleSetResponse | BibliographicResourceRuleSetResponse | ConsentRuleSetResponse | ContactPointRuleSetResponse | DistributionRuleSetResponse | LocationRuleSetResponse | OrganizationRuleSetResponse | OrganizationalUnitRuleSetResponse | PersonRuleSetResponse | PrimarySourceRuleSetResponse | ResourceRuleSetResponse | ResourceSeriesRuleSetResponse | VariableRuleSetResponse | VariableGroupRuleSetResponse | ExtractedPrimarySourceWithHardcodedIdentifiers | AnyPrimitiveType, error: Neo4jError) → list[ErrorDetails]¶: Convert an input context and a neo4j error into error details.

mex.backend.graph.transform.get_graph_rel_id(rel: GraphRel) → tuple[str, int]¶: Returns a string uniquely identifying the GraphRel.

Transform the given model into an ingestion instruction.

Converts an extracted or rule model into structured data ready for database ingestion. Handles field categorization (mutable vs final), reference field processing, and nested object preparation.

Parameters:

model – The extracted or rule model to transform for ingestion
stable_target_id – Identifier of the associated merged item

Returns:

IngestData object containing query parameters and metadata

mex.backend.graph.transform.transform_reference_filters_to_raw_fields(reference_filters: Sequence[ReferenceFilter] | None) → list[str]¶

Extract raw field names from reference filters.

Parameters:: reference_filters – Optional sequence of reference filters
Returns:: List of field names

mex.backend.graph.transform.transform_reference_filters_to_raw_filters(reference_filters: Sequence[ReferenceFilter] | None) → list[RawReferenceFilter]¶

Convert reference filters to raw dicts for cypher query parameters.

A None identifier is replaced with the no-reference sentinel so the cypher query can match items without that relationship.

Parameters:: reference_filters – Optional sequence of reference filters
Returns:: List of dictionaries containing field name and sanitized identifiers

mex.backend.graph.transform.validate_ingested_data(data_in: IngestData, data_out: IngestData) → list[ErrorDetails]¶: Validate that the ingestion has been executed successfully.

mex.backend.graph package¶

Submodules¶

mex.backend.graph.connector module¶

mex.backend.graph.constants module¶

mex.backend.graph.exceptions module¶

mex.backend.graph.models module¶

mex.backend.graph.query module¶

mex.backend.graph.transform module¶

Module contents¶

mex-backend

Navigation

Related Topics