SQLite3 Gem

Master SQLite3 database access with direct connections, prepared statements, transactions, and advanced features for lightweight database applications.

Beginner 1 min read

SQLite3 Gem in Ruby

Introduction to SQLite3

SQLite3 is a lightweight, embedded database engine that's perfect for development, testing, and smaller applications. The Ruby sqlite3 gem provides a comprehensive interface for working with SQLite databases.

Installing SQLite3 Gem

# Install the sqlite3 gem
gem install sqlite3

# Add to Gemfile
gem 'sqlite3', '~> 1.6'

# Install with bundler
bundle install

Basic SQLite3 Usage

require 'sqlite3'

# Create/connect to database
db = SQLite3::Database.new('example.db')

# Create a table
db.execute <<~SQL
  CREATE TABLE IF NOT EXISTS users (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    name TEXT NOT NULL,
    email TEXT UNIQUE NOT NULL,
    age INTEGER,
    created_at DATETIME DEFAULT CURRENT_TIMESTAMP
  )
SQL

# Insert data
db.execute("INSERT INTO users (name, email, age) VALUES (?, ?, ?)",
           ["Alice", "alice@example.com", 30])

# Query data
rows = db.execute("SELECT * FROM users")
rows.each do |row|
  puts row.inspect
end

# Close the database
db.close

Database Connection Management

Proper database connection management is crucial for application stability and performance.

Connection Options

require 'sqlite3'

# Basic connection
db = SQLite3::Database.new('myapp.db')

# In-memory database (for testing)
memory_db = SQLite3::Database.new(':memory:')

# Read-only connection
readonly_db = SQLite3::Database.new('data.db', readonly: true)

# Connection with options
db = SQLite3::Database.new('app.db') do |database|
  database.busy_timeout(5000)  # 5 seconds timeout
  database.execute("PRAGMA foreign_keys = ON")
  database.execute("PRAGMA journal_mode = WAL")
end

# Always close connections
db.close

Database Pool Class

class DatabasePool
  def initialize(database_path, pool_size: 5)
    @database_path = database_path
    @pool_size = pool_size
    @pool = []
    @mutex = Mutex.new
    initialize_pool
  end

  def with_connection
    connection = acquire_connection
    begin
      yield connection
    ensure
      release_connection(connection)
    end
  end

  def close_all
    @mutex.synchronize do
      @pool.each(&:close)
      @pool.clear
    end
  end

  private

  def initialize_pool
    @pool_size.times do
      @pool << create_connection
    end
  end

  def create_connection
    db = SQLite3::Database.new(@database_path)
    db.busy_timeout(5000)
    db.execute("PRAGMA foreign_keys = ON")
    db
  end

  def acquire_connection
    @mutex.synchronize do
      if @pool.empty?
        create_connection
      else
        @pool.pop
      end
    end
  end

  def release_connection(connection)
    @mutex.synchronize do
      @pool.push(connection) if @pool.size < @pool_size
    end
  end
end

# Usage
pool = DatabasePool.new('myapp.db')

pool.with_connection do |db|
  users = db.execute("SELECT * FROM users WHERE active = 1")
  # Process users...
end

Prepared Statements

Prepared statements provide better performance and security by pre-compiling SQL statements and protecting against SQL injection.

Basic Prepared Statements

require 'sqlite3'

db = SQLite3::Database.new('example.db')

# Prepare statement
stmt = db.prepare("SELECT * FROM users WHERE age > ? AND city = ?")

# Execute with different parameters
results1 = stmt.execute(25, 'New York')
results1.each { |row| puts row.inspect }

results2 = stmt.execute(30, 'Los Angeles')
results2.each { |row| puts row.inspect }

# Always close prepared statements
stmt.close

# Alternative: using block syntax (auto-closes)
db.prepare("INSERT INTO users (name, email, age) VALUES (?, ?, ?)") do |stmt|
  stmt.execute("Bob", "bob@example.com", 28)
  stmt.execute("Carol", "carol@example.com", 35)
  stmt.execute("Dave", "dave@example.com", 42)
