neo4pydantic 0.1.1

A Pydantic-based Neo4j ORM with async/sync support

neo4pydantic

A Pydantic-based Neo4j ORM with async/sync support.

Overview

neo4pydantic provides an easy way to define Neo4j nodes and relationships as Pydantic models, and interact with a Neo4j database using both synchronous and asynchronous clients. It supports automatic type conversion, unique and indexed fields, and convenient CRUD operations.

Features

• Define Neo4j nodes and relationships as Pydantic models
• Sync and async client support
• Automatic type conversion for Neo4j temporal types
• Unique and indexed field support for efficient queries
• Simple CRUD operations for nodes and relationships

Installation

pip install neo4pydantic

Quick Start

1. Define Your Models

from neo4pydantic import BaseNode, BaseRelationship

class Person(BaseNode):
    __label__ = "Person"
    __unique_fields__ = ["email"]

    name: str
    email: str
    age: int | None = None
    city: str | None = None

class Company(BaseNode):
    __label__ = "Company"
    __unique_fields__ = ["name"]

    name: str
    industry: str | None = None
    founded_year: int | None = None

class WorksAt(BaseRelationship):
    __type__ = "WORKS_AT"
    position: str
    start_date: str | None = None
    salary: int | None = None

2. Synchronous Usage

from neo4pydantic import Neo4jClient

client = Neo4jClient(uri="bolt://localhost:7687", user="neo4j", password="your_password")
with client.session() as session:
  person = Person(name="John Doe", email="[e-mail]", age=30, city="New York").save(session)
  company = Company(name="Tech Corp", industry="Technology", founded_year=2010).save(session)
  relationship = WorksAt(position="Software Engineer", start_date="2023-01-15", salary=75000)
  relationship.save(session, person, company)
  # Query
  people = Person.find_by(session, city="New York")

3. Asynchronous Usage

Note that all asynchronous usage is the same as synchronous function

import asyncio
from neo4pydantic.async_ import AsyncNeo4jClient, BaseNode, BaseRelationship


async def main():
  client = AsyncNeo4jClient(uri="bolt://localhost:7687", user="neo4j", password="your_password")
  async with client.session() as session:
    person = Person(name="Jane Doe", email="[e-mail]", age=28, city="San Francisco")
    person = await person.save(session)
    company = Company(name="Startup Inc", industry="Technology", founded_year=2020)
    company = await company.save(session)
    relationship = WorksAt(position="Senior Developer", start_date="2023-03-01", salary=90000)
    await relationship.save(session, person, company)
    # Query
    people = await Person.find_by(session, city="San Francisco")


asyncio.run(main())

Querying and Advanced Usage

Find Multiple Nodes (find_by)

# Find all people in New York
people = Person.find_by(session, city="New York")
for person in people:
    print(person)

Find a Single Node (find_one_by)

# Find a person by email
john = Person.find_one_by(session, email="[e-mail]")
print(john)

Delete A Node (find_one_by)

# Delete a person
john = Person(id=10, mail="[e-mail]").delete(session)

Automatically create indexes when startup

from neo4pydantic import IndexManager, Neo4jClient

client = Neo4jClient(uri="bolt://localhost:7687", user="neo4j", password="your_password")

def init_indexes():
    # Import your custom model here so that the IndexManger can acknowledge your model before create the indexes
    from your_custom_model import Person, Company, WorkFor
    
    client.connect()
    created_indexes = IndexManager(client.get_driver()).sync_indexes()
    print(created_indexes)

---

Create Relationship with Custom Parameters (save_with_custom_params)

relationship = WorksAt(position="Junior Developer", start_date="2022-03-01", salary=90000)
relationship.save_with_custom_params(
    session,
    from_node_label=person.get_label(),
    to_node_label=company.get_label(),
    from_node_params={"email": person.email},
    to_node_params={"name": company.name},
)

---

Directory Structure

• neo4pydantic/
  • sync/ - Synchronous client and base classes
  • async_/ - Asynchronous client and base classes
  • core/ - Core logic and base models
  • utils/ - Type conversion utilities
  • examples/ - Example scripts for sync and async usage
• tests/ - Unit tests

Requirements

• Python 3.8+
• Neo4j 5.x
• Pydantic 2.x

License

MIT License © 2025 Edward Chu