1656 lines
47 KiB
Python
1656 lines
47 KiB
Python
# Copyright 2023 Memgraph Ltd.
|
||
#
|
||
# Use of this software is governed by the Business Source License
|
||
# included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
|
||
# License, and you may not use this file except in compliance with the Business Source License.
|
||
#
|
||
# As of the Change Date specified in that file, in accordance with
|
||
# the Business Source License, use of this software will be governed
|
||
# by the Apache License, Version 2.0, included in the file
|
||
# licenses/APL.txt.
|
||
|
||
"""
|
||
This module provides a mock Python API for easy development of custom openCypher procedures.
|
||
The API's interface is fully consistent with the Python API in mgp.py. Because of that, you
|
||
can develop procedures without setting Memgraph up for each run, and run them from Memgraph
|
||
when they're implemented in full.
|
||
"""
|
||
|
||
import datetime
|
||
import inspect
|
||
import sys
|
||
import typing
|
||
from collections import namedtuple
|
||
from copy import deepcopy
|
||
from functools import wraps
|
||
|
||
import _mgp_mock
|
||
|
||
|
||
class InvalidContextError(Exception):
|
||
"""
|
||
Signals using a graph element instance outside of the registered procedure.
|
||
"""
|
||
|
||
pass
|
||
|
||
|
||
class UnknownError(Exception):
|
||
"""
|
||
Signals unspecified failure.
|
||
"""
|
||
|
||
pass
|
||
|
||
|
||
class LogicErrorError(Exception):
|
||
"""
|
||
Signals faulty logic within the program such as violating logical
|
||
preconditions or class invariants and may be preventable.
|
||
"""
|
||
|
||
pass
|
||
|
||
|
||
class DeletedObjectError(Exception):
|
||
"""
|
||
Signals accessing an already deleted object.
|
||
"""
|
||
|
||
pass
|
||
|
||
|
||
class InvalidArgumentError(Exception):
|
||
"""
|
||
Signals that some of the arguments have invalid values.
|
||
"""
|
||
|
||
pass
|
||
|
||
|
||
class ImmutableObjectError(Exception):
|
||
"""
|
||
Signals modification of an immutable object.
|
||
"""
|
||
|
||
pass
|
||
|
||
|
||
class ValueConversionError(Exception):
|
||
"""
|
||
Signals that conversion between python and cypher values failed.
|
||
"""
|
||
|
||
pass
|
||
|
||
|
||
class Label:
|
||
"""A vertex label."""
|
||
|
||
__slots__ = ("_name",)
|
||
|
||
def __init__(self, name: str):
|
||
self._name = name
|
||
|
||
@property
|
||
def name(self) -> str:
|
||
"""
|
||
Get the name of the label.
|
||
|
||
Returns:
|
||
A string with the name of the label.
|
||
|
||
Example:
|
||
```label.name```
|
||
"""
|
||
return self._name
|
||
|
||
def __eq__(self, other) -> bool:
|
||
if isinstance(other, Label):
|
||
return self._name == other.name
|
||
|
||
if isinstance(other, str):
|
||
return self._name == other
|
||
|
||
return NotImplemented
|
||
|
||
|
||
# Named property value of a Vertex or an Edge.
|
||
# It would be better to use typing.NamedTuple with typed fields, but that is
|
||
# not available in Python 3.5.
|
||
Property = namedtuple("Property", ("name", "value"))
|
||
|
||
|
||
class Properties:
|
||
"""Collection of the properties of a vertex or an edge."""
|
||
|
||
__slots__ = ("_vertex_or_edge", "_len")
|
||
|
||
def __init__(self, vertex_or_edge):
|
||
if not isinstance(vertex_or_edge, (_mgp_mock.Vertex, _mgp_mock.Edge)):
|
||
raise TypeError(f"Expected _mgp_mock.Vertex or _mgp_mock.Edge, got {type(vertex_or_edge)}")
|
||
|
||
self._len = None
|
||
self._vertex_or_edge = vertex_or_edge
|
||
|
||
def __deepcopy__(self, memo):
|
||
# In line with the Python API, this is the same as the shallow copy.
|
||
return Properties(self._vertex_or_edge)
|
||
|
||
def get(self, property_name: str, default=None) -> object:
|
||
"""
|
||
Get the value of the property with the given name, otherwise return the default value.
|
||
|
||
Args:
|
||
property_name: String with the property name.
|
||
default: The value to return if there is no `property_name` property.
|
||
|
||
Returns:
|
||
The value associated with `property_name` or, if there’s no such property, the `default` argument.
|
||
|
||
Raises:
|
||
InvalidContextError: If the edge or vertex is out of context.
|
||
DeletedObjectError: If the edge has been deleted.
|
||
|
||
Examples:
|
||
```
|
||
vertex.properties.get(property_name)
|
||
edge.properties.get(property_name)
|
||
```
|
||
"""
|
||
if not self._vertex_or_edge.is_valid():
|
||
raise InvalidContextError()
|
||
|
||
try:
|
||
return self[property_name]
|
||
except KeyError:
|
||
return default
|
||
|
||
def set(self, property_name: str, value: object) -> None:
|
||
"""
|
||
Set the value of the given property. If `value` is `None`, the property is removed.
|
||
|
||
Args:
|
||
property_name: String with the property name.
|
||
value: The new value of the `property_name` property.
|
||
|
||
Raises:
|
||
ImmutableObjectError: If the object is immutable.
|
||
DeletedObjectError: If the edge has been deleted.
|
||
ValueConversionError: If `value` is vertex, edge or path.
|
||
|
||
Examples:
|
||
```
|
||
vertex.properties.set(property_name, value)
|
||
edge.properties.set(property_name, value)
|
||
```
|
||
"""
|
||
if not self._vertex_or_edge.underlying_graph_is_mutable():
|
||
raise ImmutableObjectError("Cannot modify immutable object.")
|
||
|
||
self[property_name] = value
|
||
|
||
def items(self) -> typing.Iterable[Property]:
|
||
"""
|
||
Iterate over the properties. Doesn’t return a dynamic view of the properties, but copies the
|
||
current properties.
|
||
|
||
Returns:
|
||
Iterable `Property` of names and values.
|
||
|
||
Raises:
|
||
InvalidContextError: If the edge or vertex is out of context.
|
||
DeletedObjectError: If the edge or vertex has been deleted.
|
||
|
||
Examples:
|
||
```
|
||
items = vertex.properties.items()
|
||
for it in items:
|
||
name = it.name
|
||
value = it.value
|
||
```
|
||
```
|
||
items = edge.properties.items()
|
||
for it in items:
|
||
name = it.name
|
||
value = it.value
|
||
```
|
||
"""
|
||
if not self._vertex_or_edge.is_valid():
|
||
raise InvalidContextError()
|
||
|
||
if self._vertex_or_edge.is_deleted():
|
||
raise DeletedObjectError("Accessing deleted object.")
|
||
|
||
vertex_or_edge_properties = self._vertex_or_edge.properties
|
||
|
||
for property in vertex_or_edge_properties:
|
||
yield Property(*property)
|
||
|
||
def keys(self) -> typing.Iterable[str]:
|
||
"""
|
||
Iterate over property names. Doesn’t return a dynamic view of the property names, but copies the
|
||
name of the current properties.
|
||
|
||
Returns:
|
||
`Iterable` of strings that represent the property names (keys).
|
||
|
||
Raises:
|
||
InvalidContextError: If edge or vertex is out of context.
|
||
DeletedObjectError: If the edge or vertex has been deleted.
|
||
|
||
Examples:
|
||
```
|
||
graph.vertex.properties.keys()
|
||
graph.edge.properties.keys()
|
||
```
|
||
"""
|
||
if not self._vertex_or_edge.is_valid():
|
||
raise InvalidContextError()
|
||
|
||
for item in self.items():
|
||
yield item.name
|
||
|
||
def values(self) -> typing.Iterable[object]:
|
||
"""
|
||
Iterate over property values. Doesn’t return a dynamic view of the property values, but copies the
|
||
values of the current properties.
|
||
|
||
Returns:
|
||
`Iterable` of property values.
|
||
|
||
Raises:
|
||
InvalidContextError: If edge or vertex is out of context.
|
||
DeletedObjectError: If the edge or vertex has been deleted.
|
||
|
||
Examples:
|
||
```
|
||
vertex.properties.values()
|
||
edge.properties.values()
|
||
```
|
||
"""
|
||
if not self._vertex_or_edge.is_valid():
|
||
raise InvalidContextError()
|
||
|
||
for item in self.items():
|
||
yield item.value
|
||
|
||
def __len__(self) -> int:
|
||
"""
|
||
Get the count of the vertex or edge’s properties.
|
||
|
||
Returns:
|
||
The count of the stored properties.
|
||
|
||
Raises:
|
||
InvalidContextError: If the edge or vertex is out of context.
|
||
DeletedObjectError: If the edge or vertex has been deleted.
|
||
|
||
Examples:
|
||
```
|
||
len(vertex.properties)
|
||
len(edge.properties)
|
||
```
|
||
"""
|
||
if not self._vertex_or_edge.is_valid():
|
||
raise InvalidContextError()
|
||
|
||
if self._len is None:
|
||
self._len = sum(1 for _ in self.items())
|
||
|
||
return self._len
|
||
|
||
def __iter__(self) -> typing.Iterable[str]:
|
||
"""
|
||
Iterate over property names.
|
||
|
||
Returns:
|
||
`Iterable` of strings that represent property names.
|
||
|
||
Raises:
|
||
InvalidContextError: If edge or vertex is out of context.
|
||
DeletedObjectError: If the edge or vertex has been deleted.
|
||
|
||
Examples:
|
||
```
|
||
iter(vertex.properties)
|
||
iter(edge.properties)
|
||
```
|
||
"""
|
||
if not self._vertex_or_edge.is_valid():
|
||
raise InvalidContextError()
|
||
|
||
for item in self.items():
|
||
yield item.name
|
||
|
||
def __getitem__(self, property_name: str) -> object:
|
||
"""
|
||
Get the value of the property with the given name, otherwise raise a KeyError.
|
||
|
||
Args:
|
||
property_name: String with the property name.
|
||
|
||
Returns:
|
||
Value of the named property.
|
||
|
||
Raises:
|
||
InvalidContextError: If edge or vertex is out of context.
|
||
DeletedObjectError: If the edge or vertex has been deleted.
|
||
|
||
Examples:
|
||
```
|
||
vertex.properties[property_name]
|
||
edge.properties[property_name]
|
||
```
|
||
"""
|
||
if not self._vertex_or_edge.is_valid():
|
||
raise InvalidContextError()
|
||
|
||
if self._vertex_or_edge.is_deleted():
|
||
raise DeletedObjectError("Accessing deleted object.")
|
||
|
||
return self._vertex_or_edge.get_property(property_name)
|
||
|
||
def __setitem__(self, property_name: str, value: object) -> None:
|
||
"""
|
||
Set the value of the given property. If `value` is `None`, the property
|
||
is removed.
|
||
|
||
Args:
|
||
property_name: String with the property name.
|
||
value: Object that represents the value to be set.
|
||
|
||
Raises:
|
||
InvalidContextError: If the edge or vertex is out of context.
|
||
ImmutableObjectError: If the object is immutable.
|
||
DeletedObjectError: If the edge or vertex has been deleted.
|
||
ValueConversionError: If `value` is vertex, edge or path.
|
||
|
||
Examples:
|
||
```
|
||
vertex.properties[property_name] = value
|
||
edge.properties[property_name] = value
|
||
```
|
||
"""
|
||
if not self._vertex_or_edge.is_valid():
|
||
raise InvalidContextError()
|
||
|
||
if not self._vertex_or_edge.underlying_graph_is_mutable():
|
||
raise ImmutableObjectError("Cannot modify immutable object.")
|
||
|
||
if isinstance(value, (Vertex, Edge, Path)):
|
||
raise ValueConversionError("Value conversion failed")
|
||
|
||
if self._vertex_or_edge.is_deleted():
|
||
raise DeletedObjectError("Accessing deleted object.")
|
||
|
||
self._vertex_or_edge.set_property(property_name, value)
|
||
|
||
def __contains__(self, property_name: str) -> bool:
|
||
"""
|
||
Check if there is a property with the given name.
|
||
|
||
Args:
|
||
property_name: String with the property name.
|
||
|
||
Returns:
|
||
Boolean value that represents whether a property with the given name exists.
|
||
|
||
Raises:
|
||
InvalidContextError: If edge or vertex is out of context.
|
||
DeletedObjectError: If the edge or vertex has been deleted.
|
||
|
||
Examples:
|
||
```
|
||
if property_name in vertex.properties:
|
||
```
|
||
```
|
||
if property_name in edge.properties:
|
||
```
|
||
"""
|
||
if not self._vertex_or_edge.is_valid():
|
||
raise InvalidContextError()
|
||
|
||
try:
|
||
_ = self[property_name]
|
||
return True
|
||
except KeyError:
|
||
return False
|
||
|
||
|
||
class EdgeType:
|
||
"""Type of an Edge."""
|
||
|
||
__slots__ = ("_name",)
|
||
|
||
def __init__(self, name):
|
||
self._name = name
|
||
|
||
@property
|
||
def name(self) -> str:
|
||
"""
|
||
Get the name of an EdgeType.
|
||
|
||
Returns:
|
||
The string with the name of the EdgeType.
|
||
|
||
Example:
|
||
```edge.type.name```
|
||
"""
|
||
return self._name
|
||
|
||
def __eq__(self, other) -> bool:
|
||
if isinstance(other, EdgeType):
|
||
return self.name == other.name
|
||
|
||
if isinstance(other, str):
|
||
return self.name == other
|
||
|
||
return NotImplemented
|
||
|
||
|
||
if sys.version_info >= (3, 5, 2):
|
||
EdgeId = typing.NewType("EdgeId", int)
|
||
else:
|
||
EdgeId = int
|
||
|
||
|
||
class Edge:
|
||
"""Represents a graph edge.
|
||
|
||
Access to an Edge is only valid during a single execution of a procedure in
|
||
a query. You should not globally store an instance of an Edge. Using an
|
||
invalid Edge instance will raise an InvalidContextError.
|
||
"""
|
||
|
||
__slots__ = ("_edge",)
|
||
|
||
def __init__(self, edge):
|
||
if not isinstance(edge, _mgp_mock.Edge):
|
||
raise TypeError(f"Expected '_mgp_mock.Edge', got '{type(edge)}'")
|
||
|
||
self._edge = edge
|
||
|
||
def __deepcopy__(self, memo):
|
||
# In line with the Python API, this is the same as the shallow copy.
|
||
return Edge(self._edge)
|
||
|
||
def is_valid(self) -> bool:
|
||
"""
|
||
Check if the `Edge` is in a valid context, i.e. if it may be used.
|
||
|
||
Returns:
|
||
A `bool` value that represents whether the edge is in a valid context.
|
||
|
||
Examples:
|
||
```edge.is_valid()```
|
||
"""
|
||
return self._edge.is_valid()
|
||
|
||
def underlying_graph_is_mutable(self) -> bool:
|
||
"""
|
||
Check if the underlying `Graph` is mutable.
|
||
|
||
Returns:
|
||
A `bool` value that represents whether the graph is mutable.
|
||
|
||
Raises:
|
||
InvalidContextError: If the context is not valid.
|
||
|
||
Examples:
|
||
```edge.underlying_graph_is_mutable()```
|
||
"""
|
||
if not self.is_valid():
|
||
raise InvalidContextError()
|
||
|
||
return self._edge.underlying_graph_is_mutable()
|
||
|
||
@property
|
||
def id(self) -> EdgeId:
|
||
"""
|
||
Get the ID of the edge.
|
||
|
||
Returns:
|
||
An `EdgeId` representing the edge’s ID.
|
||
|
||
Raises:
|
||
InvalidContextError: If edge is out of context.
|
||
|
||
Examples:
|
||
```edge.id```
|
||
"""
|
||
if not self.is_valid():
|
||
raise InvalidContextError()
|
||
|
||
return self._edge.id
|
||
|
||
@property
|
||
def type(self) -> EdgeType:
|
||
"""
|
||
Get the type of the `Edge`.
|
||
|
||
Returns:
|
||
`EdgeType` representing the edge’s type.
|
||
|
||
Raises:
|
||
InvalidContextError: If edge is out of context.
|
||
|
||
Examples:
|
||
```edge.type```
|
||
"""
|
||
if not self.is_valid():
|
||
raise InvalidContextError()
|
||
|
||
return EdgeType(self._edge.get_type_name())
|
||
|
||
@property
|
||
def from_vertex(self) -> "Vertex":
|
||
"""
|
||
Get the source (tail) vertex of the edge.
|
||
|
||
Returns:
|
||
`Vertex` that is the edge’s source/tail.
|
||
|
||
Raises:
|
||
InvalidContextError: If edge is out of context.
|
||
|
||
Examples:
|
||
```edge.from_vertex```
|
||
"""
|
||
if not self.is_valid():
|
||
raise InvalidContextError()
|
||
|
||
return Vertex(self._edge.from_vertex())
|
||
|
||
@property
|
||
def to_vertex(self) -> "Vertex":
|
||
"""
|
||
Get the destination (head) vertex of the edge.
|
||
|
||
Returns:
|
||
`Vertex` that is the edge’s destination/head.
|
||
|
||
Raises:
|
||
InvalidContextError: If edge is out of context.
|
||
|
||
Examples:
|
||
```edge.to_vertex```
|
||
"""
|
||
if not self.is_valid():
|
||
raise InvalidContextError()
|
||
|
||
return Vertex(self._edge.to_vertex())
|
||
|
||
@property
|
||
def properties(self) -> Properties:
|
||
"""
|
||
Get the edge’s properties.
|
||
|
||
Returns:
|
||
`Properties` containing all properties of the edge.
|
||
|
||
Raises:
|
||
InvalidContextError: If edge is out of context.
|
||
|
||
Examples:
|
||
```edge.properties```
|
||
"""
|
||
if not self.is_valid():
|
||
raise InvalidContextError()
|
||
|
||
return Properties(self._edge)
|
||
|
||
def __eq__(self, other) -> bool:
|
||
if not self.is_valid():
|
||
raise InvalidContextError()
|
||
|
||
if not isinstance(other, Edge):
|
||
return NotImplemented
|
||
|
||
return self.id == other.id
|
||
|
||
def __hash__(self) -> int:
|
||
return hash(self.id)
|
||
|
||
|
||
if sys.version_info >= (3, 5, 2):
|
||
VertexId = typing.NewType("VertexId", int)
|
||
else:
|
||
VertexId = int
|
||
|
||
|
||
class Vertex:
|
||
"""Represents a graph vertex.
|
||
|
||
Access to a Vertex is only valid during a single execution of a procedure
|
||
in a query. You should not globally store an instance of a Vertex. Using an
|
||
invalid Vertex instance will raise an InvalidContextError.
|
||
"""
|
||
|
||
__slots__ = ("_vertex",)
|
||
|
||
def __init__(self, vertex):
|
||
if not isinstance(vertex, _mgp_mock.Vertex):
|
||
raise TypeError(f"Expected '_mgp_mock.Vertex', got '{type(vertex)}'")
|
||
|
||
self._vertex = vertex
|
||
|
||
def __deepcopy__(self, memo):
|
||
# In line with the Python API, this is the same as the shallow copy.
|
||
return Vertex(self._vertex)
|
||
|
||
def is_valid(self) -> bool:
|
||
"""
|
||
Check if the vertex is in a valid context, i.e. if it may be used.
|
||
|
||
Returns:
|
||
A `bool` value that represents whether the vertex is in a valid context.
|
||
|
||
Examples:
|
||
```vertex.is_valid()```
|
||
"""
|
||
return self._vertex.is_valid()
|
||
|
||
def underlying_graph_is_mutable(self) -> bool:
|
||
"""
|
||
Check if the underlying graph is mutable.
|
||
|
||
Returns:
|
||
A `bool` value that represents whether the graph is mutable.
|
||
|
||
Raises:
|
||
InvalidContextError: If the context is not valid.
|
||
|
||
Examples:
|
||
```edge.underlying_graph_is_mutable()```
|
||
"""
|
||
if not self.is_valid():
|
||
raise InvalidContextError()
|
||
|
||
return self._vertex.underlying_graph_is_mutable()
|
||
|
||
@property
|
||
def id(self) -> VertexId:
|
||
"""
|
||
Get the ID of the vertex.
|
||
|
||
Returns:
|
||
A `VertexId` representing the vertex’s ID.
|
||
|
||
Raises:
|
||
InvalidContextError: If vertex is out of context.
|
||
|
||
Examples:
|
||
```vertex.id```
|
||
"""
|
||
if not self.is_valid():
|
||
raise InvalidContextError()
|
||
|
||
return self._vertex.id
|
||
|
||
@property
|
||
def labels(self) -> typing.Tuple[Label]:
|
||
"""
|
||
Get the labels of the vertex.
|
||
|
||
Returns:
|
||
A tuple of `Label` instances representing individual labels.
|
||
|
||
Raises:
|
||
InvalidContextError: If vertex is out of context.
|
||
DeletedObjectError: If `Vertex` has been deleted.
|
||
|
||
Examples:
|
||
```vertex.labels```
|
||
"""
|
||
if not self.is_valid():
|
||
raise InvalidContextError()
|
||
|
||
if self._vertex.is_deleted():
|
||
raise DeletedObjectError("Accessing deleted object.")
|
||
|
||
return tuple(Label(label) for label in self._vertex.labels)
|
||
|
||
def add_label(self, label: str) -> None:
|
||
"""
|
||
Add the given label to the vertex.
|
||
|
||
Args:
|
||
label: The label (`str`) to be added.
|
||
|
||
Raises:
|
||
InvalidContextError: If `Vertex` is out of context.
|
||
ImmutableObjectError: If `Vertex` is immutable.
|
||
DeletedObjectError: If `Vertex` has been deleted.
|
||
|
||
Examples:
|
||
```vertex.add_label(label)```
|
||
"""
|
||
if not self.is_valid():
|
||
raise InvalidContextError()
|
||
|
||
if self._vertex.is_deleted():
|
||
raise DeletedObjectError("Accessing deleted object.")
|
||
|
||
return self._vertex.add_label(label)
|
||
|
||
def remove_label(self, label: str) -> None:
|
||
"""
|
||
Remove the given label from the vertex.
|
||
|
||
Args:
|
||
label: The label (`str`) to be removed.
|
||
|
||
Raises:
|
||
InvalidContextError: If `Vertex` is out of context.
|
||
ImmutableObjectError: If `Vertex` is immutable.
|
||
DeletedObjectError: If `Vertex` has been deleted.
|
||
|
||
Examples:
|
||
```vertex.remove_label(label)```
|
||
"""
|
||
if not self.is_valid():
|
||
raise InvalidContextError()
|
||
|
||
if self._vertex.is_deleted():
|
||
raise DeletedObjectError("Accessing deleted object.")
|
||
|
||
return self._vertex.remove_label(label)
|
||
|
||
@property
|
||
def properties(self) -> Properties:
|
||
"""
|
||
Get the properties of the vertex.
|
||
|
||
Returns:
|
||
The `Properties` of the vertex.
|
||
|
||
Raises:
|
||
InvalidContextError: If `Vertex` is out of context.
|
||
|
||
Examples:
|
||
```vertex.properties```
|
||
"""
|
||
if not self.is_valid():
|
||
raise InvalidContextError()
|
||
|
||
return Properties(self._vertex)
|
||
|
||
@property
|
||
def in_edges(self) -> typing.Iterable[Edge]:
|
||
"""
|
||
Iterate over the inbound edges of the vertex.
|
||
Doesn’t return a dynamic view of the edges, but copies the current inbound edges.
|
||
|
||
Returns:
|
||
An `Iterable` of all `Edge` objects directed towards the vertex.
|
||
|
||
Raises:
|
||
InvalidContextError: If `Vertex` is out of context.
|
||
DeletedObjectError: If `Vertex` has been deleted.
|
||
|
||
Examples:
|
||
```for edge in vertex.in_edges:```
|
||
"""
|
||
if not self.is_valid():
|
||
raise InvalidContextError()
|
||
|
||
if self._vertex.is_deleted():
|
||
raise DeletedObjectError("Accessing deleted object.")
|
||
|
||
for edge in self._vertex.in_edges:
|
||
yield Edge(edge)
|
||
|
||
@property
|
||
def out_edges(self) -> typing.Iterable[Edge]:
|
||
"""
|
||
Iterate over the outbound edges of the vertex.
|
||
Doesn’t return a dynamic view of the edges, but copies the current outbound edges.
|
||
|
||
Returns:
|
||
An `Iterable` of all `Edge` objects directed outwards from the vertex.
|
||
|
||
Raises:
|
||
InvalidContextError: If `Vertex` is out of context.
|
||
DeletedObjectError: If `Vertex` has been deleted.
|
||
|
||
Examples:
|
||
```for edge in vertex.in_edges:```
|
||
"""
|
||
if not self.is_valid():
|
||
raise InvalidContextError()
|
||
|
||
if self._vertex.is_deleted():
|
||
raise DeletedObjectError("Accessing deleted object.")
|
||
|
||
for edge in self._vertex.out_edges:
|
||
yield Edge(edge)
|
||
|
||
def __eq__(self, other) -> bool:
|
||
if not self.is_valid():
|
||
raise InvalidContextError()
|
||
|
||
if not isinstance(other, Vertex):
|
||
return NotImplemented
|
||
|
||
return self.id == other.id
|
||
|
||
def __hash__(self) -> int:
|
||
return hash(self.id)
|
||
|
||
|
||
class Path:
|
||
"""Represents a path comprised of `Vertex` and `Edge` instances."""
|
||
|
||
__slots__ = ("_path", "_vertices", "_edges")
|
||
|
||
def __init__(self, starting_vertex_or_path: typing.Union[_mgp_mock.Path, Vertex]):
|
||
"""Initialize with a starting `Vertex`.
|
||
|
||
Raises:
|
||
InvalidContextError: If the given vertex is invalid.
|
||
"""
|
||
# Accepting _mgp.Path is just for internal usage.
|
||
if isinstance(starting_vertex_or_path, _mgp_mock.Path):
|
||
self._path = starting_vertex_or_path
|
||
|
||
elif isinstance(starting_vertex_or_path, Vertex):
|
||
# For consistency with the Python API, the `_vertex` attribute isn’t a “public” property.
|
||
vertex = starting_vertex_or_path._vertex
|
||
if not vertex.is_valid():
|
||
raise InvalidContextError()
|
||
|
||
self._path = _mgp_mock.Path.make_with_start(vertex)
|
||
|
||
else:
|
||
raise TypeError(f"Expected 'Vertex' or '_mgp_mock.Path', got '{type(starting_vertex_or_path)}'")
|
||
|
||
self._vertices = None
|
||
self._edges = None
|
||
|
||
def __copy__(self):
|
||
if not self.is_valid():
|
||
raise InvalidContextError()
|
||
|
||
assert len(self.vertices) >= 1
|
||
|
||
path = Path(self.vertices[0])
|
||
for e in self.edges:
|
||
path.expand(e)
|
||
|
||
return path
|
||
|
||
def __deepcopy__(self, memo):
|
||
# In line with the Python API, this is the same as the shallow copy.
|
||
try:
|
||
return Path(memo[id(self._path)])
|
||
except KeyError:
|
||
pass
|
||
path = self.__copy__()
|
||
memo[id(self._path)] = path._path
|
||
return path
|
||
|
||
def is_valid(self) -> bool:
|
||
"""
|
||
Check if the path is in a valid context, i.e. if it may be used.
|
||
|
||
Returns:
|
||
A `bool` value that represents whether the path is in a valid context.
|
||
|
||
Examples:
|
||
```path.is_valid()```
|
||
"""
|
||
return self._path.is_valid()
|
||
|
||
def expand(self, edge: Edge):
|
||
"""
|
||
Append an edge continuing from the last vertex on the path.
|
||
The destination (head) of the given edge will become the last vertex in the path.
|
||
|
||
Args:
|
||
edge: The `Edge` to be added to the path.
|
||
|
||
Raises:
|
||
InvalidContextError: If using an invalid `Path` instance or if the given `Edge` is invalid.
|
||
LogicErrorError: If the current last vertex in the path is not part of the given edge.
|
||
|
||
Examples:
|
||
```path.expand(edge)```
|
||
"""
|
||
if not isinstance(edge, Edge):
|
||
raise TypeError(f"Expected 'Edge', got '{type(edge)}'")
|
||
|
||
if not self.is_valid() or not edge.is_valid():
|
||
raise InvalidContextError()
|
||
|
||
# For consistency with the Python API, the `_edge` attribute isn’t a “public” property.
|
||
self._path.expand(edge._edge)
|
||
|
||
self._vertices = None
|
||
self._edges = None
|
||
|
||
@property
|
||
def vertices(self) -> typing.Tuple[Vertex, ...]:
|
||
"""
|
||
Get the path’s vertices in a fixed order.
|
||
|
||
Returns:
|
||
A `Tuple` of the path’s vertices (`Vertex`) ordered from the start to the end of the path.
|
||
|
||
Raises:
|
||
InvalidContextError: If using an invalid Path instance.
|
||
|
||
Examples:
|
||
```path.vertices```
|
||
"""
|
||
if not self.is_valid():
|
||
raise InvalidContextError()
|
||
|
||
if self._vertices is None:
|
||
num_vertices = self._path.size() + 1
|
||
self._vertices = tuple(Vertex(self._path.vertex_at(i)) for i in range(num_vertices))
|
||
|
||
return self._vertices
|
||
|
||
@property
|
||
def edges(self) -> typing.Tuple[Edge, ...]:
|
||
"""
|
||
Get the path’s edges in a fixed order.
|
||
|
||
Returns:
|
||
A `Tuple` of the path’s edges (`Edges`) ordered from the start to the end of the path.
|
||
|
||
Raises:
|
||
InvalidContextError: If using an invalid Path instance.
|
||
|
||
Examples:
|
||
```path.vertices```
|
||
"""
|
||
if not self.is_valid():
|
||
raise InvalidContextError()
|
||
|
||
if self._edges is None:
|
||
num_edges = self._path.size()
|
||
self._edges = tuple(Edge(self._path.edge_at(i)) for i in range(num_edges))
|
||
|
||
return self._edges
|
||
|
||
|
||
class Vertices:
|
||
"""An iterable structure of the vertices in a graph."""
|
||
|
||
__slots__ = ("_graph", "_len")
|
||
|
||
def __init__(self, graph):
|
||
if not isinstance(graph, _mgp_mock.Graph):
|
||
raise TypeError(f"Expected '_mgp_mock.Graph', got '{type(graph)}'")
|
||
|
||
self._graph = graph
|
||
self._len = None
|
||
|
||
def __deepcopy__(self, memo):
|
||
# In line with the Python API, this is the same as the shallow copy.
|
||
return Vertices(self._graph)
|
||
|
||
def is_valid(self) -> bool:
|
||
"""
|
||
Check if the `Vertices` object is in a valid context, i.e. if it may be used.
|
||
|
||
Returns:
|
||
A `bool` value that represents whether the object is in a valid context.
|
||
|
||
Examples:
|
||
```vertices.is_valid()```
|
||
"""
|
||
return self._graph.is_valid()
|
||
|
||
def __iter__(self) -> typing.Iterable[Vertex]:
|
||
"""
|
||
Iterate over a graph’s vertices.
|
||
|
||
Returns:
|
||
An `Iterable` of `Vertex` objects.
|
||
|
||
Raises:
|
||
InvalidContextError: If context is invalid.
|
||
|
||
Examples:
|
||
```
|
||
for vertex in graph.vertices:
|
||
```
|
||
```
|
||
iter(graph.vertices)
|
||
```
|
||
"""
|
||
if not self.is_valid():
|
||
raise InvalidContextError()
|
||
|
||
for vertex in self._graph.vertices:
|
||
yield Vertex(vertex)
|
||
|
||
def __contains__(self, vertex: Vertex):
|
||
"""
|
||
Check if the given vertex is one of the graph vertices.
|
||
|
||
Args:
|
||
vertex: The `Vertex` to be checked.
|
||
|
||
Returns:
|
||
A Boolean value that represents whether the given vertex is one of the graph vertices.
|
||
|
||
Raises:
|
||
InvalidContextError: If the `Vertices` instance or the givern vertex is not in a valid context.
|
||
|
||
Examples:
|
||
```if vertex in graph.vertices:```
|
||
"""
|
||
if not self.is_valid() or not vertex.is_valid():
|
||
raise InvalidContextError()
|
||
|
||
return self._graph.has_node(vertex.id)
|
||
|
||
def __len__(self):
|
||
"""
|
||
Get the count of the graph vertices.
|
||
|
||
Returns:
|
||
The count of the vertices in the graph.
|
||
|
||
Raises:
|
||
InvalidContextError: If the `Vertices` instance is not in a valid context.
|
||
|
||
Examples:
|
||
```len(graph.vertices)```
|
||
"""
|
||
if not self.is_valid():
|
||
raise InvalidContextError()
|
||
|
||
if not self._len:
|
||
self._len = sum(1 for _ in self)
|
||
|
||
return self._len
|
||
|
||
|
||
class Record:
|
||
"""Represents a record as returned by a query procedure.
|
||
Records are comprised of key (field name) - value pairs."""
|
||
|
||
__slots__ = ("fields",)
|
||
|
||
def __init__(self, **kwargs):
|
||
"""Initialize with {name}={value} fields in kwargs."""
|
||
self.fields = kwargs
|
||
|
||
def __str__(self):
|
||
return str(self.fields)
|
||
|
||
|
||
class Graph:
|
||
"""The graph that stands in for Memgraph’s graph."""
|
||
|
||
__slots__ = ("_graph",)
|
||
|
||
def __init__(self, graph):
|
||
if not isinstance(graph, _mgp_mock.Graph):
|
||
raise TypeError(f"Expected '_mgp_mock.Graph', got '{type(graph)}'")
|
||
|
||
self._graph = graph
|
||
|
||
def __deepcopy__(self, memo):
|
||
# In line with the Python API, this is the same as the shallow copy.
|
||
return Graph(self._graph)
|
||
|
||
def is_valid(self) -> bool:
|
||
"""
|
||
Check if the graph is in a valid context, i.e. if it may be used.
|
||
|
||
Returns:
|
||
A `bool` value that represents whether the graph is in a valid context.
|
||
|
||
Examples:
|
||
```graph.is_valid()```
|
||
"""
|
||
return self._graph.is_valid()
|
||
|
||
def get_vertex_by_id(self, vertex_id: VertexId) -> Vertex:
|
||
"""
|
||
Return the graph vertex with the given vertex_id.
|
||
Access to a `Vertex` is only valid during a single execution of a
|
||
procedure in a query. You should not globally store the returned
|
||
vertex.
|
||
|
||
Args:
|
||
vertex_id: A Memgraph vertex ID (`Vertex ID`)
|
||
|
||
Returns:
|
||
The `Vertex` with the given ID.
|
||
|
||
Raises:
|
||
IndexError: If unable to find the given vertex_id.
|
||
InvalidContextError: If context is invalid.
|
||
|
||
Examples:
|
||
```graph.get_vertex_by_id(vertex_id)```
|
||
"""
|
||
if not self.is_valid():
|
||
raise InvalidContextError()
|
||
|
||
return Vertex(self._graph.get_vertex_by_id(vertex_id))
|
||
|
||
@property
|
||
def vertices(self) -> Vertices:
|
||
"""
|
||
Get all graph vertices.
|
||
|
||
Access to a `Vertex` is only valid during a single execution of a
|
||
query procedure. You should not globally store the returned `Vertex`
|
||
instances.
|
||
|
||
Returns:
|
||
`Vertices` in the graph.
|
||
|
||
Raises:
|
||
InvalidContextError: If context is invalid.
|
||
|
||
Examples:
|
||
Iteration over all graph vertices.
|
||
|
||
```
|
||
graph = context.graph
|
||
for vertex in graph.vertices:
|
||
```
|
||
"""
|
||
if not self.is_valid():
|
||
raise InvalidContextError()
|
||
|
||
return Vertices(self._graph)
|
||
|
||
def is_mutable(self) -> bool:
|
||
"""
|
||
Check if the graph is mutable, i.e. if it can be modified.
|
||
|
||
Returns:
|
||
A `bool` value that represents whether the graph is mutable.
|
||
|
||
Raises:
|
||
InvalidContextError: If the graph is not in a valid context.
|
||
|
||
Examples:
|
||
```graph.is_mutable()```
|
||
"""
|
||
if not self.is_valid():
|
||
raise InvalidContextError()
|
||
|
||
return not self._graph.is_immutable()
|
||
|
||
def create_vertex(self) -> Vertex:
|
||
"""
|
||
Create an empty vertex.
|
||
|
||
Returns:
|
||
The created `Vertex`.
|
||
|
||
Raises:
|
||
InvalidContextError: If the graph is not in a valid context.
|
||
ImmutableObjectError: If the graph is immutable.
|
||
|
||
Examples:
|
||
Creating an empty vertex:
|
||
```vertex = graph.create_vertex()```
|
||
"""
|
||
if not self.is_valid():
|
||
raise InvalidContextError()
|
||
|
||
if self._graph.is_immutable():
|
||
raise ImmutableObjectError("Cannot modify immutable object.")
|
||
|
||
return Vertex(self._graph.create_vertex())
|
||
|
||
def delete_vertex(self, vertex: Vertex) -> None:
|
||
"""
|
||
Delete the given vertex if it’s isolated, i.e. if there are no edges connected to it.
|
||
|
||
Args:
|
||
vertex: The `Vertex` to be deleted.
|
||
|
||
Raises:
|
||
InvalidContextError: If the graph is not in a valid context.
|
||
ImmutableObjectError: If the graph is immutable.
|
||
LogicErrorError: If the vertex is not isolated.
|
||
|
||
Examples:
|
||
```graph.delete_vertex(vertex)```
|
||
"""
|
||
if not self.is_valid():
|
||
raise InvalidContextError()
|
||
|
||
if self._graph.is_immutable():
|
||
raise ImmutableObjectError("Cannot modify immutable object.")
|
||
|
||
if not self._graph.vertex_is_isolate(vertex.id):
|
||
raise LogicErrorError("Logic error.")
|
||
|
||
self._graph.delete_vertex(vertex.id)
|
||
|
||
def detach_delete_vertex(self, vertex: Vertex) -> None:
|
||
"""
|
||
Delete the given vertex together with all connected edges.
|
||
|
||
Args:
|
||
vertex: The `Vertex` to be deleted.
|
||
|
||
Raises:
|
||
InvalidContextError: If the graph is not in a valid context.
|
||
ImmutableObjectError: If the graph is immutable.
|
||
|
||
Examples:
|
||
```graph.detach_delete_vertex(vertex)```
|
||
"""
|
||
if not self.is_valid():
|
||
raise InvalidContextError()
|
||
|
||
if self._graph.is_immutable():
|
||
raise ImmutableObjectError("Cannot modify immutable object.")
|
||
|
||
self._graph.delete_vertex(vertex.id)
|
||
|
||
def create_edge(self, from_vertex: Vertex, to_vertex: Vertex, edge_type: EdgeType) -> Edge:
|
||
"""
|
||
Create an empty edge.
|
||
|
||
Args:
|
||
from_vertex: The source (tail) `Vertex`.
|
||
to_vertex: The destination (head) `Vertex`.
|
||
edge_type: `EdgeType` specifying the new edge’s type.
|
||
|
||
Returns:
|
||
The created `Edge`.
|
||
|
||
Raises:
|
||
InvalidContextError: If the graph is not in a valid context.
|
||
ImmutableObjectError: If the graph is immutable.
|
||
DeletedObjectError: If `from_vertex` or `to_vertex` have been deleted.
|
||
|
||
Examples:
|
||
```edge = graph.create_edge(from_vertex, vertex, edge_type)```
|
||
"""
|
||
|
||
if not self.is_valid():
|
||
raise InvalidContextError()
|
||
|
||
if self._graph.is_immutable():
|
||
raise ImmutableObjectError("Cannot modify immutable object.")
|
||
|
||
new_edge = self._graph.create_edge(from_vertex._vertex, to_vertex._vertex, edge_type.name)
|
||
return Edge(new_edge)
|
||
|
||
def delete_edge(self, edge: Edge) -> None:
|
||
"""
|
||
Delete the given edge.
|
||
|
||
Args:
|
||
edge: The `Edge` to be deleted.
|
||
|
||
Raises:
|
||
InvalidContextError: If the graph is not in a valid context.
|
||
ImmutableObjectError: If the graph is immutable.
|
||
|
||
Examples:
|
||
```graph.delete_edge(edge)```
|
||
"""
|
||
if not self.is_valid():
|
||
raise InvalidContextError()
|
||
|
||
if self._graph.is_immutable():
|
||
raise ImmutableObjectError("Cannot modify immutable object.")
|
||
|
||
self._graph.delete_edge(edge.from_vertex.id, edge.to_vertex.id, edge.id)
|
||
|
||
|
||
class AbortError(Exception):
|
||
"""Signals that the procedure was asked to abort its execution."""
|
||
|
||
pass
|
||
|
||
|
||
class ProcCtx:
|
||
"""The context of the procedure being executed.
|
||
|
||
Access to a `ProcCtx` is only valid during a single execution of a query procedure.
|
||
You should not globally store a `ProcCtx` instance.
|
||
"""
|
||
|
||
__slots__ = ("_graph",)
|
||
|
||
def __init__(self, graph):
|
||
if not isinstance(graph, (_mgp_mock.Graph, _mgp_mock.nx.MultiDiGraph)):
|
||
raise TypeError(f"Expected '_mgp_mock.Graph' or 'networkx.MultiDiGraph', got '{type(graph)}'")
|
||
|
||
self._graph = Graph(graph) if isinstance(graph, _mgp_mock.Graph) else Graph(_mgp_mock.Graph(graph))
|
||
|
||
def is_valid(self) -> bool:
|
||
"""
|
||
Check if the context is valid, i.e. if the contained structures may be used.
|
||
|
||
Returns:
|
||
A `bool` value that represents whether the context is valid.
|
||
|
||
Examples:
|
||
```context.is_valid()```
|
||
"""
|
||
return self._graph.is_valid()
|
||
|
||
@property
|
||
def graph(self) -> Graph:
|
||
"""
|
||
Access the graph.
|
||
|
||
Returns:
|
||
A `Graph` object representing the graph.
|
||
|
||
Raises:
|
||
InvalidContextError: If the procedure context is not valid.
|
||
|
||
Examples:
|
||
```context.graph```
|
||
"""
|
||
if not self.is_valid():
|
||
raise InvalidContextError()
|
||
|
||
return self._graph
|
||
|
||
|
||
# Additional typing support
|
||
|
||
Number = typing.Union[int, float]
|
||
|
||
Map = typing.Union[dict, Edge, Vertex]
|
||
|
||
Date = datetime.date
|
||
|
||
LocalTime = datetime.time
|
||
|
||
LocalDateTime = datetime.datetime
|
||
|
||
Duration = datetime.timedelta
|
||
|
||
Any = typing.Union[bool, str, Number, Map, list, Date, LocalTime, LocalDateTime, Duration]
|
||
|
||
List = typing.List
|
||
|
||
Nullable = typing.Optional
|
||
|
||
|
||
# Procedure registration
|
||
|
||
|
||
def raise_if_does_not_meet_requirements(func: typing.Callable[..., Record]):
|
||
if not callable(func):
|
||
raise TypeError(f"Expected a callable object, got an instance of '{type(func)}'")
|
||
if inspect.iscoroutinefunction(func):
|
||
raise TypeError("Callable must not be 'async def' function")
|
||
if sys.version_info >= (3, 6):
|
||
if inspect.isasyncgenfunction(func):
|
||
raise TypeError("Callable must not be 'async def' function")
|
||
if inspect.isgeneratorfunction(func):
|
||
raise NotImplementedError("Generator functions are not supported")
|
||
|
||
|
||
def _register_proc(func: typing.Callable[..., Record], is_write: bool):
|
||
raise_if_does_not_meet_requirements(func)
|
||
|
||
sig = inspect.signature(func)
|
||
|
||
params = tuple(sig.parameters.values())
|
||
if params and params[0].annotation is ProcCtx:
|
||
|
||
@wraps(func)
|
||
def wrapper(ctx, *args):
|
||
result_record = None
|
||
|
||
if is_write:
|
||
ctx_copy = ProcCtx(deepcopy(ctx._graph._graph.nx))
|
||
|
||
result_record = func(ctx_copy, *args)
|
||
|
||
ctx._graph._graph = deepcopy(ctx_copy._graph._graph)
|
||
|
||
# Invalidate context after execution
|
||
ctx_copy._graph._graph.invalidate()
|
||
else:
|
||
ctx._graph._graph.make_immutable()
|
||
|
||
result_record = func(ctx, *args)
|
||
|
||
# Invalidate context after execution
|
||
ctx._graph._graph.invalidate()
|
||
|
||
return result_record
|
||
|
||
else:
|
||
|
||
@wraps(func)
|
||
def wrapper(*args):
|
||
return func(*args)
|
||
|
||
if sig.return_annotation is not sig.empty:
|
||
record = sig.return_annotation
|
||
|
||
if not isinstance(record, Record):
|
||
raise TypeError(f"Expected '{func.__name__}' to return 'mgp.Record', got '{type(record)}'")
|
||
|
||
return wrapper
|
||
|
||
|
||
def read_proc(func: typing.Callable[..., Record]):
|
||
"""
|
||
Register a function as a Memgraph read-only procedure.
|
||
|
||
The `func` needs to be a callable and optionally take `ProcCtx` as its first argument.
|
||
Other parameters of `func` will be bound to the passed arguments.
|
||
The full signature of `func` needs to be annotated with types. The return type must
|
||
be `Record(field_name=type, ...)`, and the procedure must produce either a complete
|
||
Record or None. Multiple records can be produced by returning an iterable of them.
|
||
Registering generator functions is currently not supported.
|
||
|
||
Example:
|
||
```
|
||
import mgp_mock
|
||
|
||
@mgp_mock.read_proc
|
||
def procedure(context: mgp_mock.ProcCtx,
|
||
required_arg: mgp_mock.Nullable[mgp_mock.Any],
|
||
optional_arg: mgp_mock.Nullable[mgp_mock.Any] = None
|
||
) -> mgp_mock.Record(result=str, args=list):
|
||
args = [required_arg, optional_arg]
|
||
# Multiple rows can be produced by returning an iterable of mgp_mock.Record:
|
||
return mgp_mock.Record(args=args, result="Hello World!")
|
||
```
|
||
|
||
The above example procedure returns two fields: `args` and `result`.
|
||
* `args` is a copy of the arguments passed to the procedure.
|
||
* `result` is "Hello World!".
|
||
|
||
Any errors can be reported by raising an Exception.
|
||
"""
|
||
return _register_proc(func, False)
|
||
|
||
|
||
def write_proc(func: typing.Callable[..., Record]):
|
||
"""
|
||
Register a function as a Memgraph write procedure.
|
||
|
||
The `func` needs to be a callable and optionally take `ProcCtx` as its first argument.
|
||
Other parameters of `func` will be bound to the passed arguments.
|
||
The full signature of `func` needs to be annotated with types. The return type must
|
||
be `Record(field_name=type, ...)`, and the procedure must produce either a complete
|
||
Record or None. Multiple records can be produced by returning an iterable of them.
|
||
Registering generator functions is currently not supported.
|
||
|
||
Example:
|
||
```
|
||
import mgp_mock
|
||
|
||
@mgp_mock.write_proc
|
||
def procedure(context: mgp_mock.ProcCtx,
|
||
required_arg: str,
|
||
optional_arg: mgp_mock.Nullable[str] = None
|
||
) -> mgp_mock.Record(result=mgp_mock.Vertex):
|
||
vertex = context.graph.create_vertex()
|
||
vertex_properties = vertex.properties
|
||
vertex_properties["required_arg"] = required_arg
|
||
if optional_arg is not None:
|
||
vertex_properties["optional_arg"] = optional_arg
|
||
|
||
return mgp.Record(result=vertex)
|
||
```
|
||
|
||
The above example procedure returns a newly created vertex that has
|
||
up to 2 properties:
|
||
* `required_arg` is always present and its value is the first
|
||
argument of the procedure.
|
||
* `optional_arg` is present if the second argument of the procedure
|
||
is not `null`.
|
||
|
||
Any errors can be reported by raising an Exception.
|
||
"""
|
||
return _register_proc(func, True)
|
||
|
||
|
||
class FuncCtx:
|
||
"""The context of the function being executed.
|
||
|
||
Access to a `FuncCtx` is only valid during a single execution of a transformation.
|
||
You should not globally store a `FuncCtx` instance.
|
||
The graph object within `FuncCtx` is not mutable.
|
||
"""
|
||
|
||
__slots__ = ("_graph",)
|
||
|
||
def __init__(self, graph):
|
||
if not isinstance(graph, (_mgp_mock.Graph, _mgp_mock.nx.MultiDiGraph)):
|
||
raise TypeError(f"Expected '_mgp_mock.Graph' or 'networkx.MultiDiGraph', got '{type(graph)}'")
|
||
|
||
self._graph = Graph(graph) if isinstance(graph, _mgp_mock.Graph) else Graph(_mgp_mock.Graph(graph))
|
||
|
||
def is_valid(self) -> bool:
|
||
"""
|
||
Check if the context is valid, i.e. if the contained structures may be used.
|
||
|
||
Returns:
|
||
A `bool` value that represents whether the context is valid.
|
||
|
||
Examples:
|
||
```context.is_valid()```
|
||
"""
|
||
return self._graph.is_valid()
|
||
|
||
|
||
def function(func: typing.Callable):
|
||
"""
|
||
Register a function as a Memgraph function.
|
||
|
||
The `func` needs to be a callable and optionally take `ProcCtx` as its first argument.
|
||
Other parameters of `func` will be bound to the passed arguments.
|
||
Only the function arguments need to be annotated with types. The return type doesn’t
|
||
need to be specified, but it has to be within `mgp_mock.Any`.
|
||
Registering generator functions is currently not supported.
|
||
|
||
Example:
|
||
```
|
||
import mgp_mock
|
||
|
||
@mgp_mock.function
|
||
def procedure(context: mgp_mock.FuncCtx,
|
||
required_arg: str,
|
||
optional_arg: mgp_mock.Nullable[str] = None
|
||
):
|
||
return_args = [required_arg]
|
||
if optional_arg is not None:
|
||
return_args.append(optional_arg)
|
||
# Return any result whose type is within mgp_mock.Any
|
||
return return_args
|
||
```
|
||
|
||
The above example function returns a list of the passed parameters:
|
||
* `required_arg` is always present and its value is the first
|
||
argument of the procedure.
|
||
* `optional_arg` is present if the second argument of the procedure
|
||
is not `null`.
|
||
|
||
Any errors can be reported by raising an Exception.
|
||
"""
|
||
raise_if_does_not_meet_requirements(func)
|
||
|
||
sig = inspect.signature(func)
|
||
|
||
params = tuple(sig.parameters.values())
|
||
if params and params[0].annotation is FuncCtx:
|
||
|
||
@wraps(func)
|
||
def wrapper(ctx, *args):
|
||
ctx._graph._graph.make_immutable()
|
||
|
||
result = func(ctx, *args)
|
||
|
||
# Invalidate context after execution
|
||
ctx._graph._graph.invalidate()
|
||
|
||
return result
|
||
|
||
else:
|
||
|
||
@wraps(func)
|
||
def wrapper(*args):
|
||
return func(*args)
|
||
|
||
return wrapper
|
||
|
||
|
||
def _wrap_exceptions():
|
||
def wrap_function(func):
|
||
@wraps(func)
|
||
def wrapped_func(*args, **kwargs):
|
||
try:
|
||
return func(*args, **kwargs)
|
||
except _mgp_mock.LogicErrorError as e:
|
||
raise LogicErrorError(e)
|
||
except _mgp_mock.ImmutableObjectError as e:
|
||
raise ImmutableObjectError(e)
|
||
|
||
return wrapped_func
|
||
|
||
def wrap_prop_func(func):
|
||
return None if func is None else wrap_function(func)
|
||
|
||
def wrap_member_functions(cls: type):
|
||
for name, obj in inspect.getmembers(cls):
|
||
if inspect.isfunction(obj):
|
||
setattr(cls, name, wrap_function(obj))
|
||
elif isinstance(obj, property):
|
||
setattr(
|
||
cls,
|
||
name,
|
||
property(
|
||
wrap_prop_func(obj.fget),
|
||
wrap_prop_func(obj.fset),
|
||
wrap_prop_func(obj.fdel),
|
||
obj.__doc__,
|
||
),
|
||
)
|
||
|
||
def defined_in_this_module(obj: object):
|
||
return getattr(obj, "__module__", "") == __name__
|
||
|
||
module = sys.modules[__name__]
|
||
for name, obj in inspect.getmembers(module):
|
||
if not defined_in_this_module(obj):
|
||
continue
|
||
if inspect.isclass(obj):
|
||
wrap_member_functions(obj)
|
||
elif inspect.isfunction(obj) and not name.startswith("_"):
|
||
setattr(module, name, wrap_function(obj))
|
||
|
||
|
||
_wrap_exceptions()
|