Drive Elasticsearch and Neo4j from Python

Karim, Veille's developer, wants to wire the product API to both engines without rewriting the queries in raw HTTP. Sami, for his part, wants to automate a monthly report: "give me the top authors over the last six months and their favorite categories". This module shows how all of that fits into fifteen lines of Python, inside a kit container where both official clients are already installed.

The veille-python container: Python without Python

The kit ships a ready-to-use veille-python container. Its Dockerfile installs two packages and nothing else: elasticsearch>=9,<10 and neo4j>=5.28,<6. The kit's python/ folder is mounted at /work: any python/my_script.py you write is immediately runnable inside the container. The environment variables ES_URL, ES_USER, ES_PASSWORD, NEO4J_URI (bolt://neo4j:7687), NEO4J_USER, NEO4J_PASSWORD are injected by docker-compose.yml: the script reads them without hardcoding anything.

Two commands are enough:

./lab.sh python veille.py "climate change"   # runs python/veille.py with an argument
./lab.sh python-shell                        # opens an interactive Python interpreter

The Windows PowerShell equivalent is .\lab.ps1 python veille.py "climate change". No pip install on the host machine: that is the kit's principle, and it is what makes the course reproducible from one workstation to another.

Where do you write?

Your .py files go in the kit's python/ folder. The container sees it under /work. Save the file from your editor, rerun ./lab.sh python …: the new version is picked up immediately, without rebuilding the image.

The elasticsearch 9 client in five gestures

The Python Elasticsearch client mirrors the REST API. You write idiomatic Python, the library builds the HTTP requests.

Connect

from elasticsearch import Elasticsearch
import os

es = Elasticsearch(
    os.environ["ES_URL"],
    basic_auth=(os.environ["ES_USER"], os.environ["ES_PASSWORD"]),
    request_timeout=30,
)
info = es.info()
print(f"Elasticsearch {info['version']['number']} — cluster « {info['cluster_name']} »")

Expected output:

Elasticsearch 9.5.3 — cluster « veille »

Elasticsearch(url, basic_auth=(u, p)) is the canonical call in version 9. The client reuses a pool of HTTP connections; do not create a client per request.

Search

Any Query DSL query goes through es.search(index=..., query=..., size=..., source=[...]). Note: starting with client 9, arguments are written directly (query=...), no longer wrapped in body={"query": ...}.

rep = es.search(
    index="news",
    size=3,
    query={
        "multi_match": {
            "query": "climate change",
            "fields": ["headline^3", "short_description"],
        }
    },
    source=["headline", "date", "category"],
)
for h in rep["hits"]["hits"]:
    print(round(h["_score"], 2), h["_source"]["headline"])

Expected output (scores may vary slightly with BM25):

19.12 Change Is Here. Climate Change.
15.83 Ellen DeGeneres Warns Climate Change Will Be 'Dangerous'
14.44 Climate Change: Time for Action Is Now

The total number of results is read from rep["hits"]["total"]["value"] — 2,834 for climate change.

Read, index, update

doc = es.get(index="news", id="1")               # a document by _id
es.index(index="news", id="200854", document={   # create or replace
    "headline": "Veille lance sa v2",
    "short_description": "Un moteur combinant Elasticsearch et Neo4j.",
    "category": "TECH",
    "authors": "Inès Bouraoui",
    "date": "2026-09-09",
})
es.update(index="news", id="200854", doc={"category": "BUSINESS"})

Each method returns a dictionary with _id, _version and result (created, updated, noop).

Bulk indexing: helpers.bulk

Writing fifty thousand separate es.index calls costs too much. The elasticsearch.helpers package provides bulk, which prepares the NDJSON format and handles retries.

from elasticsearch import helpers

actions = [
    {"_index": "news", "_id": str(200_855 + i), "_source": {
        "headline": f"Article de test numéro {i}",
        "short_description": "Généré par le module 13.",
        "category": "TECH",
        "authors": "Sami Karray",
        "date": "2026-09-09",
    }}
    for i in range(200)
]
ok, erreurs = helpers.bulk(es, actions, chunk_size=100, request_timeout=60)
print(f"{ok} documents indexés, {len(erreurs)} erreurs")

