今天突然想起上次知识图谱系列埋了一个坑（【知识图谱】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 Neo4jGraph

import 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: str
    skills: List[str]


class JobPositionModel(BaseModel):
    title: str
    required_skills: List[str]


# Add job seeker
@app.post("/add_job_seeker")
async def add_job_seeker(request: Request):
    try:
        # Parse JSON data from request body
        data = 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 fallback
    job_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 body
        data = 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 fallback
    job_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 uvicorn

    uvicorn.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等工具测试

运行结果，由于我执行过，数据看着比较混乱

计划下一篇更新 neo4j图数据库结合大模型应用的例子

原文链接：【知识图谱】3、Python操作neo4j示例