# python-connector-neo4j This library extends the [neo4j python driver](https://pypi.org/project/neo4j/) and offers several benefits for easier development and improved application performance. ## Features * Verifies connection is valid when opening new session and attempts to recover gracefully if possible, applications will not need to be re-deployed on a neo4j server issue * Automatic transaction management and rollback on uncaught exception * Application level access to open session, eliminating the need to pass around a session object between modules while remaining transaction safe * Defaults to `READ` mode to avoid connections to main database node unless explicitly necessary * Session isolation per web request if using Flask * Robust test suite handling edge cases ## Usage When activated, the context manager makes `connector_neo4j.get_session()` available to access a neo4j connection for running queries. More than one session **cannot** be opened in the same context. Transactions are automatically opened, closed, or rolledback by the context manager. During an open transaction, un-committed data can be queried, **but** it will not be commited if an uncaught exception occurs inside of the context manager. ## Storage Modes The library can operate in different "storage modes" which define the **scope** that a neo4j session objects is stored in for access by parts of an application. The session mode can be explicitly set when passing in neo4j connection information, or inferred based on packages installed. * `SINGLE` => One session is stored for an application, not thread safe. * `FLASK` => One session is stored per web request, thread safe. ## Usage Examples The following examples connect to a local neo4j server that can be spun up using the following command with docker. Alternatively, configuration can be modified to send traffic to another cluster. ``` $ docker run -p7687:7687 -p7474:7474 -v=$HOME/neo4j/plugins:/plugins -v=$HOME/neo4j/data:/data -v $HOME/neo4j/certificats:/certificates --env=NEO4J_AUTH=neo4j/blerg neo4j:3.5.18 ``` ### Initialize Configuration Configuration parameters need to be set once during an application's runtime. ```python import connector_neo4j connector_neo4j.configure( 'bolt://localhost:7687', 'neo4j', 'blerg', override_storage_mode=connector_neo4j.SessionStorageMode.SINGLE ) ``` :bulb: Lower level [neo4j driver settings](https://neo4j.com/docs/api/python-driver/current/api.html#driver-configuration) can be passed as additional kwargs. :bangbang: This step **only** sets configuration options and selects a storage mode if one is not supplied. Connectivity to the server is **not** checked until a session context manager is activated! ### Singleton Script :warning: This is just an example, many of the benefits of this library are not obvious or useful when using a single module script. The lower level neo4j library can work in a very similar manner as a context manager. ```python import connector_neo4j def main(): print('----') try_create('Neo', True) match() print('----') try_create('Morpheus', False) match() print('----') cleanup() def try_create(name, fail): print(f'Try to create Person node, fail = {fail}') try: with connector_neo4j.Neo4jSession(transaction=True): session = connector_neo4j.get_session() session.run( 'CREATE (n:Person {name: $name})', name=name ) results = session.run('MATCH (n:Person) RETURN n') for results in results: print(results['n']) if fail: raise Exception('something went wrong!') except Exception: pass @connector_neo4j.Neo4jSession() def match(): print('Match all Person nodes.') session = connector_neo4j.get_session() results = session.run('MATCH (n:Person) RETURN n') for result in results: print(result['n']) @connector_neo4j.Neo4jSession(transaction=True) def cleanup(): session = connector_neo4j.get_session() session.run('MATCH (n:Person) DETACH DELETE n') if __name__ == '__main__': connector_neo4j.configure( 'bolt://localhost:7687', 'neo4j', 'blerg', connector_neo4j.SessionStorageMode.SINGLE ) main() ``` _Output:_ ``` ---- Try to create Person node, fail = True Match all Person nodes. ---- Try to create Person node, fail = False Match all Person nodes. ``` ### Flask Web Server If Flask is installed, and storage mode is not overridden, the session will be stored using [Flask.g](https://flask.palletsprojects.com/en/2.0.x/appcontext/#storing-data) object. Using a session context as a decorator on a handler can be very useful, keeping the entire request transaction safe, as shown in the example below. _Flask App with Handlers:_ ```python from connector_neo4j import Neo4jSession from connector_neo4j import configure from db import do_create from db import do_delete from db import do_match from flask import Flask from flask import jsonify from flask import request app = Flask(__name__) configure( 'bolt://localhost:7687', 'neo4j', 'blerg' ) @app.route('/people', methods=['POST']) @Neo4jSession(transaction=True) def create(): do_create(request.json['name']) if request.args.get('error'): raise Exception('something went wrong!') return jsonify({}) @app.route('/people', methods=['GET']) @Neo4jSession() def match(): return jsonify(do_match()) @app.route('/people', methods=['DELETE']) @Neo4jSession(transaction=True) def delete(): do_delete() return jsonify({}) ``` _DB Layer:_ ```python from connector_neo4j import get_session def do_create(name): session = get_session() session.run( 'CREATE (n:Person {name: $name})', name=name ) def do_match(): session = get_session() results = session.run('MATCH (n:Person) RETURN n') return [x['n']['name'] for x in results] def do_delete(): session = get_session() session.run('MATCH (n:Person) DETACH DELETE n') ``` ### Writing Tests Patching the `get_session()` call allows for easy testing to make sure the queries are being made as expected and/or to return fake neo4j nodes. It is not necessary to patch the low level neo4j session and driver, there are ample tests in this repository covering that process. ```python from unittest.mock import patch from db import do_create from db import do_match from neo4j.graph import Graph from neo4j.graph import Node @patch('db.get_session') def test_mock_session(mock_get_session): mock_session = mock_get_session.return_value assert not mock_get_session.called do_create('Neo') assert mock_get_session.called assert mock_session.run.called_once_with( 'CREATE (n:Person {name: $name})', name='Neo' ) @patch('db.get_session') def test_mock_return_node(mock_get_session): mock_session = mock_get_session.return_value mock_node = Node( Graph(), '1', n_labels=['Person'], properties={'name': 'Neo'} ) mock_session.run.return_value = [ {'n': mock_node} ] results = do_match() assert results == ['Neo'] ``` ## Package Development This project uses [uv](https://docs.astral.sh/uv/) for packaging and dependency management, with `hatchling` as the build backend. ### Setup Create the virtual environment and install dependencies (including the `dev` group): ``` $ make env # uv sync ``` The Flask integration lives in the `flask` extra: ``` $ uv sync --extra flask ``` ### Running Tests Run tests: ``` $ make test_unit # uv run pytest tests ``` Run a single test: ``` $ uv run pytest tests/test_session_flask.py::test_read_connection -vv ``` Coverage report: ``` $ uv run pytest tests --cov connector_neo4j --cov-report html ``` ### Linting Check formatting and lint (`ruff`): ``` $ make lint # ruff check + ruff format --diff ``` Auto-fix and format: ``` $ make fmt # ruff check --fix + ruff format ```