RushDB is an instant database for modern apps and DS/ML ops built on top of Neo4j.It automates data normalization, manages relationships, and infers data types.

📖 Documentation •🌐 Website •☁️ Cloud Platform

Installation

pip install rushdb

Quick Start

fromrushdbimportRushDB# Initialize the clientdb=RushDB("YOUR_API_TOKEN")# Create a recorduser=db.records.create(label="USER",data={"name":"John Doe","email":"john@example.com","age":30    })# Find recordsresults=db.records.find({"where": {"age": {"$gte":18},"name": {"$startsWith":"J"}    },"limit":10})# Create relationshipscompany=db.records.create(label="COMPANY",data={"name":"Acme Inc."})# Attach records with a relationshipuser.attach(target=company,options={"type":"WORKS_AT","direction":"out"})

Pushing Nested JSON

# Push nested JSON with automatic relationship creationdb.records.create_many("COMPANY", {"name":"Google LLC","rating":4.9,"DEPARTMENT": [{"name":"Research & Development","PROJECT": [{"name":"Bard AI","EMPLOYEE": [{"name":"Jeff Dean","position":"Head of AI Research"            }]        }]    }]})

Complete Documentation

Support

set()

defset(self,record_id:str,data:Dict[str,Any],transaction:Optional[Transaction]=None)->Dict[str,str]

# Update entire record datanew_data= {"name":"Updated Company Name","rating":5.0}response=db.records.set(record_id="record-123",data=new_data)

update()

defupdate(self,record_id:str,data:Dict[str,Any],transaction:Optional[Transaction]=None)->Dict[str,str]

# Update specific fieldsupdates= {"rating":4.8,"status":"active"}response=db.records.update(record_id="record-123",data=updates)

find()

deffind(self,query:Optional[SearchQuery]=None,record_id:Optional[str]=None,transaction:Optional[Transaction]=None)->List[Record]

# Search for records with complex criteriaquery= {"where": {"$and": [            {"age": {"$gte":18}},            {"status":"active"},            {"department":"Engineering"}        ]    },"orderBy": {"created_at":"desc"},"limit":10}records=db.records.find(query=query)

delete()

defdelete(self,query:SearchQuery,transaction:Optional[Transaction]=None)->Dict[str,str]

# Delete records matching criteriaquery= {"where": {"status":"inactive","lastActive": {"$lt":"2023-01-01"}    }}response=db.records.delete(query)

delete_by_id()

defdelete_by_id(self,id_or_ids:Union[str,List[str]],transaction:Optional[Transaction]=None)->Dict[str,str]

# Delete single recordresponse=db.records.delete_by_id("record-123")# Delete multiple recordsresponse=db.records.delete_by_id(["record-123","record-456","record-789"])

attach()

defattach(self,source:Union[str,Dict[str,Any]],target:Union[str,List[str],Dict[str,Any],List[Dict[str,Any]],Record,List[Record]],options:Optional[RelationshipOptions]=None,transaction:Optional[Transaction]=None)->Dict[str,str]

# Create relationship between recordsoptions=RelationshipOptions(type="HAS_EMPLOYEE",direction="out")response=db.records.attach(source="company-123",target=["employee-456","employee-789"],options=options)

detach()

defdetach(self,source:Union[str,Dict[str,Any]],target:Union[str,List[str],Dict[str,Any],List[Dict[str,Any]],Record,List[Record]],options:Optional[RelationshipDetachOptions]=None,transaction:Optional[Transaction]=None)->Dict[str,str]

# Remove relationships between recordsoptions=RelationshipDetachOptions(typeOrTypes=["HAS_EMPLOYEE","MANAGES"],direction="out")response=db.records.detach(source="company-123",target="employee-456",options=options)

import_csv()

defimport_csv(self,label:str,csv_data:Union[str,bytes],options:Optional[Dict[str,bool]]=None,transaction:Optional[Transaction]=None)->List[Dict[str,Any]]

# Import records from CSVcsv_data="""name,age,department,roleJohn Doe,30,Engineering,Senior EngineerJane Smith,28,Product,Product ManagerBob Wilson,35,Engineering,Tech Lead"""records=db.records.import_csv(label="EMPLOYEE",csv_data=csv_data,options={"returnResult":True,"suggestTypes":True})

Record Class Documentation

Class Definition

classRecord:def__init__(self,client:"RushDB",data:Union[Dict[str,Any],None]=None)

Properties

id

record=db.records.create("USER", {"name":"John"})print(record.id)# e.g., "1234abcd-5678-..."

proptypes

record=db.records.create("USER", {"name":"John","age":25})print(record.proptypes)# Returns property type definitions

label

record=db.records.create("USER", {"name":"John"})print(record.label)# "USER"

timestamp

record=db.records.create("USER", {"name":"John"})print(record.timestamp)# Unix timestamp in milliseconds

date

record=db.records.create("USER", {"name":"John"})print(record.date)# datetime object

Methods

set()

defset(self,data:Dict[str,Any],transaction:Optional[Transaction]=None)->Dict[str,str]

record=db.records.create("USER", {"name":"John"})response=record.set({"name":"John Doe","email":"john@example.com","age":30})

update()

defupdate(self,data:Dict[str,Any],transaction:Optional[Transaction]=None)->Dict[str,str]

record=db.records.create("USER", {"name":"John","email":"john@example.com"})response=record.update({"email":"john.doe@example.com"})

attach()

defattach(self,target:Union[str,List[str],Dict[str,Any],List[Dict[str,Any]],"Record",List["Record"]],options:Optional[RelationshipOptions]=None,transaction:Optional[Transaction]=None)->Dict[str,str]