Expected output:

200 documents indexés, 0 erreurs

chunk_size=100 sends documents in packets of one hundred. On the News corpus, the kit importer (importer/import_news.py) follows exactly this logic, with batches of two thousand.

Handling errors

Two exceptions come up more often than the others. Do not let them bubble up without a clear message.

from elasticsearch import Elasticsearch, AuthenticationException, NotFoundError, TransportError

try:
    es = Elasticsearch(os.environ["ES_URL"], basic_auth=(os.environ["ES_USER"], os.environ["ES_PASSWORD"]))
    es.info()
except AuthenticationException:
    print("401 : identifiants invalides. Vérifiez ELASTIC_PASSWORD dans .env,")
    print("et si vous l'avez changé après le premier démarrage, faites ./lab.sh reset.")
    raise
except TransportError as e:
    print(f"Elasticsearch injoignable ({e}). Le service est-il « healthy » ? ./lab.sh status")
    raise

AuthenticationException corresponds to HTTP 401 "security_exception ... unable to authenticate user [elastic]". NotFoundError shows up for es.get on a missing _id. TransportError covers network problems.

The neo4j 5 client in five gestures

The official Neo4j client follows a slightly different model: a driver (driver), a session, and transactions — implicit via session.run, or explicit via execute_read and execute_write.

Connect

from neo4j import GraphDatabase
import os

pilote = GraphDatabase.driver(
    os.environ["NEO4J_URI"],
    auth=(os.environ["NEO4J_USER"], os.environ["NEO4J_PASSWORD"]),
)
pilote.verify_connectivity()
print("Neo4j accessible via", os.environ["NEO4J_URI"])

verify_connectivity() tests the connection without running a query; useful at application startup to fail fast if the driver cannot reach the server.

One query, some parameters

with pilote.session() as session:
    resultat = session.run(
        "MATCH (au:Auteur {nom: $nom})<-[:ECRIT_PAR]-(a:Article) "
        "RETURN a.titre AS titre, a.date AS date "
        "ORDER BY a.date DESC LIMIT 3",
        nom="Lee Moran",
    )
    for enregistrement in resultat:
        print(enregistrement["date"], enregistrement["titre"])

Expected output (your numbers may differ with the recent articles):

2018-05-25 Trump Cracks Terrible Joke About Assassinated Journalist
2018-05-24 Ivanka Trump's Latest Bit Of Advice Doesn't Sit Well With Some Twitter Users
2018-05-23 Rudy Giuliani Sparks Confusion With Baffling Answers

Always use parameters ($nom), never f"…{nom}…": it is faster (Cypher caches the query) and it rules out any injection.

Explicit transactions

For robust code, the good practice is to wrap a query in a function and pass it to execute_read or execute_write. The driver handles retries on network hiccups.

def top_auteurs(tx, limite: int):
    rep = tx.run(
        "MATCH (au:Auteur)<-[:ECRIT_PAR]-(a:Article) "
        "RETURN au.nom AS auteur, count(a) AS articles "
        "ORDER BY articles DESC LIMIT $limite",
        limite=limite,
    )
    return [dict(r) for r in rep]

with pilote.session() as session:
    for ligne in session.execute_read(top_auteurs, 5):
        print(f"{ligne['articles']:>5}  {ligne['auteur']}")

Expected output:

 4954  Reuters
 2433  Lee Moran
 1915  Ron Dicker
 1328  Ed Mazza
 1145  Cole Delbyck

These five lines are the most prolific authors of the News corpus, as loaded by ./lab.sh cypher 11-charger-news.cypher.

Close cleanly

pilote.close()

Or, better, use with GraphDatabase.driver(...) as pilote: — the connection closes even on an exception.

Handling errors

