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("RUSHDB_API_KEY")# Create a recorduser=db.records.create(label="USER",data={"name":"John Doe","email":"john@example.com","age":30    })# Find recordsresult=db.records.find({"where": {"age": {"$gte":18},"name": {"$startsWith":"J"}    },"limit":10})# Work with SearchResultprint(f"Found{len(result)} records out of{result.total} total")# Iterate over resultsforrecordinresult:print(f"User:{record.get('name')} (Age:{record.get('age')})")# Check if there are more resultsifresult.has_more:print("There are more records available")# Access specific recordsfirst_user=result[0]ifresultelseNone# 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"})# Run a raw Cypher query (cloud-only)raw=db.query.raw({"query":"MATCH (u:USER) RETURN u LIMIT $limit","params": {"limit":5}})print(raw.get("data"))

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"            }]        }]    }]})

Importing CSV Data with Parse Config

csv_data="""name,email,age\nJohn,john@example.com,30\nJane,jane@example.com,25"""response=db.records.import_csv(label="USER",data=csv_data,options={"returnResult":True,"suggestTypes":True},parse_config={"header":True,"skipEmptyLines":True,"dynamicTyping":True})print(response.get("data"))

SearchResult API

SearchResult Features

Basic Usage

# Perform a searchresult=db.records.find({"where": {"status":"active"},"limit":10,"skip":20})# Check if we have resultsifresult:print(f"Found{len(result)} records")# Access search result informationprint(f"Total matching records:{result.total}")print(f"Has more results:{result.has_more}")print(f"Search query:{result.search_query}")# Get detailed pagination infopage_info=result.get_page_info()print(f"Page info:{page_info}")# Iterate over resultsforrecordinresult:print(f"Record:{record.get('name')}")# List comprehensions worknames= [r.get('name')forrinresult]# Indexing and slicingfirst_record=result[0]ifresultelseNonefirst_five=result[:5]# String representationprint(repr(result))# SearchResult(count=10, total=42)

SearchResult Constructor

def__init__(self,data:List[T],total:Optional[int]=None,search_query:Optional[SearchQuery]=None,):"""    Initialize search result.    Args:        data: List of result items        total: Total number of matching records (defaults to len(data) if not provided)        search_query: The search query used to generate this result (defaults to {})    """

SearchResult Properties

SearchResult Methods

Pagination Example

# Paginated search using skip/limit in querydefpaginate_results(query_base,page_size=10):current_skip=0whileTrue:# Add pagination to queryquery= {**query_base,"limit":page_size,"skip":current_skip}result=db.records.find(query)ifnotresult:breakprint(f"Processing{len(result)} records (skip:{current_skip})")forrecordinresult:process_record(record)ifnotresult.has_more:breakcurrent_skip+=len(result)# Usagepaginate_results({"where": {"category":"electronics"},"orderBy": {"created_at":"desc"}})

RecordSearchResult Type

# Type alias for record search resultsRecordSearchResult=SearchResult[Record]

Improved Record API

Enhanced Data Access

# Create a recorduser=db.records.create("User", {"name":"John Doe","email":"john@example.com","age":30,"department":"Engineering"})# Safe field access with defaultsname=user.get("name")# "John Doe"phone=user.get("phone","Not provided")# "Not provided"# Get clean user data (excludes internal fields like __id, __label)user_data=user.get_data()# Returns: {"name": "John Doe", "email": "john@example.com", "age": 30, "department": "Engineering"}# Get all data including internal fieldsfull_data=user.get_data(exclude_internal=False)# Includes: __id, __label, __proptypes, etc.# Convenient fields propertyfields=user.fields# Same as user.get_data()# Dictionary conversionuser_dict=user.to_dict()# Clean user datafull_dict=user.to_dict(exclude_internal=False)# All data# Direct field accessuser_name=user["name"]# Direct accessuser_id=user["__id"]# Internal field access

Record Existence Checking

# Safe existence checking (no exceptions)ifuser.exists():print("Record is valid and accessible")user.update({"status":"active"})else:print("Record doesn't exist or is not accessible")# Perfect for validation workflowsdefprocess_record_safely(record):ifnotrecord.exists():returnNonereturnrecord.get_data()# Conditional operationsrecords=db.records.find({"where": {"status":"pending"}})forrecordinrecords:ifrecord.exists():record.update({"processed_at":datetime.now()})

String Representations

user=db.records.create("User", {"name":"Alice Johnson"})print(repr(user))# Record(id='abc-123', label='User')print(str(user))# User: Alice Johnson# For records without namesproduct=db.records.create("Product", {"sku":"ABC123"})print(str(product))# Product (product-id-here)

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,search_query:Optional[SearchQuery]=None,record_id:Optional[str]=None,transaction:Optional[Transaction]=None)->RecordSearchResult

# Search for records with complex criteriasearch_query= {"where": {"$and": [            {"age": {"$gte":18}},            {"status":"active"},            {"department":"Engineering"}        ]    },"orderBy": {"created_at":"desc"},"limit":10}result=db.records.find(search_query=search_query)# Work with SearchResultprint(f"Found{len(result)} out of{result.total} total records")# Iterate over resultsforrecordinresult:print(f"Employee:{record.get('name')} -{record.get('department')}")# Check paginationifresult.has_more:print("More results available")# Access specific recordsfirst_employee=result[0]ifresultelseNone# List operationssenior_employees= [rforrinresultifr.get('age',0)>30]

delete()

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

# Delete records matching criteriasearch_query= {"where": {"status":"inactive","lastActive": {"$lt":"2023-01-01"}    }}response=db.records.delete(search_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,data:str,options:Optional[Dict[str,bool]]=None,transaction:Optional[Transaction]=None)->List[Dict[str,Any]]

