知识图谱（Knowledge Graph）- Neo4j 5.10.0 使用 - Python 操作

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

# common requirements
-r requirements.txt
# sync web framework
Flask==2.3.2

# 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()