from neo4j.exceptions import ServiceUnavailable, AuthError

try:
    pilote = GraphDatabase.driver(os.environ["NEO4J_URI"], auth=(os.environ["NEO4J_USER"], os.environ["NEO4J_PASSWORD"]))
    pilote.verify_connectivity()
except AuthError:
    print("Neo4j : mot de passe refusé. Si vous avez modifié NEO4J_PASSWORD après le premier démarrage,")
    print("le mot de passe vit dans le volume : ./lab.sh reset puis ./lab.sh up.")
    raise
except ServiceUnavailable as e:
    print(f"Neo4j injoignable ({e}). Vérifiez ./lab.sh status.")
    raise

AuthError covers the HTTP 401 returned by the Bolt server. ServiceUnavailable covers timeouts and a service that is not yet ready.

Guided reading of python/veille.py

The kit includes python/veille.py, a script that combines both engines: Elasticsearch finds the articles relevant to a query, Neo4j starts from their authors and walks back to other articles to recommend. It is the nucleus of module 16.

rechercher(): the Elasticsearch part

def rechercher(es: Elasticsearch, texte: str, taille: int = 5) -> list[dict]:
    rep = es.search(
        index="news",
        size=taille,
        query={
            "multi_match": {
                "query": texte,
                "fields": ["headline^3", "short_description"],
                "type": "best_fields",
            }
        },
        source=["headline", "authors", "category", "date"],
    )
    return [{"id": int(h["_id"]), "score": round(h["_score"], 2), **h["_source"]} for h in rep["hits"]["hits"]]

Three points to note. headline^3 gives the title a weight three times higher than the description: a word in the title carries more in the score. type: "best_fields" keeps, for each document, the best of the two fields rather than summing them. source=[...] filters the returned fields to save bandwidth — the script needs neither the link nor the full text.

The flattened return merges _score, _id (the row number, converted to int for use on the Cypher side) and the _source into a single dictionary per article.

recommander(): the Neo4j part

def recommander(session, ids: list[int], limite: int = 5) -> list[dict]:
    cypher = """
        MATCH (a:Article)-[:ECRIT_PAR]->(au:Auteur)<-[:ECRIT_PAR]-(reco:Article)-[:PUBLIE_DANS]->(c:Categorie)
        WHERE a.id IN $ids AND NOT reco.id IN $ids AND au.nom <> 'Reuters'
        RETURN reco.titre AS titre, au.nom AS auteur, c.nom AS categorie, reco.date AS date
        ORDER BY date DESC
        LIMIT $limite
    """
    return [dict(r) for r in session.run(cypher, ids=ids, limite=limite)]

The pattern reads like a sentence: "I start from the articles found, walk up to their authors, walk back down to other articles written by these same authors, and pick up the category". The two guardrails — NOT reco.id IN $ids and au.nom <> 'Reuters' — respectively avoid recommending the article the user just read and swamping the result with the Reuters wire, five thousand articles by itself.

main(): the glue between the two

def main() -> None:
    es = Elasticsearch(ES_URL, basic_auth=(ES_USER, ES_PASSWORD))
    info = es.info()
    print(f"Elasticsearch {info['version']['number']} — cluster « {info['cluster_name']} »")

    resultats = rechercher(es, REQUETE)
    print(f"\n{len(resultats)} articles pour « {REQUETE} » :")
    for r in resultats:
        print(f"  [{r['score']:>5}] {r['date']}  {r['headline']}  — {r['authors'] or 'sans auteur'} ({r['category']})")

    pilote = GraphDatabase.driver(NEO4J_URI, auth=(NEO4J_USER, NEO4J_PASSWORD))
    with pilote.session() as session:
        recos = recommander(session, [r["id"] for r in resultats])
    pilote.close()

    print(f"\n{len(recos)} recommandations (mêmes auteurs, autres articles) :")
    for r in recos:
        print(f"  {r['date']}  {r['titre']}  — {r['auteur']} ({r['categorie']})")

