如何用Python异步实现JSON文件的写入及更新（含现有代码）

Hey there! Let's get your async JSON file handling up and running. You've got a solid start with the menu structure, so we'll expand that to use proper asynchronous I/O (since synchronous file operations can block your program unnecessarily).

First, we'll use aiofiles—a popular library for async file operations in Python—since the standard open() is synchronous. You'll need to install it first:

pip install aiofiles

Here's the complete, async-ready code with all three functions (Create/Update/Delete) implemented, plus error handling for common cases like missing files or invalid JSON:

import os
import json
import asyncio
import aiofiles

# Sample initial data (you can adjust this or let users input new data from scratch)
initial_data = [
    {"username": "name1", "email": "mail1@example.com", "password": "password1"},
    {"username": "name2", "email": "mail2@example.com", "password": "password2"}
]

async def create_file(filename):
    """Async function to create a JSON file with initial data"""
    if os.path.exists(filename):
        print(f"Warning: {filename} already exists! Overwriting...")
    async with aiofiles.open(filename, 'w') as f:
        await f.write(json.dumps(initial_data, indent=4))
    print(f"Successfully created {filename} with initial data!")

async def update_file(filename):
    """Async function to update existing JSON data (add new user or modify existing)"""
    if not os.path.exists(filename):
        print(f"Error: {filename} doesn't exist! Create it first with option 1.")
        return
    
    # Ask user what they want to do: add new user or update existing
    update_choice = int(input("Update options:\n1.Add new user\n2.Modify existing user "))
    
    try:
        # Read existing data
        async with aiofiles.open(filename, 'r') as f:
            content = await f.read()
            data = json.loads(content)
        
        if update_choice == 1:
            # Add new user
            new_user = {
                "username": input("Enter new username: "),
                "email": input("Enter new email: "),
                "password": input("Enter new password: ")
            }
            data.append(new_user)
        elif update_choice == 2:
            # Modify existing user
            target_username = input("Enter username to modify: ")
            user_found = False
            for user in data:
                if user["username"] == target_username:
                    user["email"] = input("Enter new email (leave blank to keep current): ") or user["email"]
                    user["password"] = input("Enter new password (leave blank to keep current): ") or user["password"]
                    user_found = True
                    break
            if not user_found:
                print(f"No user found with username: {target_username}")
                return
        
        # Write updated data back to file
        async with aiofiles.open(filename, 'w') as f:
            await f.write(json.dumps(data, indent=4))
        print(f"Successfully updated {filename}!")
    
    except json.JSONDecodeError:
        print(f"Error: {filename} contains invalid JSON. Can't update it.")

async def delete_file_entry(filename):
    """Async function to delete a user entry from the JSON file"""
    if not os.path.exists(filename):
        print(f"Error: {filename} doesn't exist! Create it first with option 1.")
        return
    
    try:
        # Read existing data
        async with aiofiles.open(filename, 'r') as f:
            content = await f.read()
            data = json.loads(content)
        
        target_username = input("Enter username to delete: ")
        original_length = len(data)
        data = [user for user in data if user["username"] != target_username]
        
        if len(data) == original_length:
            print(f"No user found with username: {target_username}")
            return
        
        # Write updated data back to file
        async with aiofiles.open(filename, 'w') as f:
            await f.write(json.dumps(data, indent=4))
        print(f"Successfully deleted user '{target_username}' from {filename}!")
    
    except json.JSONDecodeError:
        print(f"Error: {filename} contains invalid JSON. Can't delete from it.")

async def main():
    while True:
        try:
            choice = int(input("\nWhat you want to do:\n1.Create\n2.Update\n3.Delete\n4.Exit "))
            if choice == 1:
                filename = input("Enter the filename (e.g., users.json): ")
                await create_file(filename)
            elif choice == 2:
                filename = input("Enter the filename to update: ")
                await update_file(filename)
            elif choice == 3:
                filename = input("Enter the filename to delete from: ")
                await delete_file_entry(filename)
            elif choice == 4:
                print("Exiting program...")
                break
            else:
                print("Invalid choice! Please enter 1, 2, 3, or 4.")
        except ValueError:
            print("Please enter a valid number!")

if __name__ == "__main__":
    asyncio.run(main())

Key Notes: #

• Async File Operations: We use aiofiles.open() with async with to handle file I/O without blocking the event loop.
• Error Handling: Added checks for missing files, invalid JSON, and invalid user input to make the program more robust.
• Flexible Updates: The update function lets users either add a new user or modify an existing one's email/password.
• Loop Until Exit: The main menu runs in a loop until the user chooses to exit, which is more user-friendly than a single-run program.

Just run the script, and you'll be able to create, update, and delete JSON data asynchronously!

内容的提问来源于stack exchange，提问作者Rohit