# Import records from CSVdata="""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",data=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 transactiontransaction=db.transactions.begin()try:# 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    )# Explicitly commit the transaction to make changes permanenttransaction.commit()exceptExceptionase:# Rollback if any error occurstransaction.rollback()raisee# Alternative: Using context managerwithdb.transactions.begin()astransaction:# Perform operations...user=db.records.create("USER",        {"name":"John Doe"},transaction=transaction    )# Must explicitly commit - transactions are NOT automatically committedtransaction.commit()

PropertiesAPI Documentation

Class Definition

classPropertiesAPI(BaseAPI):

Methods

find()

deffind(self,search_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 transaction with explicit committransaction=db.transactions.begin()try:# 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    )# Explicitly commit the transactiontransaction.commit()exceptExceptionase:# Rollback if any error occurstransaction.rollback()raisee# Alternative: Using context manager (auto-rollback on error)withdb.transactions.begin()astransaction:# Perform operationsproperty_to_delete=db.properties.find(        {"where": {"name":"temp_property"}},transaction=transaction    )[0]db.properties.delete(property_id=property_to_delete['id'],transaction=transaction    )# Must explicitly commit - transactions are NOT automatically committedtransaction.commit()

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}")

LabelsAPI Documentation

Class Definition

classLabelsAPI(BaseAPI):

Methods

find()

deffind(self,search_query:Optional[SearchQuery]=None,transaction:Optional[Transaction]=None)->Dict[str,int]

# Get all labels in the databaseall_labels=db.labels.find()print("Available labels:",all_labels)# Output: {'USER': 150, 'DEPARTMENT': 12, 'PROJECT': 45, 'COMPANY': 3}# Search for labels amongst records matching a patternfromrushdb.models.search_queryimportSearchQueryquery=SearchQuery(where={"name": {"$contains":"alice"}})user_labels=db.labels.find(query)print("Labels for records containing 'alice':",user_labels)# Output: {'USER': 2, 'EMPLOYEE': 1}

Complete Usage Example

# Discover all record types in the databaseall_labels=db.labels.find()print(f"Database contains{len(all_labels)} record types:")forlabel,countinall_labels.items():print(f"  -{label}:{count} records")# Find labels for records with specific criteriaquery=SearchQuery(where={"status":"active","created_date": {"$gte":"2023-01-01"}})active_labels=db.labels.find(query)print("Labels for active records:")forlabel,countinactive_labels.items():print(f"  -{label}:{count} active records")# Use with transactiontransaction=db.transactions.begin()try:labels_in_tx=db.labels.find(transaction=transaction)# Process labels...transaction.commit()exceptExceptionase:transaction.rollback()raisee

RelationshipsAPI Documentation

Class Definition

classRelationshipsAPI(BaseAPI):

Methods

find()

asyncdeffind(self,search_query:Optional[SearchQuery]=None,pagination:Optional[PaginationParams]=None,transaction:Optional[Union[Transaction,str]]=None)->List[Relationship]

importasynciofromrushdb.models.search_queryimportSearchQueryasyncdefmain():# Find all relationshipsall_relationships=awaitdb.relationships.find()print(f"Total relationships:{len(all_relationships)}")# Find relationships with paginationpagination= {"limit":50,"skip":0}first_page=awaitdb.relationships.find(pagination=pagination)# Find specific relationship typesquery=SearchQuery(where={"type":"BELONGS_TO"})belongs_to_rels=awaitdb.relationships.find(search_query=query)# Find relationships involving specific recordsuser_query=SearchQuery(where={"$or": [            {"source_id":"user-123"},            {"target_id":"user-123"}        ]    })user_relationships=awaitdb.relationships.find(search_query=user_query)# Run the async functionasyncio.run(main())

PaginationParams

classPaginationParams(TypedDict,total=False):limit:int# Maximum number of relationships to returnskip:int# Number of relationships to skip

Complete Usage Example

importasynciofromrushdb.models.search_queryimportSearchQueryasyncdefexplore_relationships():# Get overview of all relationshipsall_rels=awaitdb.relationships.find()print(f"Database contains{len(all_rels)} relationships")# Paginate through relationshipspage_size=25page=0whileTrue:pagination= {"limit":page_size,"skip":page*page_size}relationships=awaitdb.relationships.find(pagination=pagination)ifnotrelationships:breakprint(f"Page{page+1}:{len(relationships)} relationships")forrelinrelationships:print(f"{rel['source_id']} --[{rel['type']}]-->{rel['target_id']}")page+=1iflen(relationships)<page_size:break# Find relationships by typequery=SearchQuery(where={"type":"WORKS_ON"})work_relationships=awaitdb.relationships.find(search_query=query)print(f"Found{len(work_relationships)} 'WORKS_ON' relationships")# Find relationships within a transactiontransaction=db.transactions.begin()try:tx_rels=awaitdb.relationships.find(transaction=transaction)# Process relationships...transaction.commit()exceptExceptionase:transaction.rollback()raisee# Run the exampleasyncio.run(explore_relationships())

Working with Transactions

importasyncioasyncdeftransaction_example():transaction=db.transactions.begin()try:# Find labels within transactionlabels=db.labels.find(transaction=transaction)# Find relationships within transactionrelationships=awaitdb.relationships.find(transaction=transaction)# Perform operations based on discovered data...# Explicitly commit the transactiontransaction.commit()exceptExceptionase:# Rollback on any errortransaction.rollback()raiseeasyncio.run(transaction_example())