知识图谱（Knowledge Graph）- Neo4j 5.10.0 使用 - Python 操作-阿里云开发者社区

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

2024-08-15 465

版权

本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。

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

数据基于：知识图谱（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()