end

db.close

Prepared Statement Manager

class PreparedStatements
  def initialize(database)
    @db = database
    @statements = {}
  end

  def prepare(name, sql)
    @statements[name] = @db.prepare(sql)
  end

  def execute(name, *params)
    statement = @statements[name]
    raise "Statement '#{name}' not found" unless statement

    statement.execute(*params)
  end

  def execute_single(name, *params)
    result = execute(name, *params)
    row = result.next
    result.close
    row
  end

  def execute_all(name, *params)
    result = execute(name, *params)
    rows = result.to_a
    result.close
    rows
  end

  def close_all
    @statements.each_value(&:close)
    @statements.clear
  end
end

# Usage
db = SQLite3::Database.new('example.db')
stmts = PreparedStatements.new(db)

# Prepare commonly used statements
stmts.prepare(:find_user_by_id, "SELECT * FROM users WHERE id = ?")
stmts.prepare(:find_users_by_age, "SELECT * FROM users WHERE age > ?")
stmts.prepare(:insert_user, "INSERT INTO users (name, email, age) VALUES (?, ?, ?)")
stmts.prepare(:update_user_age, "UPDATE users SET age = ? WHERE id = ?")

# Use prepared statements
user = stmts.execute_single(:find_user_by_id, 1)
puts user.inspect

young_users = stmts.execute_all(:find_users_by_age, 25)
young_users.each { |user| puts user.inspect }

stmts.execute(:insert_user, "Eve", "eve@example.com", 29)
stmts.execute(:update_user_age, 31, 1)

# Cleanup
stmts.close_all
db.close

Advanced SQLite3 Features

SQLite3 offers many advanced features for performance optimization and data management.

Transactions and Performance

require 'sqlite3'
require 'benchmark'

db = SQLite3::Database.new('performance_test.db')

# Setup test table
db.execute <<~SQL
  CREATE TABLE IF NOT EXISTS items (
    id INTEGER PRIMARY KEY,
    name TEXT,
    value REAL
  )
SQL

# Without transaction (slow)
time_without_transaction = Benchmark.measure do
  1000.times do |i|
    db.execute("INSERT INTO items (name, value) VALUES (?, ?)",
               ["Item #{i}", rand * 100])
  end
end

# With transaction (fast)
time_with_transaction = Benchmark.measure do
  db.transaction do
    1000.times do |i|
      db.execute("INSERT INTO items (name, value) VALUES (?, ?)",
                 ["Item #{i + 1000}", rand * 100])
    end
  end
end

puts "Without transaction: #{time_without_transaction.real.round(2)}s"
puts "With transaction: #{time_with_transaction.real.round(2)}s"

# Explicit transaction handling
begin
  db.execute("BEGIN TRANSACTION")

  # Multiple operations
  db.execute("UPDATE items SET value = value * 1.1 WHERE id < 500")
  db.execute("DELETE FROM items WHERE value < 10")

  # Commit if everything succeeds
  db.execute("COMMIT")
rescue SQLite3::Exception => e
  # Rollback on error
  db.execute("ROLLBACK")
  puts "Transaction failed: #{e.message}"
end

db.close

Result Set Handling

require 'sqlite3'

db = SQLite3::Database.new('example.db')

# Configure result format
db.results_as_hash = true  # Return results as hashes instead of arrays

# Create sample data
db.execute <<~SQL
  CREATE TABLE IF NOT EXISTS products (
    id INTEGER PRIMARY KEY,
    name TEXT,
    price DECIMAL(10,2),
    category TEXT,
    in_stock BOOLEAN DEFAULT 1
  )
SQL

# Insert sample data
products = [
  ['Laptop', 999.99, 'Electronics', 1],
  ['Book', 19.99, 'Education', 1],
  ['Chair', 149.99, 'Furniture', 0]
]

