第三方触发多进程Python脚本写入SQLite的并发访问问题咨询

Hey there! Let's break down how to handle SQLite concurrent writes when multiple independent script/EXE processes are spawning (one per file scanned by that third-party app), plus make sure your next app only gets called once. Here are practical, battle-tested approaches:

1. Leverage SQLite's Built-In Locking & Connection Best Practices #

SQLite handles concurrency with file-level locks, so you need to configure your connections to play nice:

• Use the timeout parameter: When opening your database connection, set a timeout (e.g., 10 seconds) so processes wait instead of immediately throwing errors when another holds a write lock. Example:

  import sqlite3
  conn = sqlite3.connect('your_files.db', timeout=10)
  

• Always close connections properly: Use a context manager (with statement) to auto-handle connection closure, which prevents stale locks from hanging around. Like this:

  with sqlite3.connect('your_files.db', timeout=10) as conn:
      # Your write operations here
      conn.execute("INSERT INTO file_paths (path) VALUES (?)", (file_path,))
  # Connection closes automatically when exiting the block
  

• Enable WAL Mode: Switch SQLite to Write-Ahead Logging for better concurrency. This allows reads to happen even while a write is in progress (instead of blocking all reads). Enable it once per connection (it sticks for the database after first use):

  with sqlite3.connect('your_files.db', timeout=10) as conn:
      conn.execute("PRAGMA journal_mode=WAL")
      # Rest of your code
  

2. Ensure Atomic Writes & Avoid Duplicate Entries #

Since each process is handling a unique file, you might still run into edge cases where a file gets processed twice. Fix this with:

• Unique Constraint + INSERT OR IGNORE: Add a unique index on your file_path column first:

  CREATE TABLE IF NOT EXISTS file_paths (
      id INTEGER PRIMARY KEY AUTOINCREMENT,
      path TEXT NOT NULL UNIQUE
  );
  

  Then use INSERT OR IGNORE to write entries—this will skip duplicate paths without throwing errors:

  conn.execute("INSERT OR IGNORE INTO file_paths (path) VALUES (?)", (file_path,))
  

3. Guarantee Your Next App Only Runs Once #

Even if you’ve tried to limit the call in your script, multiple processes might still race to trigger it. Use SQLite as an atomic "gatekeeper":

• Create a status table: First, set up a simple table to track whether the app has been triggered:

  CREATE TABLE IF NOT EXISTS app_trigger (
      id INTEGER PRIMARY KEY,
      triggered BOOLEAN NOT NULL DEFAULT 0,
      triggered_at TIMESTAMP
  );
  -- Insert the initial status if it doesn't exist
  INSERT OR IGNORE INTO app_trigger (id) VALUES (1);
  

• Atomic update to check and trigger: Each process, after writing its file path, tries to flip the triggered flag from 0 to 1. Only the process that succeeds (i.e., the first one) runs the next app:

  with sqlite3.connect('your_files.db', timeout=10) as conn:
      # First, write the file path
      conn.execute("INSERT OR IGNORE INTO file_paths (path) VALUES (?)", (file_path,))
      
      # Now check if we need to trigger the next app
      cursor = conn.execute("UPDATE app_trigger SET triggered = 1, triggered_at = CURRENT_TIMESTAMP WHERE id = 1 AND triggered = 0")
      if cursor.rowcount == 1:
          # This process is the first to trigger—run your next app here
          import subprocess
          subprocess.run(["path/to/your/next/app.exe"])
  

This works because SQLite executes the UPDATE atomically—only one process will get a rowcount of 1, all others will get 0 and skip the app call.

4. Extra Tips for Smooth Operation #

• Keep transactions short: Don’t hold open connections or transactions longer than necessary. The faster you release locks, the less chance of timeouts or bottlenecks.
• Handle exceptions gracefully: Wrap database operations in try/except blocks to catch timeout errors or lock issues, and maybe add a retry logic for transient failures:

  import time
  max_retries = 3
  retry_count = 0
  success = False
  
  while retry_count < max_retries and not success:
      try:
          with sqlite3.connect('your_files.db', timeout=10) as conn:
              conn.execute("INSERT OR IGNORE INTO file_paths (path) VALUES (?)", (file_path,))
          success = True
      except sqlite3.OperationalError as e:
          if "database is locked" in str(e):
              retry_count += 1
              time.sleep(2)
          else:
              raise
  

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