From f-string SQL to Parameterized Queries: Preventing Injection in Database Scripts

import sqlite3
import os
import logging
from contextlib import contextmanager

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger(__name__)

# Database path from environment — avoids hardcoding filesystem assumptions
DB_PATH = os.getenv("USER_DB_PATH", "users.db")


@contextmanager
def get_connection(db_path: str = DB_PATH):
  """Context manager for database connections — guarantees cleanup on error."""
  conn = sqlite3.connect(db_path)
  conn.execute("PRAGMA foreign_keys = ON")   # Enforce referential integrity
  conn.execute("PRAGMA journal_mode = WAL")  # Better concurrent read performance
  try:
      yield conn
      conn.commit()
  except Exception:
      conn.rollback()
      raise
  finally:
      conn.close()


def create_database(db_path: str = DB_PATH) -> None:
  """Initialize the database schema with constrained column definitions."""
  with get_connection(db_path) as conn:
      conn.execute("""
          CREATE TABLE IF NOT EXISTS users (
              id INTEGER PRIMARY KEY AUTOINCREMENT,
              name TEXT NOT NULL,
              email TEXT NOT NULL UNIQUE,
              role TEXT NOT NULL DEFAULT 'viewer'
                  CHECK(role IN ('admin', 'editor', 'viewer'))
          )
      """)
  logger.info("Database initialized")


def add_user(name: str, email: str, role: str = "viewer",
           db_path: str = DB_PATH) -> int:
  """Insert a new user using parameterized queries. Returns the new user ID."""
  # Validate role before touching the database — fail fast
  allowed_roles = {"admin", "editor", "viewer"}
  if role not in allowed_roles:
      raise ValueError(f"Invalid role '{role}'. Must be one of: {allowed_roles}")

  with get_connection(db_path) as conn:
      # Parameterized query: the ? placeholders are filled by the database driver,
      # which escapes all special characters. SQL injection is structurally impossible.
      cursor = conn.execute(
          "INSERT INTO users (name, email, role) VALUES (?, ?, ?)",
          (name, email, role),
      )
      user_id = cursor.lastrowid
      logger.info(f"Added user '{name}' (id={user_id}, role={role})")
      return user_id


def search_users(name: str, db_path: str = DB_PATH) -> list[dict]:
  """Search for users by name using a parameterized LIKE query."""
  with get_connection(db_path) as conn:
      # The % wildcards are part of the parameter value, not the SQL string.
      # This prevents injection through the search term.
      cursor = conn.execute(
          "SELECT id, name, email, role FROM users WHERE name LIKE ?",
          (f"%{name}%",),
      )
      columns = [desc[0] for desc in cursor.description]
      results = [dict(zip(columns, row)) for row in cursor.fetchall()]
      logger.info(f"Search for '{name}' returned {len(results)} result(s)")
      return results


def delete_user(user_id: int, db_path: str = DB_PATH) -> bool:
  """Delete a user by ID. Returns True if a row was actually deleted."""
  # Coerce to int — if someone passes "1; DROP TABLE users", this raises ValueError
  user_id = int(user_id)
  with get_connection(db_path) as conn:
      cursor = conn.execute(
          "DELETE FROM users WHERE id = ?",
          (user_id,),
      )
      deleted = cursor.rowcount > 0
      if deleted:
          logger.info(f"Deleted user id={user_id}")
      else:
          logger.warning(f"No user found with id={user_id}")
      return deleted


if __name__ == "__main__":
  create_database()
  uid = add_user("Alice", "alice@example.com", "admin")
  results = search_users("Alice")
  for user in results:
      print(user)