products.each do |product|
  db.execute("INSERT INTO products (name, price, category, in_stock) VALUES (?, ?, ?, ?)", product)
end

# Query with hash results
puts "=== Hash Results ==="
db.execute("SELECT * FROM products") do |row|
  puts "Product: #{row['name']}, Price: $#{row['price']}, Category: #{row['category']}"
end

# Query with array results
db.results_as_hash = false
puts "\n=== Array Results ==="
db.execute("SELECT name, price, category FROM products") do |row|
  name, price, category = row
  puts "Product: #{name}, Price: $#{price}, Category: #{category}"
end

# Using column names
db.results_as_hash = true
result = db.execute2("SELECT * FROM products LIMIT 1")
headers = result.first
data = result[1..-1]

puts "\n=== Structured Output ==="
puts headers.join(" | ")
puts "-" * (headers.join(" | ").length)
data.each do |row|
  puts headers.map { |h| row[h] }.join(" | ")
end

db.close

Custom Functions and Aggregates

require 'sqlite3'

db = SQLite3::Database.new(':memory:')

# Create custom function
db.create_function('upper_case', 1) do |func, value|
  func.result = value.to_s.upcase
end

# Create custom aggregate function
db.create_aggregate('string_concat', 1) do
  def initialize
    @result = []
  end

  def step(value)
    @result << value.to_s
  end

  def finalize
    @result.join(', ')
  end
end

# Test table
db.execute <<~SQL
  CREATE TABLE test_data (
    id INTEGER PRIMARY KEY,
    name TEXT,
    category TEXT
  )
SQL

# Insert test data
['apple', 'banana', 'cherry'].each_with_index do |fruit, i|
  db.execute("INSERT INTO test_data (name, category) VALUES (?, ?)",
             [fruit, i.even? ? 'sweet' : 'tart'])
end

# Use custom function
puts "=== Custom Function ==="
db.execute("SELECT name, upper_case(name) as upper_name FROM test_data") do |row|
  puts "#{row['name']} => #{row['upper_name']}"
end

# Use custom aggregate
puts "\n=== Custom Aggregate ==="
result = db.execute("SELECT category, string_concat(name) as fruits FROM test_data GROUP BY category")
result.each do |row|
  puts "#{row['category']}: #{row['fruits']}"
end

db.close

Building a Complete Database Layer

Let's create a comprehensive database abstraction layer using SQLite3.

Database Model Base Class

require 'sqlite3'

class DatabaseModel
  @@db = nil

  def self.connection
    @@db ||= SQLite3::Database.new('app.db').tap do |db|
      db.busy_timeout(5000)
      db.results_as_hash = true
      db.execute("PRAGMA foreign_keys = ON")
    end
  end

  def self.execute(sql, params = [])
    connection.execute(sql, params)
  end

  def self.execute_single(sql, params = [])
    result = execute(sql, params)
    result.first
  end

  def self.table_name
    name.downcase + 's'
  end

  def self.create_table
    raise NotImplementedError, "Subclasses must implement create_table"
  end

  def self.find(id)
    row = execute_single("SELECT * FROM #{table_name} WHERE id = ?", [id])
    row ? new(row) : nil
  end

  def self.find_by(conditions = {})
    where_clause = conditions.keys.map { |key| "#{key} = ?" }.join(' AND ')
    sql = "SELECT * FROM #{table_name} WHERE #{where_clause}"
    execute(sql, conditions.values).map { |row| new(row) }
  end

  def self.all
    execute("SELECT * FROM #{table_name}").map { |row| new(row) }
  end

  def initialize(attributes = {})
    @attributes = attributes.transform_keys(&:to_s)
    @persisted = attributes.key?('id') || attributes.key?(:id)
  end

  def id
    @attributes['id']
  end

  def [](key)
    @attributes[key.to_s]
  end

  def []=(key, value)
    @attributes[key.to_s] = value
  end

  def persisted?
    @persisted
  end

  def save
    if persisted?
      update
    else
      create
    end
  end

  def update(attrs = {})
    @attributes.merge!(attrs.transform_keys(&:to_s))

    set_clause = @attributes.keys.reject { |k| k == 'id' }.map { |key| "#{key} = ?" }.join(', ')
    values = @attributes.values_at(*@attributes.keys.reject { |k| k == 'id' })
    values << id

    self.class.execute("UPDATE #{self.class.table_name} SET #{set_clause} WHERE id = ?", values)
    self
  end

  def delete
    return false unless persisted?

    self.class.execute("DELETE FROM #{self.class.table_name} WHERE id = ?", [id])
    @persisted = false
    true
  end

  private

  def create
    columns = @attributes.keys.reject { |k| k == 'id' }
    values = @attributes.values_at(*columns)
    placeholders = (['?'] * columns.length).join(', ')

    sql = "INSERT INTO #{self.class.table_name} (#{columns.join(', ')}) VALUES (#{placeholders})"
    self.class.execute(sql, values)

    @attributes['id'] = self.class.connection.last_insert_row_id
    @persisted = true
    self
  end
