Quick and dirty neo4j database dump when all else fails

neo4dump.py

import json
import click
import logging
from pathlib import Path
from py2neo import Graph

@click.command(help='''
  Quick and dirty neo4j database dump when all else fails. The format is:

  nodes.jsonl: {
    id: neo4j_node_id,
    meta: node_attributes,
  }

  relationships.jsonl: {
    id: neo4j_rel_id,
    source: neo4_node_id,
    target: neo4j_node_id,
    meta: rel_attributes,
  }

  Not recommended, you should use neo4j-dump utils or some extension,
  this is only useful when you're trying to dump someone else's neo4j
  database... :facepalm
''')
@click.option('-u', '--uri', envvar='NEO4J_URI', default='bolt://localhost:7687', help='URI to access Neo4J database', show_default=True)
@click.option('-U', '--username', envvar='NEO4J_USERNAME', default='neo4j', type=str, help='Username to access Neo4J Database', show_default=True)
@click.option('-P', '--password', envvar='NEO4J_PASSWORD', default=None, type=str, help='Password to access Neo4J Database', show_default=True)
@click.option('-o', '--output', type=click.Path(path_type=Path), default=Path.cwd(), help='Output directory for dump', show_default=True)
@click.option('-c', '--continue', type=bool, is_flag=True, default=False, help='Whether or not to continue where it left off')
@click.option('-v', '--verbose', count=True, default=0, help='How verbose this should be, more -v = more verbose')
def cli(**args):
  uri = args.get('uri')
  username = args.get('username')
  password = args.get('password')
  output = args.get('output')
  continue_ = args.get('continue')
  verbose = args.get('verbose')
  logging.basicConfig(level=30 - (verbose*10))

  logging.info('Connecting to neo4j')
  kwargs = {}
  if password is not None:
    kwargs['auth'] = (username, password)
  graph = Graph(uri, **kwargs)

  output.mkdir(exist_ok=True)

  node_ids = set()
  rel_ids = set()
  if continue_: 
    if (output / 'nodes.jsonl').exists():
      logging.info('Loading existing node ids')
      with (output / 'nodes.jsonl').open('r') as fr:
        node_ids = {node['id'] for node in map(json.loads, fr)}

    if (output / 'relationships.jsonl').exists():
      logging.info('Loading existing relationship ids')
      with (output / 'relationships.jsonl').open('r') as fr:
        rel_ids = {rel['id'] for rel in map(json.loads, fr)}

  logging.info('Dumping nodes')
  with (output / 'nodes.jsonl').open('a' if continue_ else 'w') as fw:
    for node_id in graph.nodes:
      if node_id in node_ids: continue
      n = dict(id=node_id, meta=dict(graph.nodes[node_id]))
      print(json.dumps(n), file=fw)

  logging.info('Dumping relationships')
  with (output / 'relationships.jsonl').open('a' if continue_ else 'w') as fw:
    for rel_id in graph.relationships:
      if rel_id in rel_ids: continue
      r = graph.relationships[rel_id]
      source, target = r.nodes
      n = dict(
        id=rel_id,
        source=source.identity,
        target=target.identity,
        meta=dict(r),
      )
      print(json.dumps(n), file=fw)

if __name__ == '__main__':
  cli()