数据基于： 知识图谱（Knowledge Graph）- Neo4j 5.10.0 使用 - CQL - 太极拳传承谱系表

这是一个非常简单的web应用程序，它使用我们的Movie图形数据集来提供列表搜索、详细视图和图形可视化。

我们提供了两种不同的方式来运行应用程序：同步和异步(使用asyncio)。

Web framework:

• sync: Flask (Micro-Webframework)
• async: FastAPI (Micro-Webframework)

前端:

• jquery
• bootstrap
• d3.js

Neo4j 数据连接: Neo4j Python Driver for Cypher Docs

依赖文件

基础依赖：requirements.txt

# common requirements for sync and async example
neo4j==5.10.0
typing_extensions==4.7.1

同步依赖：requirements-sync.txt

# common requirements
-r requirements.txt

# sync web framework
Flask==2.3.2

同步依赖：requirements-async.txt

# common requirements
-r requirements.txt

# async web framework
uvicorn==0.23.2
fastapi==0.101.1

安装依赖

同步依赖

D:\OpenSource\Neo4j\Python\dependencies> pip install -r requirements-sync.txt -i https://pypi.tuna.tsinghua.edu.cn/simple

异步依赖

D:\OpenSource\Neo4j\Python\dependencies> pip install -r requirements-async.txt -i https://pypi.tuna.tsinghua.edu.cn/simple

代码

#!/usr/bin/env python
import logging
import os
from json import dumps
from textwrap import dedent
from typing import cast

import neo4j
from flask import Flask, Response, request
from neo4j import GraphDatabase, basic_auth
from typing_extensions import LiteralString #这边红的波浪线，不用管

app = Flask(__name__, static_url_path="/static/")

# 获取环境变量值，如果没有就返回默认值
url = os.getenv("NEO4J_URI", "neo4j://172.16.3.64:7687")
username = os.getenv("NEO4J_USER", "neo4j")
password = os.getenv("NEO4J_PASSWORD", "password")
neo4j_version = os.getenv("NEO4J_VERSION", "5")
database = os.getenv("NEO4J_DATABASE", "neo4j")
port = int(os.getenv("PORT", 8080))

driver = GraphDatabase.driver(url, auth=basic_auth(username, password))

@app.route("/")
def get_index():
    return app.send_static_file("index.html")

def query(q: LiteralString) -> LiteralString:
    # this is a safe transform:
    # no way for cypher injection by trimming whitespace
    # hence, we can safely cast to LiteralString
    return cast(LiteralString, dedent(q).strip())

def serialize_person(person):
    return {
        "id": person["id"],
        "name": person["name"],
        "generation": person["generation"],
        "votes": person.get("votes", 0)
    }

@app.route("/search")
def get_search():
    try:
        q = request.args["q"]
    except KeyError:
        return []
    else:
        cql = query("""
                MATCH (p:Person) WHERE p.name CONTAINS $name RETURN p
            """)
        records, _, _ = driver.execute_query(
            cql,
            name=q,  #将参数 q 传给cql 变量
            database_=database,
            routing_="r",
        )

        for record in records:
            # 打印出 record 属性
            logging.info("%s, %s", record["p"]["name"], record["p"]["generation"])

        # desc = [record['p']["name"] for record in records]
        # logging.info(desc)

        return Response(
            # WEB 会显示序列化后的 JSON，汉字没有直观显示，属于正常现象
            dumps([serialize_person(record["p"]) for record in records]),
            mimetype="application/json"
        )

if __name__ == "__main__":
    logging.root.setLevel(logging.INFO)
    logging.info("Starting on port %d, database is at %s", port, url)
    try:
        app.run(port=port)
    finally:
        driver.close()

测试

http://127.0.0.1:8080/search?q=陈长兴