Nothing more: two calls to the clients, one display loop. All the business logic fits in two functions and one entry point.

Real execution

./lab.sh python veille.py climate change

Observed output on the full corpus (scores may vary slightly):

Elasticsearch 9.5.3 — cluster « veille »

5 articles pour « climate change » :
  [19.12] 2015-06-24  Change Is Here. Climate Change.  — Dominique Browning (ENVIRONMENT)
  [15.83] 2016-04-25  Ellen DeGeneres Warns Climate Change Will Be 'Dangerous'  — Kate Sheppard (ENVIRONMENT)
  [15.44] 2013-12-04  Climate Change Isn't Real, Say Fewer Than Ever  — Kate Sheppard (POLITICS)
  [14.44] 2014-11-19  Climate Change: Time for Action Is Now  — Karen Feridun (POLITICS)
  [13.98] 2016-08-15  Louisiana Flooding Is Climate Change  — Kate Sheppard (ENVIRONMENT)

5 recommandations (mêmes auteurs, autres articles) :
  2018-04-19  The E.P.A. Rolls Back A Regulation On Coal Ash  — Kate Sheppard (ENVIRONMENT)
  2017-11-08  A Vote For Common Sense And Public Health  — Karen Feridun (POLITICS)
  2017-06-01  Trump Just Pulled Out Of The Paris Agreement  — Kate Sheppard (POLITICS)
  2016-11-14  What Do We Tell The Children?  — Dominique Browning (ENVIRONMENT)
  2016-09-08  The State Of The Air We Breathe  — Karen Feridun (ENVIRONMENT)

The first article has a score almost four points above the second: the two words "climate" and "change" are both in its title, and headline^3 weighs. The recommendations bring back articles published after those found, written by the same authors — that is the value of the Elasticsearch + Neo4j pairing.

Writing your own script: python/top_auteurs.py

Sami wants, for a monthly report, the authors who have written the most over the last six months of 2017, and for each their three main categories. Elasticsearch does the aggregation, Neo4j completes with the graph.

Create the file python/top_auteurs.py :

#!/usr/bin/env python3
"""Top auteurs sur une période + leurs catégories favorites."""

from __future__ import annotations

import os
from elasticsearch import Elasticsearch
from neo4j import GraphDatabase

ES = Elasticsearch(os.environ["ES_URL"], basic_auth=(os.environ["ES_USER"], os.environ["ES_PASSWORD"]))
PILOTE = GraphDatabase.driver(os.environ["NEO4J_URI"], auth=(os.environ["NEO4J_USER"], os.environ["NEO4J_PASSWORD"]))


def top_auteurs_es(du: str, au: str, taille: int = 5) -> list[tuple[str, int]]:
    """Agrégation terms sur authors.raw filtrée par date."""
    rep = ES.search(
        index="news",
        size=0,
        query={"range": {"date": {"gte": du, "lte": au}}},
        aggs={
            "auteurs": {
                "terms": {"field": "authors.raw", "size": taille + 1, "exclude": ["", "Reuters"]}
            }
        },
    )
    seaux = rep["aggregations"]["auteurs"]["buckets"]
    return [(s["key"], s["doc_count"]) for s in seaux[:taille]]


def categories_neo4j(nom: str, limite: int = 3) -> list[tuple[str, int]]:
    """Pour un auteur donné, ses catégories les plus fréquentes."""
    cypher = """
        MATCH (au:Auteur {nom: $nom})<-[:ECRIT_PAR]-(a:Article)-[:PUBLIE_DANS]->(c:Categorie)
        RETURN c.nom AS categorie, count(a) AS articles
        ORDER BY articles DESC LIMIT $limite
    """
    with PILOTE.session() as session:
        return [(r["categorie"], r["articles"]) for r in session.run(cypher, nom=nom, limite=limite)]


