实现json存储文件
This commit is contained in:
@@ -1,2 +1,11 @@
|
||||
# 存储接口
|
||||
|
||||
class BaseStorage:
|
||||
def save(self, key: str, value):
|
||||
raise NotImplementedError
|
||||
|
||||
def load(self, key: str, default=None):
|
||||
raise NotImplementedError
|
||||
|
||||
def delete(self, key: str):
|
||||
raise NotImplementedError
|
||||
|
||||
@@ -1 +1,40 @@
|
||||
# json 文件实现
|
||||
# json 文件实现
|
||||
|
||||
import json
|
||||
|
||||
from storage.base_storage import BaseStorage
|
||||
|
||||
|
||||
class JSONStorage(BaseStorage):
|
||||
def __init__(self, filename="data.json"):
|
||||
self.filename = filename
|
||||
# 如果文件不存在,先创建空字典
|
||||
try:
|
||||
with open(self.filename, "r") as f:
|
||||
pass
|
||||
except OSError:
|
||||
with open(self.filename, "w") as f:
|
||||
f.write("{}")
|
||||
|
||||
def _read_all(self):
|
||||
with open(self.filename, "r") as f:
|
||||
return json.load(f)
|
||||
|
||||
def _write_all(self, data):
|
||||
with open(self.filename, "w") as f:
|
||||
json.dump(data, f)
|
||||
|
||||
def save(self, key, value):
|
||||
data = self._read_all()
|
||||
data[key] = value
|
||||
self._write_all(data)
|
||||
|
||||
def load(self, key, default=None):
|
||||
data = self._read_all()
|
||||
return data.get(key, default)
|
||||
|
||||
def delete(self, key):
|
||||
data = self._read_all()
|
||||
if key in data:
|
||||
del data[key]
|
||||
self._write_all(data)
|
||||
|
||||
57
utils/fs.py
Normal file
57
utils/fs.py
Normal file
@@ -0,0 +1,57 @@
|
||||
# 兼容PC和MicroPython的文件操作
|
||||
|
||||
# compat_fs.py
|
||||
try:
|
||||
import uos as os # MicroPython
|
||||
MICROPYTHON = True
|
||||
except ImportError:
|
||||
import os # CPython
|
||||
MICROPYTHON = False
|
||||
|
||||
def list_dir(path="."):
|
||||
"""列出目录内容"""
|
||||
return os.listdir(path)
|
||||
|
||||
def make_dir(path):
|
||||
"""创建目录"""
|
||||
if MICROPYTHON:
|
||||
os.mkdir(path)
|
||||
else:
|
||||
os.makedirs(path, exist_ok=True)
|
||||
|
||||
def remove_file(path):
|
||||
"""删除文件"""
|
||||
os.remove(path)
|
||||
|
||||
def read_file(path, mode="r"):
|
||||
"""读取文件内容"""
|
||||
with open(path, mode) as f:
|
||||
return f.read()
|
||||
|
||||
def write_file(path, data, mode="w"):
|
||||
"""写入文件内容"""
|
||||
with open(path, mode) as f:
|
||||
f.write(data)
|
||||
|
||||
def is_file(path):
|
||||
"""判断是否是文件"""
|
||||
try:
|
||||
st = os.stat(path)
|
||||
# MicroPython: stat()[0] >> 14 & 0xF == 8 表示普通文件
|
||||
if MICROPYTHON:
|
||||
return (st[0] >> 14) & 0xF == 8
|
||||
else:
|
||||
return os.path.isfile(path)
|
||||
except OSError:
|
||||
return False
|
||||
|
||||
def is_dir(path):
|
||||
"""判断是否是目录"""
|
||||
try:
|
||||
st = os.stat(path)
|
||||
if MICROPYTHON:
|
||||
return (st[0] >> 14) & 0xF == 2
|
||||
else:
|
||||
return os.path.isdir(path)
|
||||
except OSError:
|
||||
return False
|
||||
Reference in New Issue
Block a user