API开发方案咨询：基于服务器JSON文件的多机读写接口实现

Hey there! Let's break down how to build this API that lets 20 machines read/write to a JSON-backed "table" on your server's VM. I'll cover two practical approaches—one lightweight for simpler use cases, and a more robust option for better concurrency handling.

方案1：轻量API + 文件锁（适合低到中等读写频率） #

If your 20 machines aren't hammering the JSON file with simultaneous writes every second, this lightweight approach works great. We'll use file locking to prevent race conditions when multiple clients try to write at the same time.

实施步骤 #

1. 服务器VM准备

   • SSH into your VM and create a dedicated directory for your project:

     mkdir /opt/machine-api && cd /opt/machine-api
     

   • Create the initial JSON "table" file (machines.json) with a clean structure:

     {
       "machines": [
         {"id": 1, "name": "Machine-1", "status": "active"},
         {"id": 2, "name": "Machine-2", "status": "inactive"}
       ]
     }
     

   • Set proper permissions so the API process can read/write the file:

     chmod 644 machines.json
     

2. 本地开发API（以Python + FastAPI为例）
   FastAPI is fast, easy to build with, and has great built-in tools. Here's a minimal implementation with safe file locking:

   • First, install dependencies locally:

     pip install fastapi uvicorn filelock
     

   • Create a main.py file with this code:

     from fastapi import FastAPI, HTTPException
     from filelock import FileLock
     import json
     import os
     
     app = FastAPI()
     JSON_FILE = "/opt/machine-api/machines.json"
     LOCK_FILE = "/opt/machine-api/machines.lock"
     
     # Safe read helper with lock
     def read_json():
         with FileLock(LOCK_FILE):
             if not os.path.exists(JSON_FILE):
                 return {"machines": []}
             with open(JSON_FILE, "r") as f:
                 return json.load(f)
     
     # Safe write helper with lock
     def write_json(data):
         with FileLock(LOCK_FILE):
             with open(JSON_FILE, "w") as f:
                 json.dump(data, f, indent=2)
     
     # Get all machines
     @app.get("/machines")
     def get_machines():
         return read_json()
     
     # Add or update a machine
     @app.post("/machines")
     def update_machine(machine_data: dict):
         data = read_json()
         # Check if machine exists to update or add
         existing_idx = next((i for i, m in enumerate(data["machines"]) if m["id"] == machine_data["id"]), None)
         if existing_idx is not None:
             data["machines"][existing_idx] = machine_data
         else:
             data["machines"].append(machine_data)
         write_json(data)
         return {"status": "success", "machine": machine_data}
     
     # Delete a machine
     @app.delete("/machines/{machine_id}")
     def delete_machine(machine_id: int):
         data = read_json()
         original_count = len(data["machines"])
         data["machines"] = [m for m in data["machines"] if m["id"] != machine_id]
         if len(data["machines"]) == original_count:
             raise HTTPException(status_code=404, detail="Machine not found")
         write_json(data)
         return {"status": "success", "deleted_id": machine_id}
     

3. 本地测试API

   • Run the API locally to validate functionality:

     uvicorn main:app --reload
     

   • Test endpoints with curl or Postman:

     # Fetch all machines
     curl http://localhost:8000/machines
     
     # Add a new machine
     curl -X POST -H "Content-Type: application/json" -d '{"id":3,"name":"Machine-3","status":"active"}' http://localhost:8000/machines
     

4. 部署API到服务器VM

   • Upload your main.py to the VM using scp:

     scp main.py your-username@your-vm-ip:/opt/machine-api/
     

   • Install dependencies on the VM:

     pip install fastapi uvicorn filelock
     

   • Set up a systemd service to keep the API running reliably:
     • Create /etc/systemd/system/machine-api.service:

       [Unit]
       Description=Machine JSON API
       After=network.target
       
       [Service]
       User=your-username
       WorkingDirectory=/opt/machine-api
       ExecStart=/usr/bin/uvicorn main:app --host 0.0.0.0 --port 8000
       Restart=always
       
       [Install]
       WantedBy=multi-user.target
       

     • Reload systemd and start the service:

       sudo systemctl daemon-reload
       sudo systemctl start machine-api
       sudo systemctl enable machine-api
       

   • Verify accessibility from other machines:

     curl http://your-vm-ip:8000/machines
     