def main() -> None:
    du, au = "2017-07-01", "2017-12-31"
    print(f"Top 5 auteurs entre {du} et {au} (hors Reuters)\n")
    for auteur, articles in top_auteurs_es(du, au):
        cats = categories_neo4j(auteur)
        etiquettes = ", ".join(f"{c} ({n})" for c, n in cats)
        print(f"  {articles:>4}  {auteur:<25}  {etiquettes}")
    PILOTE.close()


if __name__ == "__main__":
    main()

Run it:

./lab.sh python top_auteurs.py

Expected output (the counts for the second half of 2017 may differ slightly depending on the corpus version):

Top 5 auteurs entre 2017-07-01 et 2017-12-31 (hors Reuters)

   312  Lee Moran                  COMEDY (779), WEIRD NEWS (391), ENTERTAINMENT (387)
   246  Ed Mazza                   POLITICS (546), ENTERTAINMENT (240), COMEDY (155)
   198  Ron Dicker                 SPORTS (612), COMEDY (355), ENTERTAINMENT (198)
   187  Cole Delbyck               ENTERTAINMENT (886), STYLE & BEAUTY (76), POLITICS (54)
   142  Nina Golgowski             POLITICS (298), U.S. NEWS (188), CRIME (161)

The numbers on the right in parentheses come from the graph: they count all the author's articles, not only those from the second half of 2017. Depending on the business need, you can filter on the Cypher side by adding a clause WHERE a.date >= date($du).

Why separate the two engines?

Elasticsearch excels at filtering and aggregating over millions of documents; Neo4j excels at walking relationships. This division is stable: when a question mixes the two, solving it in two steps is almost always more readable and faster than piling all the logic into a single engine.

Try it 1 — Convert a Kibana query to Python

Take the following query, written in Query DSL in Kibana Dev Tools:

GET news/_search
{
  "size": 3,
  "query": {
    "bool": {
      "must":   [ { "match": { "headline": "election" } } ],
      "filter": [ { "range": { "date": { "gte": "2016-11-01", "lte": "2016-11-30" } } } ]
    }
  },
  "_source": ["headline", "date"]
}

Rewrite it in a python/election_novembre.py script, run it with ./lab.sh python election_novembre.py and display the three titles.

Solution

import os
from elasticsearch import Elasticsearch

es = Elasticsearch(os.environ["ES_URL"], basic_auth=(os.environ["ES_USER"], os.environ["ES_PASSWORD"]))

rep = es.search(
    index="news",
    size=3,
    query={
        "bool": {
            "must":   [{"match": {"headline": "election"}}],
            "filter": [{"range": {"date": {"gte": "2016-11-01", "lte": "2016-11-30"}}}],
        }
    },
    source=["headline", "date"],
)
for h in rep["hits"]["hits"]:
    print(h["_source"]["date"], h["_source"]["headline"])

The order of the bool in Python is the same as the order in JSON: the client just serializes. The only change: _source becomes the argument source (the _ prefix is reserved for server-side metadata).

Try it 2 — Recommendation with execute_read

Write python/reco_auteur.py, which takes an author name as argument and displays the three most recent articles by authors sharing at least two categories with them. Use session.execute_read.

Solution

import os, sys
from neo4j import GraphDatabase

nom = " ".join(sys.argv[1:]) or "Lee Moran"

pilote = GraphDatabase.driver(os.environ["NEO4J_URI"], auth=(os.environ["NEO4J_USER"], os.environ["NEO4J_PASSWORD"]))

def voisins(tx, nom_auteur, limite=3):
    cypher = """
        MATCH (moi:Auteur {nom: $nom})<-[:ECRIT_PAR]-(:Article)-[:PUBLIE_DANS]->(c:Categorie)
              <-[:PUBLIE_DANS]-(:Article)-[:ECRIT_PAR]->(voisin:Auteur)
        WHERE voisin <> moi
        WITH voisin, count(DISTINCT c) AS communes
        WHERE communes >= 2
        MATCH (voisin)<-[:ECRIT_PAR]-(a:Article)
        RETURN voisin.nom AS auteur, a.titre AS titre, a.date AS date
        ORDER BY a.date DESC LIMIT $limite
    """
    return [dict(r) for r in tx.run(cypher, nom=nom_auteur, limite=limite)]

