fix:improve save

Author: cunla

scheduler/redis_models/base.py

@@ -108,7 +108,8 @@ def deserialize(cls, data: Dict[str, Any]) -> Self:
                 logger.warning(f"Unknown field {k} in {cls.__name__}")
                 continue
             data[k] = _deserialize(data[k], types[k])
-        return cls(**data)
+        res= cls(**data)
+        return res
 
 
 @dataclasses.dataclass(slots=True, kw_only=True)
@@ -125,10 +126,17 @@ def __post_init__(self):
         self._save_all = True
 
     def __setattr__(self, key, value):
-        if key != "_dirty_fields" and hasattr(self, "_dirty_fields"):
+        if not key.startswith("_") and hasattr(self, "_dirty_fields"):
             self._dirty_fields.add(key)
         super(HashModel, self).__setattr__(key, value)
 
+    @classmethod
+    def deserialize(cls, data: Dict[str, Any]) -> Self:
+        instance = super(HashModel, cls).deserialize(data)
+        instance._dirty_fields = set()
+        instance._save_all = False
+        return instance
+
     @property
     def _parent_key(self) -> Optional[str]:
         if self.parent is None:
@@ -171,27 +179,30 @@ def get(cls, name: str, connection: ConnectionType) -> Optional[Self]:
 
     @classmethod
     def get_many(cls, names: Sequence[str], connection: ConnectionType) -> List[Optional[Self]]:
-        pipeline = connection.pipeline()
-        for name in names:
-            pipeline.hgetall(cls._element_key_template.format(name))
-        values = pipeline.execute()
+        with connection.pipeline() as pipeline:
+            for name in names:
+                pipeline.hgetall(cls._element_key_template.format(name))
+            values = pipeline.execute()
         return [(cls.deserialize(decode_dict(v, set())) if v else None) for v in values]
 
-    def save(self, connection: ConnectionType) -> None:
-        connection.sadd(self._list_key, self.name)
-        if self._parent_key is not None:
-            connection.sadd(self._parent_key, self.name)
-        mapping = self.serialize(with_nones=True)
-        if not self._save_all and len(self._dirty_fields) > 0:
-            mapping = {k: v for k, v in mapping.items() if k in self._dirty_fields}
-        none_values = {k for k, v in mapping.items() if v is None}
-        if none_values:
-            connection.hdel(self._key, *none_values)
-        mapping = {k: v for k, v in mapping.items() if v is not None}
-        if mapping:
-            connection.hset(self._key, mapping=mapping)
-        self._dirty_fields = set()
-        self._save_all = False
+    def save(self, connection: ConnectionType, save_all:bool=False) -> None:
+        save_all = save_all or self._save_all
+        with connection.pipeline() as pipeline:
+            pipeline.sadd(self._list_key, self.name)
+            if self._parent_key is not None:
+                pipeline.sadd(self._parent_key, self.name)
+            mapping = self.serialize(with_nones=True)
+            if not save_all:
+                mapping = {k: v for k, v in mapping.items() if k in self._dirty_fields}
+            none_values = {k for k, v in mapping.items() if v is None}
+            if none_values:
+                pipeline.hdel(self._key, *none_values)
+            mapping = {k: v for k, v in mapping.items() if v is not None}
+            if mapping:
+                pipeline.hset(self._key, mapping=mapping)
+            pipeline.execute()
+            self._dirty_fields = set()
+            self._save_all = False
 
     def delete(self, connection: ConnectionType) -> None:
         connection.srem(self._list_key, self._key)