neo4pydantic 0.1.0

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

# For development and testing:
pip install -r requirements.txt

Quick Start

1. Define Your Models

from neo4pydantic.sync import BaseNode, BaseRelationship

class Person(BaseNode):
    __label__ = "Person"
    __unique_fields__ = ["email"]
    __indexed_fields__ = ["name", "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.sync import SyncClient

client = SyncClient(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

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

async def main():
    client = AsyncClient(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())

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