# Create two recordsuser=db.records.create("USER", {"name":"John"})group=db.records.create("GROUP", {"name":"Admins"})# Attach user to groupresponse=user.attach(target=group,options=RelationshipOptions(type="BELONGS_TO",direction="out"    ))

detach()

defdetach(self,target:Union[str,List[str],Dict[str,Any],List[Dict[str,Any]],"Record",List["Record"]],options:Optional[RelationshipDetachOptions]=None,transaction:Optional[Transaction]=None)->Dict[str,str]

# Detach user from groupresponse=user.detach(target=group,options=RelationshipDetachOptions(typeOrTypes="BELONGS_TO",direction="out"    ))

delete()

defdelete(self,transaction:Optional[Transaction]=None)->Dict[str,str]

user=db.records.create("USER", {"name":"John"})response=user.delete()

Complete Usage Example

# Create a new recorduser=db.records.create("USER", {"name":"John Doe","email":"john@example.com","age":30})# Access propertiesprint(f"Record ID:{user.id}")print(f"Label:{user.label}")print(f"Created at:{user.date}")# Update record datauser.update({"age":31,"title":"Senior Developer"})# Create related recordsdepartment=db.records.create("DEPARTMENT", {"name":"Engineering"})project=db.records.create("PROJECT", {"name":"Secret Project"})# Create relationshipsuser.attach(target=department,options=RelationshipOptions(type="BELONGS_TO",direction="out"    ))user.attach(target=project,options=RelationshipOptions(type="WORKS_ON",direction="out"    ))# Remove relationshipuser.detach(target=project,options=RelationshipDetachOptions(typeOrTypes="WORKS_ON",direction="out"    ))# Delete recorduser.delete()

Working with Transactions

# Start a transactionwithdb.transactions.begin()astransaction:# Create useruser=db.records.create("USER",        {"name":"John Doe"},transaction=transaction    )# Update useruser.update(        {"status":"active"},transaction=transaction    )# Create and attach departmentdept=db.records.create("DEPARTMENT",        {"name":"Engineering"},transaction=transaction    )user.attach(target=dept,options=RelationshipOptions(type="BELONGS_TO"),transaction=transaction    )# Transaction will automatically commit if no errors occur# If an error occurs, it will automatically rollback

PropertiesAPI Documentation

Class Definition

classPropertiesAPI(BaseAPI):

Methods

find()

deffind(self,query:Optional[SearchQuery]=None,transaction:Optional[Transaction]=None)->List[Property]

# Find all propertiesproperties=db.properties.find()# Find properties with specific criteriaquery= {"where": {"name": {"$startsWith":"user_"},# Properties starting with 'user_'"type":"string"# Only string type properties    },"limit":10# Limit to 10 results}filtered_properties=db.properties.find(query)

find_by_id()

deffind_by_id(self,property_id:str,transaction:Optional[Transaction]=None)->Property

# Retrieve a specific property by IDproperty_details=db.properties.find_by_id("prop_123456")

delete()

defdelete(self,property_id:str,transaction:Optional[Transaction]=None)->None

# Delete a propertydb.properties.delete("prop_123456")

values()

defvalues(self,property_id:str,sort:Optional[Literal["asc","desc"]]=None,skip:Optional[int]=None,limit:Optional[int]=None,transaction:Optional[Transaction]=None)->PropertyValuesData

# Get property valuesvalues_data=db.properties.values(property_id="prop_age",sort="desc",# Sort values in descending orderskip=0,# Start from the first valuelimit=100# Return up to 100 values)# Access valuesprint(values_data.get('values', []))# List of property valuesprint(values_data.get('min'))# Minimum value (for numeric properties)print(values_data.get('max'))# Maximum value (for numeric properties)

Comprehensive Usage Example

# Find all propertiesall_properties=db.properties.find()forpropinall_properties:print(f"Property ID:{prop['id']}")print(f"Name:{prop['name']}")print(f"Type:{prop['type']}")print(f"Metadata:{prop.get('metadata','No metadata')}")print("---")# Detailed property searchquery= {"where": {"type":"number",# Only numeric properties"name": {"$contains":"score"}# Properties with 'score' in name    },"limit":5# Limit to 5 results}numeric_score_properties=db.properties.find(query)# Get values for a specific propertyifnumeric_score_properties:first_prop=numeric_score_properties[0]prop_values=db.properties.values(property_id=first_prop['id'],sort="desc",limit=50    )print(f"Values for{first_prop['name']}:")print(f"Min:{prop_values.get('min')}")print(f"Max:{prop_values.get('max')}")# Detailed property examinationdetailed_prop=db.properties.find_by_id(first_prop['id'])print("Detailed Property Info:",detailed_prop)

Property Types and Structures

Property Structure Example

property= {"id":"prop_unique_id","name":"user_score","type":"number","metadata":Optional[str]# Optional additional information}property_with_value= {"id":"prop_unique_id","name":"user_score","type":"number","value":95.5# Actual property value}

Transactions

# Using a transactionwithdb.transactions.begin()astransaction:# Perform multiple property-related operationsproperty_to_delete=db.properties.find(        {"where": {"name":"temp_property"}},transaction=transaction    )[0]db.properties.delete(property_id=property_to_delete['id'],transaction=transaction    )# Transaction will automatically commit if no errors occur

Error Handling

try:# Attempt to find or delete a propertyproperty_details=db.properties.find_by_id("non_existent_prop")exceptRushDBErrorase:print(f"Error:{e}")print(f"Error Details:{e.details}")