【知识图谱】3、Python操作图数据库neo4j示例

本文主要是介绍【知识图谱】3、Python操作图数据库neo4j示例，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

今天突然想起上次知识图谱系列埋了一个坑（【知识图谱】1、Neo4j环境搭建入门指南:从零开始玩转图数据库），说后续写一篇关于Python操作neo4j的示例。趁着周六有充足时间，这里写个demo补上。

本文demo还是以面试的求职者、岗位要求技能为例。建2个实例对象

1、求职者具备的技能

2、岗位要求的技能

本文默认已经安装好 neo4j desktop数据库，直接先上代码

from fastapi import FastAPI, HTTPException, Request
from pydantic import BaseModel
from langchain_community.graphs import Neo4jGraphimport asyncio
from typing import List
import json# Initialize FastAPI
app = FastAPI()# Initialize Neo4j with timeout
try:graph = Neo4jGraph(url="bolt://localhost:7687",username="neo4j",password="password",database="neo4j",timeout=60  # 60 seconds timeout)
except Exception as e:print(f"Failed to connect to Neo4j: {e}")graph = None# Fallback in-memory storage
job_seekers = []
job_positions = []# Define Pydantic models for request bodies
class JobSeekerModel(BaseModel):name: strskills: List[str]class JobPositionModel(BaseModel):title: strrequired_skills: List[str]# Add job seeker
@app.post("/add_job_seeker")
async def add_job_seeker(request: Request):try:# Parse JSON data from request bodydata = await request.json()print(data)job_seeker = JobSeekerModel(**data)except json.JSONDecodeError:raise HTTPException(status_code=400, detail="Invalid JSON data")except ValueError as e:raise HTTPException(status_code=400, detail=str(e))if graph:try:query = ("CREATE (j:JobSeeker {name: $name}) ""WITH j ""UNWIND $skills AS skill ""MERGE (s:Skill {name: skill}) ""CREATE (j)-[:HAS_SKILL]->(s)")await asyncio.wait_for(asyncio.to_thread(graph.query, query, {"name": job_seeker.name, "skills": job_seeker.skills}),timeout=5.0  # 5 seconds timeout)except asyncio.TimeoutError:raise HTTPException(status_code=504, detail="Database operation timed out")except Exception as e:print(f"Neo4j error: {e}")raise HTTPException(status_code=500, detail="Failed to add job seeker to Neo4j")# Always add to in-memory storage as fallbackjob_seekers.append({"name": job_seeker.name, "skills": job_seeker.skills})return {"message": f"Added job seeker {job_seeker.name} with skills {', '.join(job_seeker.skills)}"}# Add job position
@app.post("/add_job_position")
async def add_job_position(request: Request):try:# Parse JSON data from request bodydata = await request.json()print(data)job_position = JobPositionModel(**data)except json.JSONDecodeError:raise HTTPException(status_code=400, detail="Invalid JSON data")except ValueError as e:raise HTTPException(status_code=400, detail=str(e))if graph:try:query = ("CREATE (j:JobPosition {title: $title}) ""WITH j ""UNWIND $required_skills AS skill ""MERGE (s:Skill {name: skill}) ""CREATE (j)-[:REQUIRES_SKILL]->(s)")await asyncio.wait_for(asyncio.to_thread(graph.query, query,{"title": job_position.title, "required_skills": job_position.required_skills}),timeout=5.0  # 5 seconds timeout)except asyncio.TimeoutError:raise HTTPException(status_code=504, detail="Database operation timed out")except Exception as e:print(f"Neo4j error: {e}")raise HTTPException(status_code=500, detail="Failed to add job position to Neo4j")# Always add to in-memory storage as fallbackjob_positions.append({"title": job_position.title, "required_skills": job_position.required_skills})return {"message": f"Added job position {job_position.title} requiring skills {', '.join(job_position.required_skills)}"}# Run the app
if __name__ == "__main__":import uvicornuvicorn.run(app, host="0.0.0.0", port=8001)

还是用AI解释下代码主要功能：

1、初始化FastAPI应用和Neo4j图数据库连接。

2、定义了两个Pydantic模型JobSeekerModel和JobPositionModel用于请求体验证。

3、提供两个POST接口：

/add_job_seeker：添加求职者信息到Neo4j（若连接成功）。

/add_job_position：添加职位信息到Neo4j（若连接成功）。

4、使用异步处理数据库操作，并设置超时时间以确保服务稳定性。

直接运行可能会提示APOC插件未安装，那就先安装plugins， APOC 点击install安装。我这个是安装的截图，当时忘记先截图了。

图片

写接口测试下可以用curl命令

1、添加求职者

curl -X POST http://localhost:8000/api/add_job_seeker
-H “Content-Type: application/json”
-d ‘{“name”: “Alice Smith”, “skills”: [“Python”, “JavaScript”, “Machine Learning”]}’
2、添加职位：

curl -X POST http://localhost:8000/api/add_job_position
-H “Content-Type: application/json”
-d ‘{“title”: “Senior Software Engineer”, “required_skills”: [“Python”, “Docker”, “Kubernetes”]}’

我们也可以用postman等工具测试

在这里插入图片描述