end

# Example model
class User < DatabaseModel
  def self.create_table
    execute <<~SQL
      CREATE TABLE IF NOT EXISTS users (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        name TEXT NOT NULL,
        email TEXT UNIQUE NOT NULL,
        age INTEGER,
        created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
        updated_at DATETIME DEFAULT CURRENT_TIMESTAMP
      )
    SQL
  end

  def self.find_by_email(email)
    find_by(email: email).first
  end

  def self.adults
    find_by_sql("SELECT * FROM users WHERE age >= 18")
  end

  def name
    @attributes['name']
  end

  def name=(value)
    @attributes['name'] = value
  end

  def email
    @attributes['email']
  end

  def email=(value)
    @attributes['email'] = value
  end

  def age
    @attributes['age']
  end

  def age=(value)
    @attributes['age'] = value
  end

  def adult?
    age && age >= 18
  end

  private

  def self.find_by_sql(sql, params = [])
    execute(sql, params).map { |row| new(row) }
  end
end

# Usage example
User.create_table

# Create users
user1 = User.new(name: 'Alice', email: 'alice@example.com', age: 30)
user1.save

user2 = User.new(name: 'Bob', email: 'bob@example.com', age: 25)
user2.save

# Query users
all_users = User.all
puts "All users: #{all_users.map(&:name)}"

alice = User.find_by_email('alice@example.com')
puts "Found Alice: #{alice.name}, Age: #{alice.age}"

adults = User.adults
puts "Adult users: #{adults.map(&:name)}"

# Update user
alice.age = 31
alice.save

# Delete user
bob = User.find_by_email('bob@example.com')
bob.delete

Database Migrations

Manage database schema changes with a simple migration system.

Migration System

