Memgraph Python Query Modules

Memgraph Python Query Modules

Develop custom query modules in Python for Memgraph graph database.

When to Use

• Create custom graph algorithms not available in MAGE library
• Implement business-specific logic that runs directly in the database
• Build read-only procedures (@mgp.read_proc) for data analysis
• Build write procedures (@mgp.write_proc) for data modification
• Create user-defined functions (@mgp.function) for use in Cypher queries
• Integrate external Python libraries with graph processing

Prerequisites

• Memgraph instance running
• Python 3.7.0+
• Install mgp locally for development: pip install mgp

Quick Reference

Key Types

Basic Patterns

Read Procedure

import mgp

@mgp.read_proc
def hello_world(
    context: mgp.ProcCtx,
    name: str
) -> mgp.Record(message=str):
    """A simple hello world procedure."""
    return mgp.Record(message=f"Hello, {name}!")

CALL module_name.hello_world("Memgraph") YIELD message;

Read Procedure with Graph Access

import mgp

@mgp.read_proc
def count_nodes_by_label(
    context: mgp.ProcCtx,
    label_name: str
) -> mgp.Record(count=int):
    """Count nodes with a specific label."""
    count = 0
    for vertex in context.graph.vertices:
        for label in vertex.labels:
            if label.name == label_name:
                count += 1
                break
    return mgp.Record(count=count)

Write Procedure

import mgp

@mgp.write_proc
def create_person(
    context: mgp.ProcCtx,
    name: str,
    age: mgp.Nullable[int] = None
) -> mgp.Record(vertex=mgp.Vertex):
    """Create a new Person node."""
    vertex = context.graph.create_vertex()
    vertex.add_label("Person")
    vertex.properties["name"] = name
    if age is not None:
        vertex.properties["age"] = age
    return mgp.Record(vertex=vertex)

User-Defined Function

import mgp

@mgp.function
def full_name(
    context: mgp.FuncCtx,
    first_name: str,
    last_name: str
) -> str:
    """Concatenate first and last name."""
    return f"{first_name} {last_name}"

MATCH (p:Person)
RETURN module_name.full_name(p.first_name, p.last_name) AS name;

Returning Multiple Records

import mgp

@mgp.read_proc
def get_neighbors(
    context: mgp.ProcCtx,
    node: mgp.Vertex
) -> mgp.Record(neighbor=mgp.Vertex, edge_type=str):
    """Get all neighbors of a node."""
    results = []
    for edge in node.out_edges:
        results.append(mgp.Record(
            neighbor=edge.to_vertex,
            edge_type=edge.type.name
        ))
    for edge in node.in_edges:
        results.append(mgp.Record(
            neighbor=edge.from_vertex,
            edge_type=edge.type.name
        ))
    return results

Working with Graph Elements

Vertex Operations

# Access vertex properties
vertex.id                    # Unique identifier
vertex.labels               # Iterable of labels
vertex.properties           # Dict-like properties
vertex.in_edges             # Incoming edges
vertex.out_edges            # Outgoing edges

# Modify (write procedures only)
vertex.add_label("Label")
vertex.remove_label("Label")
vertex.properties["key"] = value

Edge Operations

edge.id                     # Unique identifier
edge.type                   # EdgeType object
edge.type.name              # Relationship type name
edge.from_vertex            # Source vertex
edge.to_vertex              # Target vertex
edge.properties             # Dict-like properties

Path Operations

path = mgp.Path(start_vertex)
path.expand(edge)           # Extend path
path.vertices               # List of vertices
path.edges                  # List of edges

Advanced Patterns

Long-Running Procedures with Abort Check

@mgp.read_proc
def long_running_analysis(
    context: mgp.ProcCtx
) -> mgp.Record(result=int):
    """A procedure that can be terminated."""
    count = 0
    try:
        for vertex in context.graph.vertices:
            if context.check_must_abort():
                break
            # Expensive computation...
            count += 1
    except mgp.AbortError:
        pass
    return mgp.Record(result=count)

Using External Libraries

import mgp

try:
    import networkx as nx
    HAS_NETWORKX = True
except ImportError:
    HAS_NETWORKX = False

@mgp.read_proc
def pagerank(
    context: mgp.ProcCtx,
    damping: float = 0.85
) -> mgp.Record(node_id=int, rank=float):
    """Calculate PageRank using NetworkX."""
    if not HAS_NETWORKX:
        raise Exception("NetworkX not installed")
    
    G = nx.DiGraph()
    for vertex in context.graph.vertices:
        G.add_node(vertex.id)
    for vertex in context.graph.vertices:
        for edge in vertex.out_edges:
            G.add_edge(vertex.id, edge.to_vertex.id)
    
    ranks = nx.pagerank(G, alpha=damping)
    return [mgp.Record(node_id=nid, rank=rank) 
            for nid, rank in ranks.items()]

Logging

import mgp

logger = mgp.Logger()
logger.trace("Trace message")
logger.debug("Debug message")
logger.info("Info message")
logger.warning("Warning message")
logger.error("Error message")
logger.critical("Critical message")

Deployment

Module Locations

Copy to Container

docker cp my_module.py memgraph:/usr/lib/memgraph/query_modules/
echo "CALL mg.load_all();" | docker exec -i memgraph mgconsole

Volume Mount

docker run -d \
  -p 7687:7687 \
  -v $(pwd)/modules:/usr/lib/memgraph/query_modules \
  --name memgraph \
  memgraph/memgraph-mage

Install External Libraries

docker exec -i -u memgraph memgraph bash -c "pip install pandas networkx"
echo "CALL mg.load_all();" | docker exec -i memgraph mgconsole

Module Management (Cypher)

CALL mg.load_all();              -- Load all modules
CALL mg.load("module_name");     -- Load specific module
CALL mg.procedures() YIELD *;    -- List procedures
CALL mg.functions() YIELD *;     -- List functions

Error Handling

import mgp

@mgp.read_proc
def safe_procedure(
    context: mgp.ProcCtx,
    vertex_id: int
) -> mgp.Record(result=str):
    try:
        vertex = context.graph.get_vertex_by_id(vertex_id)
        return mgp.Record(result=f"Found vertex: {vertex.id}")
    except IndexError:
        raise Exception(f"Vertex with ID {vertex_id} not found")
    except mgp.InvalidContextError:
        raise Exception("Invalid context")

Best Practices

1. Type Annotations: Always fully annotate function signatures
2. Documentation: Use docstrings for all procedures and functions
3. Error Handling: Raise meaningful exceptions with clear messages
4. Memory: Don't store graph objects globally (they become invalid)
5. Performance: Use context.check_must_abort() in long-running procedures
6. Testing: Use the mock API for unit testing

Troubleshooting

Additional Resources

For detailed API documentation and examples, see references/REFERENCE.md.