方案2：API + 轻量数据库（适合高并发读写） #

If your 20 machines are doing frequent writes, pure JSON files can become a bottleneck even with locks. Replacing the JSON file with a lightweight database like SQLite eliminates race conditions entirely and improves performance.

实施步骤（以SQLite为例，无需额外数据库 server） #

1. 服务器VM准备

   • SSH into your VM and create the project directory:

     mkdir /opt/machine-api && cd /opt/machine-api
     

   • Create a SQLite database and table using the sqlite3 CLI:

     sqlite3 machines.db
     

     Run these SQL commands to set up your table:

     CREATE TABLE machines (
         id INTEGER PRIMARY KEY,
         name TEXT NOT NULL,
         status TEXT NOT NULL
     );
     -- Insert initial test data
     INSERT INTO machines VALUES (1, 'Machine-1', 'active');
     INSERT INTO machines VALUES (2, 'Machine-2', 'inactive');
     .exit
     

2. 本地开发API（FastAPI + SQLite）

   • Install dependencies locally:

     pip install fastapi uvicorn sqlite3
     

   • Create main.py with this database-backed code:

     from fastapi import FastAPI, HTTPException
     import sqlite3
     from typing import Dict
     
     app = FastAPI()
     DB_FILE = "/opt/machine-api/machines.db"
     
     # Helper to get a database connection
     def get_db_connection():
         conn = sqlite3.connect(DB_FILE)
         conn.row_factory = sqlite3.Row  # Return rows as dictionaries
         return conn
     
     # Get all machines
     @app.get("/machines")
     def get_machines():
         conn = get_db_connection()
         machines = conn.execute("SELECT * FROM machines").fetchall()
         conn.close()
         return {"machines": [dict(m) for m in machines]}
     
     # Add or update a machine
     @app.post("/machines")
     def update_machine(machine_data: Dict):
         conn = get_db_connection()
         try:
             # Check if machine exists
             existing = conn.execute("SELECT id FROM machines WHERE id = ?", (machine_data["id"],)).fetchone()
             if existing:
                 conn.execute("UPDATE machines SET name = ?, status = ? WHERE id = ?",
                             (machine_data["name"], machine_data["status"], machine_data["id"]))
             else:
                 conn.execute("INSERT INTO machines (id, name, status) VALUES (?, ?, ?)",
                             (machine_data["id"], machine_data["name"], machine_data["status"]))
             conn.commit()
         except Exception as e:
             conn.rollback()
             raise HTTPException(status_code=500, detail=str(e))
         finally:
             conn.close()
         return {"status": "success", "machine": machine_data}
     
     # Delete a machine
     @app.delete("/machines/{machine_id}")
     def delete_machine(machine_id: int):
         conn = get_db_connection()
         cursor = conn.execute("DELETE FROM machines WHERE id = ?", (machine_id,))
         conn.commit()
         conn.close()
         if cursor.rowcount == 0:
             raise HTTPException(status_code=404, detail="Machine not found")
         return {"status": "success", "deleted_id": machine_id}
     

3. 测试与部署

   • Follow the same local testing steps as方案1 to validate endpoints.
   • Upload main.py to the VM, install dependencies, and set up the systemd service exactly like in方案1.

额外建议 #

• Security: Add API key authentication if you don't want public access. FastAPI makes this easy with simple API key checks or OAuth2 flows.
• Monitoring: Add basic logging to your API to track requests and errors. Use Python's built-in logging module to write logs to a file.
• Backup: For方案1, set up a cron job to back up the JSON file daily. For方案2, use SQLite's backup commands or cron to copy the machines.db file to a safe location.

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