知識圖譜（Knowledge Graph）- Neo4j 5.10.0 使用

資料基於： 知識圖譜（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=陳長興