RecordsAPI
The RecordsAPI
class provides methods for managing records in RushDB. It handles record creation, updates, deletion, searching, and relationship management.
Class Definition
class RecordsAPI(BaseAPI):
Methods
create()
Creates a new record in RushDB.
Signature:
def create(
self,
label: str,
data: Dict[str, Any],
options: Optional[Dict[str, bool]] = None,
transaction: Optional[Transaction] = None
) -> Record
Arguments:
label
(str): Label for the recorddata
(Dict[str, Any]): Record dataoptions
(Optional[Dict[str, bool]]): Optional parsing and response optionsreturnResult
(bool): Whether to return the created recordsuggestTypes
(bool): Whether to suggest property types
transaction
(Optional[Transaction]): Optional transaction object
Returns:
Record
: Created record object
Example:
# Create a new company record
data = {
"name": "Google LLC",
"address": "1600 Amphitheatre Parkway",
"foundedAt": "1998-09-04T00:00:00.000Z",
"rating": 4.9
}
record = client.records.create(
label="COMPANY",
data=data,
options={"returnResult": True, "suggestTypes": True}
)
create_many()
Creates multiple records in a single operation.
Signature:
def create_many(
self,
label: str,
data: Union[Dict[str, Any], List[Dict[str, Any]]],
options: Optional[Dict[str, bool]] = None,
transaction: Optional[Transaction] = None
) -> List[Record]
Arguments:
label
(str): Label for all recordsdata
(Union[Dict[str, Any], List[Dict[str, Any]]]): List or Dict of record dataoptions
(Optional[Dict[str, bool]]): Optional parsing and response optionstransaction
(Optional[Transaction]): Optional transaction object
Returns:
List[Record]
: List of created record objects
Example:
# Create multiple company records
data = [
{
"name": "Apple Inc",
"address": "One Apple Park Way",
"foundedAt": "1976-04-01T00:00:00.000Z",
"rating": 4.8
},
{
"name": "Microsoft Corporation",
"address": "One Microsoft Way",
"foundedAt": "1975-04-04T00:00:00.000Z",
"rating": 4.7
}
]
records = client.records.create_many(
label="COMPANY",
data=data,
options={"returnResult": True, "suggestTypes": True}
)
set()
Updates a record by ID, replacing all data.
Signature:
def set(
self,
record_id: str,
data: Dict[str, Any],
transaction: Optional[Transaction] = None
) -> Dict[str, str]
Arguments:
record_id
(str): ID of the record to updatedata
(Dict[str, Any]): New record datatransaction
(Optional[Transaction]): Optional transaction object
Returns:
Dict[str, str]
: Response data
Example:
# Update entire record data
new_data = {
"name": "Updated Company Name",
"rating": 5.0
}
response = client.records.set(
record_id="record-123",
data=new_data
)
update()
Updates specific fields of a record by ID.
Signature:
def update(
self,
record_id: str,
data: Dict[str, Any],
transaction: Optional[Transaction] = None
) -> Dict[str, str]
Arguments:
record_id
(str): ID of the record to updatedata
(Dict[str, Any]): Partial record data to updatetransaction
(Optional[Transaction]): Optional transaction object
Returns:
Dict[str, str]
: Response data
Example:
# Update specific fields
updates = {
"rating": 4.8,
"status": "active"
}
response = client.records.update(
record_id="record-123",
data=updates
)
find()
Searches for records matching specified criteria.
Signature:
def find(
self,
query: Optional[SearchQuery] = None,
record_id: Optional[str] = None,
transaction: Optional[Transaction] = None
) -> List[Record]
Arguments:
query
(Optional[SearchQuery]): Search query parametersrecord_id
(Optional[str]): Optional record ID to search fromtransaction
(Optional[Transaction]): Optional transaction object
Returns:
List[Record]
: List of matching records
Example:
# Search for records with complex criteria
query = {
"where": {
"$and": [
{"age": {"$gte": 18}},
{"status": "active"},
{"department": "Engineering"}
]
},
"orderBy": {"created_at": "desc"},
"limit": 10
}
records = client.records.find(query=query)
delete()
Deletes records matching a query.
Signature:
def delete(
self,
query: SearchQuery,
transaction: Optional[Transaction] = None
) -> Dict[str, str]
Arguments:
query
(SearchQuery): Query to match records for deletiontransaction
(Optional[Transaction]): Optional transaction object
Returns:
Dict[str, str]
: Response data
Example:
# Delete records matching criteria
query = {
"where": {
"status": "inactive",
"lastActive": {"$lt": "2023-01-01"}
}
}
response = client.records.delete(query)
delete_by_id()
Deletes one or more records by ID.
Signature:
def delete_by_id(
self,
id_or_ids: Union[str, List[str]],
transaction: Optional[Transaction] = None
) -> Dict[str, str]
Arguments:
id_or_ids
(Union[str, List[str]]): Single ID or list of IDs to deletetransaction
(Optional[Transaction]): Optional transaction object
Returns:
Dict[str, str]
: Response data
Example:
# Delete single record
response = client.records.delete_by_id("record-123")
# Delete multiple records
response = client.records.delete_by_id([
"record-123",
"record-456",
"record-789"
])
attach()
Creates relationships between records.
Signature:
def attach(
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]
Arguments:
source
(Union[str, Dict[str, Any]]): Source record ID or datatarget
(Union[str, List[str], Dict[str, Any], List[Dict[str, Any]], Record, List[Record]]): Target record(s)options
(Optional[RelationshipOptions]): Relationship optionsdirection
(Optional[Literal["in", "out"]]): Relationship directiontype
(Optional[str]): Relationship type
transaction
(Optional[Transaction]): Optional transaction object
Returns:
Dict[str, str]
: Response data
Example:
# Create relationship between records
options = RelationshipOptions(
type="HAS_EMPLOYEE",
direction="out"
)
response = client.records.attach(
source="company-123",
target=["employee-456", "employee-789"],
options=options
)
detach()
Removes relationships between records.
Signature:
def detach(
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]
Arguments:
source
(Union[str, Dict[str, Any]]): Source record ID or datatarget
(Union[str, List[str], Dict[str, Any], List[Dict[str, Any]], Record, List[Record]]): Target record(s)options
(Optional[RelationshipDetachOptions]): Detach optionsdirection
(Optional[Literal["in", "out"]]): Relationship directiontypeOrTypes
(Optional[Union[str, List[str]]]): Relationship type(s)
transaction
(Optional[Transaction]): Optional transaction object
Returns:
Dict[str, str]
: Response data
Example:
# Remove relationships between records
options = RelationshipDetachOptions(
typeOrTypes=["HAS_EMPLOYEE", "MANAGES"],
direction="out"
)
response = client.records.detach(
source="company-123",
target="employee-456",
options=options
)
import_csv()
Imports records from CSV data.
Signature:
def import_csv(
self,
label: str,
csv_data: Union[str, bytes],
options: Optional[Dict[str, bool]] = None,
transaction: Optional[Transaction] = None
) -> List[Dict[str, Any]]
Arguments:
label
(str): Label for imported recordscsv_data
(Union[str, bytes]): CSV data to importoptions
(Optional[Dict[str, bool]]): Import optionstransaction
(Optional[Transaction]): Optional transaction object
Returns:
List[Dict[str, Any]]
: Imported records data
Example:
# Import records from CSV
csv_data = """name,age,department,role
John Doe,30,Engineering,Senior Engineer
Jane Smith,28,Product,Product Manager
Bob Wilson,35,Engineering,Tech Lead"""
records = client.records.import_csv(
label="EMPLOYEE",
csv_data=csv_data,
options={"returnResult": True, "suggestTypes": True}
)