小编点评

## Summary of the provided content: This content describes a simple web application built using Neo4j and Flask microframework. **Key points:** * **Purpose:** Provides a list search, detailed view, and graph visualization for a movie dataset. * **Frameworks:** Flask (Web framework) and FastAPI (async framework). * **Database:** Neo4j, a graph database. * **Connectivity:** Uses Neo4j Python Driver for Cypher for data access. * **Deployment:** Can be run in both sync and async mode. **Key functionalities:** * **Search:** Allows users to search for movies by name. * **Detail view:** Shows detailed information about a movie, including id, name, generation, and number of votes. * **Visualization:** Uses Graphviz to create visual representations of the movie network. **Additional details:** * The application uses Neo4j Driver Python for Cypher for data access. * The application supports both sync and async deployment. * The application uses Flask for the web framework and FastAPI for the async framework. * The application uses the `literal_string` type for search queries to prevent cypher injection. * The application logs information for monitoring and debugging purposes. **Overall, this application provides a complete solution for movie information and visualization using Neo4j.**

正文

数据基于： 知识图谱（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
py2neo 目前不支持 neo4j 5.X 所在使用 Neo4j Driver

Python连接Neo4j工具比较 Neo4j Driver、py2neo

依赖文件

基础依赖：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=陈长兴

源码地址：https://gitee.com/VipSoft/VipNeo4j/tree/master/Python
参考：
https://github.com/neo4j-examples/movies-python-bolt
https://neo4j.com/developer/python/