PTV/ptvp35
1
1
Fork 0
Code Issues Pull Requests Packages Projects Releases 4 Wiki Activity

KVJson._load_key + aexit backup

Browse Source
This commit is contained in:
AF 2021-12-08 16:32:51 +03:00 committed by parrrate
parent 142daba766
commit cbf21dd022
Signed by: alisa
SSH Key Fingerprint: SHA256:vNY4pdIZvO1FYJKHROkdHLtvyopizvZVAEwg9AF6h04
1 changed files with 19 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
ptvp35/__init__.py
View File

@ -4,7 +4,7 @@ import pathlib
import pickle import pickle
import shutil import shutil
from io import StringIO from io import StringIO
from typing import Any, Optional, IO, Type from typing import Any, Optional, IO, Type, Hashable
class Request: class Request:
@ -66,10 +66,21 @@ class KVJson(KVRequest):
def line(self) -> str: def line(self) -> str:
return json.dumps({'key': self.key, 'value': self.value}) + "\n" return json.dumps({'key': self.key, 'value': self.value}) + "\n"
@classmethod
def _load_key(cls, key: Any) -> Hashable:
if isinstance(key, Hashable):
return key
elif isinstance(key, list):
return tuple(map(cls._load_key, key))
elif isinstance(key, dict):
return tuple((cls._load_key(k), cls._load_key(v)) for k, v in key)
else:
raise TypeError("unknown KVJson key type, cannot convert to hashable")
@classmethod @classmethod
def fromline(cls, line: str) -> 'KVJson': def fromline(cls, line: str) -> 'KVJson':
d = json.loads(line) d = json.loads(line)
return KVJson(d['key'], d['value'], None) return KVJson(cls._load_key(d['key']), d['value'], None)
class Db: class Db:
@ -226,8 +237,12 @@ class Db:
self.__task.cancel() self.__task.cancel()
await self._dump_buffer() await self._dump_buffer()
self.__file.close() self.__file.close()
with open(self.__path, "w") as file: with open(self.__path_backup, "w") as file:
self.db2io(self.__mmdb, file) self.__initial_size = self.db2io(self.__mmdb, file)
self.__path_recover.touch()
shutil.copy(self.__path_backup, self.__path)
self.__path_recover.unlink()
self.__path_backup.unlink()
def _uninitialize(self): def _uninitialize(self):
self.__mmdb = None self.__mmdb = None