class Migration
  @@db = nil

  def self.connection
    @@db ||= SQLite3::Database.new('app.db')
  end

  def self.setup_migrations_table
    connection.execute <<~SQL
      CREATE TABLE IF NOT EXISTS schema_migrations (
        version TEXT PRIMARY KEY,
        executed_at DATETIME DEFAULT CURRENT_TIMESTAMP
      )
    SQL
  end

  def self.run_migrations
    setup_migrations_table

    migration_files = Dir.glob('migrations/*.rb').sort

    migration_files.each do |file|
      version = File.basename(file, '.rb')

      # Check if migration already executed
      result = connection.execute("SELECT version FROM schema_migrations WHERE version = ?", [version])
      next unless result.empty?

      puts "Running migration: #{version}"

      # Load and execute migration
      require_relative file
      migration_class = Object.const_get(version.split('_').map(&:capitalize).join)
      migration_class.new.up

      # Record migration
      connection.execute("INSERT INTO schema_migrations (version) VALUES (?)", [version])
      puts "Migration #{version} completed"
    end
  end

  def up
    raise NotImplementedError, "Subclasses must implement 'up' method"
  end

  def down
    raise NotImplementedError, "Subclasses must implement 'down' method"
  end

  protected

  def execute(sql)
    self.class.connection.execute(sql)
  end

  def create_table(name, &block)
    builder = TableBuilder.new(name)
    builder.instance_eval(&block)
    execute(builder.to_sql)
  end

  def add_column(table, column, type, options = {})
    sql = "ALTER TABLE #{table} ADD COLUMN #{column} #{type.upcase}"
    sql += " NOT NULL" if options[:null] == false
    sql += " DEFAULT #{options[:default]}" if options[:default]
    execute(sql)
  end

  def remove_column(table, column)
    # SQLite doesn't support DROP COLUMN, need to recreate table
    puts "Warning: SQLite doesn't support DROP COLUMN. Manual intervention required for #{table}.#{column}"
  end

  def add_index(table, column, options = {})
    index_name = options[:name] || "index_#{table}_on_#{column}"
    unique = options[:unique] ? "UNIQUE" : ""
    execute("CREATE #{unique} INDEX #{index_name} ON #{table} (#{column})")
  end
end

class TableBuilder
  def initialize(name)
    @name = name
    @columns = []
  end

  def integer(name, options = {})
    add_column(name, 'INTEGER', options)
  end

  def text(name, options = {})
    add_column(name, 'TEXT', options)
  end

  def real(name, options = {})
    add_column(name, 'REAL', options)
  end

  def boolean(name, options = {})
    add_column(name, 'BOOLEAN', options)
  end

  def datetime(name, options = {})
    add_column(name, 'DATETIME', options)
  end

  def primary_key(name = 'id')
    @columns << "#{name} INTEGER PRIMARY KEY AUTOINCREMENT"
  end

  def to_sql
    "CREATE TABLE #{@name} (#{@columns.join(', ')})"
  end

  private

  def add_column(name, type, options)
    column_sql = "#{name} #{type}"
    column_sql += " NOT NULL" if options[:null] == false
    column_sql += " DEFAULT #{options[:default]}" if options[:default]
    column_sql += " UNIQUE" if options[:unique]
    @columns << column_sql
  end
end

# Example migrations

# migrations/001_create_users.rb
class CreateUsers < Migration
  def up
    create_table 'users' do
      primary_key 'id'
      text 'name', null: false
      text 'email', null: false, unique: true
      integer 'age'
      datetime 'created_at', default: 'CURRENT_TIMESTAMP'
      datetime 'updated_at', default: 'CURRENT_TIMESTAMP'
    end

    add_index 'users', 'email', unique: true
  end

  def down
    execute("DROP TABLE users")
  end
end

# migrations/002_create_posts.rb
class CreatePosts < Migration
  def up
    create_table 'posts' do
      primary_key 'id'
      text 'title', null: false
      text 'content'
      integer 'user_id', null: false
      datetime 'created_at', default: 'CURRENT_TIMESTAMP'
    end

    add_index 'posts', 'user_id'
    execute("CREATE INDEX index_posts_on_created_at ON posts (created_at)")
  end

  def down
    execute("DROP TABLE posts")
  end
end

# Run migrations
# Migration.run_migrations

SQLite3 Best Practices

  • Always use prepared statements for dynamic queries to prevent SQL injection
  • Wrap multiple operations in transactions for better performance
  • Enable foreign key constraints with PRAGMA foreign_keys = ON
  • Use WAL mode for better concurrent access: PRAGMA journal_mode = WAL
  • Set appropriate busy timeout to handle locked database scenarios
  • Close database connections and prepared statements when done
  • Use connection pooling for multi-threaded applications
  • Index frequently queried columns for better performance
  • Use EXPLAIN QUERY PLAN to optimize slow queries
  • Consider using in-memory databases for testing scenarios