with pilote.session() as session:
    for r in session.execute_read(voisins, nom):
        print(f"{r['date']}  [{r['auteur']}]  {r['titre']}")
pilote.close()

The double MATCH traverses the :PUBLIE_DANS relationship twice to find the authors who share categories with the requested person. execute_read retries automatically if Neo4j is slow to answer.

Try it 3 — Index a mini-corpus with helpers.bulk

Write python/ajouter_notes.py, which inserts into the news index five fictional articles corresponding to Veille internal notes (category INTERNE, author Inès Bouraoui), then check with GET news/_count that the index has grown by five.

Solution

import os
from elasticsearch import Elasticsearch, helpers

es = Elasticsearch(os.environ["ES_URL"], basic_auth=(os.environ["ES_USER"], os.environ["ES_PASSWORD"]))

sujets = [
    "Point hebdomadaire produit — semaine 36",
    "Ateliers Kibana : nouveaux tableaux pour Léa",
    "Revue de sécurité mensuelle",
    "Feuille de route Q4 : reco temps réel",
    "Onboarding Sami : bilan à 30 jours",
]
actions = [
    {"_index": "news", "_id": f"veille-{i}", "_source": {
        "headline": s,
        "short_description": f"Note interne du {i} septembre 2026.",
        "category": "INTERNE",
        "authors": "Inès Bouraoui",
        "date": "2026-09-09",
    }}
    for i, s in enumerate(sujets, start=1)
]
ok, erreurs = helpers.bulk(es, actions)
print(f"{ok} documents indexés")
es.indices.refresh(index="news")
print("Total news :", es.count(index="news")["count"])

Expected output (the total depends on your state; on a freshly imported corpus, it goes from 200,853 to 200,858):

5 documents indexés
Total news : 200858

The explicit refresh guarantees that the immediate count sees the documents; in production, the one-second refresh interval is enough.

Key takeaways

• The veille-python container has the elasticsearch 9 and neo4j 5 clients preinstalled; your scripts live in the kit's python/ folder and run with ./lab.sh python <script>.py.
• All the variables (ES_URL, ES_USER, ES_PASSWORD, NEO4J_URI, NEO4J_USER, NEO4J_PASSWORD) are already injected; never hardcode a password.
• On the Elasticsearch side, the key gestures are es.info(), es.search(index=, query=, size=, source=), es.get, es.index, es.update, and helpers.bulk for a massive import.
• On the Neo4j side, a driver (GraphDatabase.driver), a session (with pilote.session()), a query (session.run or session.execute_read / execute_write), and always named parameters $x.
• Three exceptions to handle cleanly: AuthenticationException (401 Elasticsearch), AuthError (Neo4j), ServiceUnavailable (Neo4j not ready).
• The pairing of the two engines, illustrated by veille.py, fits into two functions: Elasticsearch finds, Neo4j widens.

Troubleshooting

• elasticsearch.AuthenticationException: 401 security_exception → password changed after the volume was created. ./lab.sh reset then ./lab.sh up.
• neo4j.exceptions.ServiceUnavailable → Neo4j is not ready or the driver points to localhost instead of neo4j. Check ./lab.sh status and confirm that NEO4J_URI equals bolt://neo4j:7687 inside the container.
• ImportError: No module named elasticsearch → you are running the script on your host machine instead of the container. Always use ./lab.sh python <script>.py.
• ModuleNotFoundError: No module named 'elasticsearch.helpers' while the import passes → conflict with an old Python package installed in a venv on your machine. The container does not have this problem: go through ./lab.sh python.

Further reading

• Elasticsearch Python client — Getting started
• Elasticsearch Python client — Helpers (bulk)
• Neo4j Python Driver — Manual
• Neo4j Python Driver — Sessions and transactions