gqlalchemy
gqlalchemy copied to clipboard
`load_node_with_all_properties` fails for model with disk property without unique fields or id
Community report - Discord
When you create a Model with a disk property without unique fields or id it fails to load the node from load_node_with_all_properties
. It does not seem to ignore the disk properties and they get added to the where clause.
Minimum reproducible example:
class TestNode(Node):
#id: int = Field(1, unique=True, db=db)
Table : int = Field(0, db=db)
Table2 : int = Field(0, on_disk=True)
checksum = TestNode(Table2=17235).save(db)
checksum_db = TestNode(Table2=17235).load(db)
print(checksum_db)
Saves to sqlite but fails to load back with: Traceback (most recent call last):
File "/home/ubuntu/mix-alchemy/test.py", line 105, in <module>
checksum_db = TestNode(Table2=17235).load(db)
File "/home/ubuntu/.local/lib/python3.8/site-packages/gqlalchemy/models.py", line 620, in load
node = db.load_node(self)
File "/home/ubuntu/mix-alchemy/test.py", line 89, in load_node
result = self.load_node_with_all_properties(node)
File "/home/ubuntu/.local/lib/python3.8/site-packages/gqlalchemy/vendors/database_client.py", line 224, in load_node_with_all_properties
return self.get_variable_assume_one(results, "node")
Back then, I got around the error by inheriting the Node class and overriding the method
class NodeT(Node):
#patching to fix on on_disk
def _get_cypher_field_assignment_block(self, variable_name: str, operator: str) -> str:
"""Creates a cypher field assignment block joined using the operator
argument.
Example:
self = {"name": "John", "age": 34}
variable_name = "user"
operator = " AND "
returns:
"user.name = 'John' AND user.age = 34"
"""
cypher_fields = []
for field in self.fields:
value = getattr(self, field)
attributes = self.fields[field].field_info.extra
if value is not None and not attributes.get("on_disk", False):
cypher_fields.append(f"{variable_name}.{field} = {self.escape_value(value)}")
return " " + operator.join(cypher_fields) + " "
def loads(self, db: "Database") -> List["Node"]: # noqa F821
node = db.load_node(self)
for field in self.fields:
setattr(self, field, getattr(node, field))
self._id = node._id
return self
Here's the fix - just follows what is used on the other methods:
def _get_cypher_field_assignment_block(self, variable_name: str, operator: str) -> str:
"""Creates a cypher field assignment block joined using the operator
argument.
Example:
self = {"name": "John", "age": 34}
variable_name = "user"
operator = " AND "
returns:
"user.name = 'John' AND user.age = 34"
"""
cypher_fields = []
for field in self.fields:
value = getattr(self, field)
attributes = self.fields[field].field_info.extra
if value is not None and not attributes.get("on_disk", False):
cypher_fields.append(f"{variable_name}.{field} = {self.escape_value(value)}")
return " " + operator.join(cypher_fields) + " "