Python追加写入JSON文件读取报错,求可更新存储方案
问题根源
用a模式追加JSON时,每次json.dump会输出一段完整的JSON字符串,多次执行后文件里会出现多个首尾相连的JSON结构,而json.load只能解析单个合法的JSON对象/数组,所以触发JSONDecodeError。
可行解决方案
方案1:JSON Lines格式(逐行存储单个JSON对象)
这是最适合追加场景的JSON变种,每行存一个独立的JSON对象,既支持追加写入,也能逐行解析读取。
写入(支持追加)
每次把单个任务转成JSON字符串后加换行写入:
import json def append_task(task): with open('tasks.jsonl', 'a', encoding='utf8') as f: json.dump(task, f) f.write('\n') # 每行一个JSON对象 # 测试追加 task1 = ["1Name", "1c:/path/file.ext", "1Some other text", 10, ["a", "b", "c"], ""] task2 = ["2Name", "2c:/path/file.ext", "2Test text", 20, ["a", "b", "c"], ""] append_task(task1) append_task(task2)
读取
逐行读取并解析每一行的JSON:
def read_tasks(): tasks = [] with open('tasks.jsonl', 'r', encoding='utf8') as f: for line in f: line = line.strip() if line: # 跳过空行 tasks.append(json.loads(line)) return tasks print(read_tasks())
删除数据
读取所有任务,过滤掉不需要的,再重新写入覆盖文件:
def delete_task(name): tasks = read_tasks() # 过滤掉指定名称的任务(这里假设第一个元素是名称) filtered_tasks = [t for t in tasks if t[0] != name] # 重新写入覆盖 with open('tasks.jsonl', 'w', encoding='utf8') as f: for task in filtered_tasks: json.dump(task, f) f.write('\n') # 测试删除名称为"1Name"的任务 delete_task("1Name")
方案2:读写完整JSON数组(适合数据量小的场景)
每次追加前先读取现有数组,把新数据合并进去,再用w模式覆盖写入,保证文件始终是一个合法的JSON数组。
import json def append_to_json(new_data): try: # 先读取现有数据 with open('test.txt', 'r', encoding='utf8') as f: existing_data = json.load(f) except FileNotFoundError: # 文件不存在则初始化空数组 existing_data = [] # 合并新数据(如果new_data是列表就extend,是单个元素就append) if isinstance(new_data, list): existing_data.extend(new_data) else: existing_data.append(new_data) # 覆盖写入 with open('test.txt', 'w', encoding='utf8') as f: json.dump(existing_data, f, indent=2) # 测试追加 new_tasks = [["1Name", "1c:/path/file.ext", "1Some other text", 10, ["a", "b", "c"], ""], ["2Name", "2c:/path/file.ext", "2Test text", 20, ["a", "b", "c"], ""]] append_to_json(new_tasks) # 读取函数不变 def read_json(): with open('test.txt', 'r', encoding='utf8') as f: return json.load(f) print(read_json())
方案3:用SQLite数据库(适合频繁增删改查的场景)
如果数据量较大、需要复杂查询或频繁修改,直接用Python内置的SQLite数据库更省心,不用自己处理文件格式问题。
import sqlite3 import json # 初始化数据库表 def init_db(): conn = sqlite3.connect('tasks.db') cursor = conn.cursor() # 创建表:id自增,name、path、desc、num、tags(用JSON存列表)、extra cursor.execute(''' CREATE TABLE IF NOT EXISTS tasks ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT, path TEXT, desc TEXT, num INTEGER, tags TEXT, extra TEXT ) ''') conn.commit() conn.close() # 插入任务 def add_task(name, path, desc, num, tags, extra): conn = sqlite3.connect('tasks.db') cursor = conn.cursor() # 把tags列表转成JSON字符串存储 cursor.execute(''' INSERT INTO tasks (name, path, desc, num, tags, extra) VALUES (?, ?, ?, ?, ?, ?) ''', (name, path, desc, num, json.dumps(tags), extra)) conn.commit() conn.close() # 查询所有任务 def get_all_tasks(): conn = sqlite3.connect('tasks.db') cursor = conn.cursor() cursor.execute('SELECT * FROM tasks') tasks = [] for row in cursor.fetchall(): # 把tags的JSON字符串转回列表 task = list(row[1:]) # 跳过id task[4] = json.loads(task[4]) tasks.append(task) conn.close() return tasks # 删除任务(按名称) def delete_task_by_name(name): conn = sqlite3.connect('tasks.db') cursor = conn.cursor() cursor.execute('DELETE FROM tasks WHERE name = ?', (name,)) conn.commit() conn.close() # 测试使用 init_db() add_task("1Name", "1c:/path/file.ext", "1Some other text", 10, ["a", "b", "c"], "") add_task("2Name", "2c:/path/file.ext", "2Test text", 20, ["a", "b", "c"], "") print(get_all_tasks()) delete_task_by_name("1Name") print(get_all_tasks())
内容的提问来源于stack exchange,提问作者Eitan
相关产品推荐
相关产品推荐

