mage/python/graph_analyzer.py at main · memgraph/mage · GitHub

This repository was archived by the owner on Jan 23, 2026. It is now read-only.

History

296 lines (229 loc) · 9.25 KB

Raw

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

198

199

200

201

202

203

204

205

206

207

208

209

210

211

212

213

214

215

216

217

218

219

220

221

222

223

224

225

226

227

228

229

230

231

232

233

234

235

236

237

238

239

240

241

242

243

244

245

246

247

248

249

250

251

252

253

254

255

256

257

258

259

260

261

262

263

264

265

266

267

268

269

270

271

272

273

274

275

276

277

278

279

280

281

282

283

284

285

286

287

288

289

290

291

292

293

294

295

296

import sys

import mgp

from collections import OrderedDict

from itertools import chain, repeat

from inspect import cleandoc

from typing import List, Tuple

try:

import networkx as nx

except ImportError as import_error:

sys.stderr.write(

(

"\n"

"NOTE: Please install networkx to be able to use graph_analyzer "

"module. Using Python:\n" + sys.version + "\n"

)

raise import_error

# Imported last because it also depends on networkx.

from mgp_networkx import MemgraphMultiDiGraph # noqa E402

_MAX_LIST_SIZE = 10

@mgp.read_proc

def help() -> mgp.Record(name=str, value=str):

"""Shows manual page for graph_analyzer."""

records = []

def make_records(name, doc):

return (

mgp.Record(name=n, value=v)

for n, v in zip(chain([name], repeat("")), cleandoc(doc).splitlines())

)

for func in (help, analyze, analyze_subgraph):

records.extend(

make_records("Procedure '{}'".format(func.__name__), func.__doc__)

)

for m, v in _get_analysis_mapping().items():

records.extend(make_records("Analysis '{}'".format(m), v.__doc__))

return records

@mgp.read_proc

def analyze(

context: mgp.ProcCtx, analyses: mgp.Nullable[List[str]] = None

) -> mgp.Record(name=str, value=str):

"""

Shows graph information.

In case of multiple results, only the first 10 will be shown.

The optional parameter is a list of graph analyses to run.

If NULL, all available analyses are run.

Example call (give all information):

CALL graph_analyzer.analyze() YIELD *;

Example call (with parameter):

CALL graph_analyzer.analyze(['nodes', 'edges']) YIELD *;

"""

g = MemgraphMultiDiGraph(ctx=context)

recs = _analyze_graph(context, g, analyses)

return [mgp.Record(name=name, value=value) for name, value in recs]

@mgp.read_proc

def analyze_subgraph(

context: mgp.ProcCtx,

vertices: mgp.List[mgp.Vertex],

edges: mgp.List[mgp.Edge],

analyses: mgp.Nullable[List[str]] = None,

) -> mgp.Record(name=str, value=str):

"""

Shows subgraph information.

In case of multiple results, only the first 10 will be shown.

The optional parameter is a list of graph analyses to run.

If NULL, all available analyses are run.

Example call (give all information):

MATCH (n)-[e]->(m) WITH

collect(n) AS nodes,

collect(e) AS edges

CALL graph_analyzer.analyze_subgraph(nodes, edges) YIELD *

RETURN name, value;

Example call (with parameter):

MATCH (n)-[e]->(m) WITH

collect(n) AS nodes,

collect(e) AS edges

CALL graph_analyzer.analyze_subgraph(nodes, edges, ['nodes', 'edges'])

YIELD *

RETURN name, value;

"""

vertices, edges = map(set, [vertices, edges])

g = nx.subgraph_view(

MemgraphMultiDiGraph(ctx=context),

lambda n: n in vertices,

lambda n1, n2, e: e in edges,

)

recs = _analyze_graph(context, g, analyses)

return [mgp.Record(name=name, value=value) for name, value in recs]

def _get_analysis_mapping():

return OrderedDict(

[

("nodes", _number_of_nodes),

("edges", _number_of_edges),

("bridges", _bridges),

("articulation_points", _articulation_points),

("avg_degree", _avg_degree),

("sorted_nodes_degree", _sorted_nodes_degree),

("self_loops", _self_loops),

("is_bipartite", _is_bipartite),

("is_planar", _is_planar),

("is_biconnected: ", _is_biconnected),

("is_weakly_connected", _is_weakly_connected),

("number_of_weakly_components", _weakly_components),

("is_strongly_connected", _is_strongly_connected),

("strongly_components", _strongly_components),

("is_dag", _is_dag),

("is_eulerian", _is_eulerian),

("is_forest", _is_forest),

("is_tree", _is_tree),

]

)

def _get_analysis_func(name: str):

_name_to_proc = _get_analysis_mapping()

return _name_to_proc.get(name.lower())

def _get_analysis_funcs():

return _get_analysis_mapping().values()

def _analyze_graph(

context: mgp.ProcCtx, g: nx.MultiDiGraph, analyses: List[str]

) -> List[Tuple[str, str]]:

functions = (

_get_analysis_funcs()

if analyses is None

else [_get_analysis_func(name) for name in analyses]

)

records = []

for index, f in enumerate(functions):

context.check_must_abort()

if f is None:

raise KeyError("Graph analysis is not supported: " + analyses[index])

name, value = f(g)

if isinstance(value, (list, set, tuple)):

value = list(value)[:_MAX_LIST_SIZE]

records.append((name, str(value)))

return records

def _number_of_nodes(g: nx.MultiDiGraph) -> Tuple[str, int]:

"""Returns number of nodes."""

return "Number of nodes", nx.number_of_nodes(g)

def _number_of_edges(g: nx.MultiDiGraph) -> Tuple[str, int]:

"""Returns number of edges."""

return "Number of edges", nx.number_of_edges(g)

def _avg_degree(g: nx.MultiDiGraph) -> Tuple[str, float]:

"""Returns average degree."""

_, number_of_nodes = _number_of_nodes(g)

_, number_of_edges = _number_of_edges(g)

avg_degree = 0 if number_of_nodes == 0 else number_of_edges / number_of_nodes

return "Average degree", avg_degree

def _sorted_nodes_degree(g: nx.MultiDiGraph) -> Tuple[str, List[int]]:

"""Returns list of sorted nodes degree. [(node_id, degree), ...]"""

nodes_degree = [(n, g.degree(n)) for n in g.nodes()]

nodes_degree.sort(key=lambda x: x[1], reverse=True)

return "Sorted nodes degree", nodes_degree

def _self_loops(g: nx.MultiDiGraph) -> Tuple[str, int]:

"""Returns number of self loops."""

return "Self loops", sum((1 if e[0] == e[1] else 0 for e in g.edges()))

def _is_bipartite(g: nx.MultiDiGraph) -> Tuple[str, bool]:

"""Checks if graph is bipartite."""

_, number_of_nodes = _number_of_nodes(g)

ret = (

False if number_of_nodes == 0 else nx.algorithms.bipartite.basic.is_bipartite(g)

)

return "Is bipartite", ret

def _is_planar(g: nx.MultiDiGraph) -> Tuple[str, bool]:

"""Checks if graph is planar."""

_, number_of_nodes = _number_of_nodes(g)

ret = (

False if number_of_nodes == 0 else nx.algorithms.planarity.check_planarity(g)[0]

)

return "Is planar", ret

def _is_biconnected(g: nx.MultiDiGraph) -> Tuple[str, bool]:

"""Check if graph is biconnected."""

_, number_of_nodes = _number_of_nodes(g)

ret = (

False

if number_of_nodes == 0

else nx.is_biconnected(nx.MultiDiGraph.to_undirected(g))

)

return "Is biconnected", ret

def _is_weakly_connected(g: nx.MultiDiGraph) -> Tuple[str, bool]:

"""Check if graph is weakly connected."""

_, number_of_nodes = _number_of_nodes(g)

ret = False if number_of_nodes == 0 else nx.is_weakly_connected(g)

return "Is weakly connected", ret

def _is_strongly_connected(g: nx.MultiDiGraph) -> Tuple[str, bool]:

"""Checks if graph is strongly connected."""

_, number_of_nodes = _number_of_nodes(g)

ret = False if number_of_nodes == 0 else nx.is_strongly_connected(g)

return "Is strongly connected", ret

def _is_dag(g: nx.MultiDiGraph) -> Tuple[str, bool]:

"""Check if graph is directed acyclic graph (DAG)"""

_, number_of_nodes = _number_of_nodes(g)

ret = (

False

if number_of_nodes == 0

else nx.algorithms.dag.is_directed_acyclic_graph(g)

)

return "Is DAG", ret

def _is_eulerian(g: nx.MultiDiGraph) -> Tuple[str, bool]:

"""Checks if graph is Eulerian."""

_, number_of_nodes = _number_of_nodes(g)

ret = False if number_of_nodes == 0 else nx.algorithms.euler.is_eulerian(g)

return "Is eulerian", ret

def _is_forest(g: nx.MultiDiGraph) -> Tuple[str, bool]:

"""Checks if graph is forest, all components must be trees."""

_, number_of_nodes = _number_of_nodes(g)

ret = False if number_of_nodes == 0 else nx.algorithms.tree.recognition.is_forest(g)

return "Is forest", ret

def _is_tree(g: nx.MultiDiGraph) -> Tuple[str, bool]:

"""Checks if graph is tree."""

_, number_of_nodes = _number_of_nodes(g)

ret = False if number_of_nodes == 0 else nx.algorithms.tree.recognition.is_tree(g)

return "Is tree", ret

def _bridges(g: nx.MultiDiGraph) -> Tuple[str, int]:

"""Returns number of bridges, multiple edges between same nodes are

mapped to one edge."""

return "Number of bridges", sum(1 for _ in nx.bridges(nx.Graph(g)))

def _articulation_points(g: nx.MultiDiGraph):

"""Returns number of articulation points."""

undirected = nx.MultiDiGraph.to_undirected(g)

return (

"Number of articulation points",

sum(1 for _ in nx.articulation_points(undirected)),

)

def _weakly_components(g: nx.MultiDiGraph):

"""Returns number of weakly components."""

comps = nx.algorithms.components.number_weakly_connected_components(g)

return "Number of weakly connected components", comps

def _strongly_components(g: nx.MultiDiGraph):

"""Returns number of strongly connected components."""

comps = nx.algorithms.components.number_strongly_connected_components(g)

return "Number of strongly connected components", comps

mage/python/graph_analyzer.py at main · memgraph/mage · GitHub

Expand file tree

Breadcrumbs

graph_analyzer.py

Latest commit

History

Breadcrumbs

graph_analyzer.py

File metadata and controls