PostgreSQL & MySQL Adapters in Ruby

Introduction to Database Adapters

Ruby provides excellent database adapters for PostgreSQL and MySQL. The `pg` gem connects to PostgreSQL databases, while `mysql2` provides MySQL connectivity. These adapters offer high performance and comprehensive feature support.

Installing Database Adapters

# PostgreSQL adapter
gem install pg

# MySQL adapter
gem install mysql2

# Add to Gemfile
gem 'pg', '~> 1.4'
gem 'mysql2', '~> 0.5'

# Install with bundler
bundle install

PostgreSQL with the pg Gem

PostgreSQL is a powerful, feature-rich relational database. The pg gem provides comprehensive access to PostgreSQL's advanced features.

Basic PostgreSQL Connection

require 'pg'

# Connect to PostgreSQL
conn = PG.connect(
  host: 'localhost',
  port: 5432,
  dbname: 'myapp_development',
  user: 'postgres',
  password: 'password'
)

# Alternative: connection string
conn = PG.connect('postgresql://postgres:password@localhost:5432/myapp_development')

# Create table
conn.exec <<~SQL
  CREATE TABLE IF NOT EXISTS users (
    id SERIAL PRIMARY KEY,
    name VARCHAR(100) NOT NULL,
    email VARCHAR(255) UNIQUE NOT NULL,
    age INTEGER,
    active BOOLEAN DEFAULT true,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
  )
SQL

# Insert data
conn.exec_params(
  "INSERT INTO users (name, email, age) VALUES ($1, $2, $3)",
  ['Alice Johnson', 'alice@example.com', 30]
)

# Query data
result = conn.exec("SELECT * FROM users")
result.each do |row|
  puts "#{row['name']} (#{row['email']}) - Age: #{row['age']}"
end

# Close connection
conn.close

PostgreSQL Connection Pool

require 'pg'
require 'connection_pool'

class PostgreSQLPool
  def initialize(options = {})
    @pool = ConnectionPool.new(size: options[:pool_size] || 5, timeout: 5) do
      PG.connect(
        host: options[:host] || 'localhost',
        port: options[:port] || 5432,
        dbname: options[:database],
        user: options[:username],
        password: options[:password]
      )
    end
  end

  def with_connection(&block)
    @pool.with(&block)
  end

  def exec(sql, params = [])
    with_connection do |conn|
      if params.empty?
        conn.exec(sql)
      else
        conn.exec_params(sql, params)
      end
    end
  end

  def exec_single(sql, params = [])
    result = exec(sql, params)
    row = result.first
    result.clear
    row
  end

  def transaction
    with_connection do |conn|
      conn.transaction do
        yield conn
      end
    end
  end

  def close
    @pool.shutdown { |conn| conn.close }
  end
end

# Usage
db_config = {
  host: 'localhost',
  database: 'myapp_development',
  username: 'postgres',
  password: 'password',
  pool_size: 10
}

pool = PostgreSQLPool.new(db_config)

# Execute queries
user = pool.exec_single(
  "SELECT * FROM users WHERE email = $1",
  ['alice@example.com']
)

# Use transactions
pool.transaction do |conn|
  conn.exec_params("UPDATE users SET age = $1 WHERE id = $2", [31, user['id']])
  conn.exec_params("INSERT INTO user_logs (user_id, action) VALUES ($1, $2)", [user['id'], 'age_updated'])
end

PostgreSQL Advanced Features

require 'pg'

conn = PG.connect('postgresql://postgres:password@localhost:5432/myapp_development')

# JSON and JSONB support
conn.exec <<~SQL
  CREATE TABLE IF NOT EXISTS documents (
    id SERIAL PRIMARY KEY,
    title VARCHAR(255),
    metadata JSONB,
    tags TEXT[],
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
  )
SQL

# Insert JSON data
metadata = {
  author: 'John Doe',
  category: 'Technology',
  word_count: 1500,
  published: true
}

conn.exec_params(
  "INSERT INTO documents (title, metadata, tags) VALUES ($1, $2, $3)",
  [
    'PostgreSQL Guide',
    metadata.to_json,
    ['database', 'postgresql', 'ruby']
  ]
)

# Query JSON data
result = conn.exec(<<~SQL)
  SELECT
    title,
    metadata->>'author' as author,
    metadata->>'category' as category,
    metadata->'word_count' as word_count,
    tags
  FROM documents
  WHERE metadata->>'published' = 'true'
SQL

result.each do |row|
  puts "Title: #{row['title']}"
  puts "Author: #{row['author']}"
  puts "Category: #{row['category']}"
  puts "Words: #{row['word_count']}"
  puts "Tags: #{row['tags']}"
  puts "---"
end

# Array operations
conn.exec_params(
  "SELECT * FROM documents WHERE tags && $1",
  [['database', 'guide']]
)

# Full-text search
conn.exec <<~SQL
  ALTER TABLE documents ADD COLUMN search_vector tsvector;
  UPDATE documents SET search_vector = to_tsvector('english', title);
  CREATE INDEX idx_documents_search ON documents USING gin(search_vector);
SQL

# Search documents
search_results = conn.exec_params(
  "SELECT title FROM documents WHERE search_vector @@ plainto_tsquery('english', $1)",
  ['PostgreSQL']
)

search_results.each do |row|
  puts "Found: #{row['title']}"
end

conn.close

MySQL with the mysql2 Gem

MySQL is a popular relational database. The mysql2 gem provides a fast, efficient interface to MySQL databases.

Basic MySQL Connection

require 'mysql2'

# Connect to MySQL
client = Mysql2::Client.new(
  host: 'localhost',
  port: 3306,
  database: 'myapp_development',
  username: 'root',
  password: 'password',
  encoding: 'utf8mb4'
)

# Create table
client.query <<~SQL
  CREATE TABLE IF NOT EXISTS users (
    id INT AUTO_INCREMENT PRIMARY KEY,
    name VARCHAR(100) NOT NULL,
    email VARCHAR(255) UNIQUE NOT NULL,
    age INT,
    active BOOLEAN DEFAULT true,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
  ) ENGINE=InnoDB CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci
SQL

# Prepare statement
stmt = client.prepare("INSERT INTO users (name, email, age) VALUES (?, ?, ?)")

# Execute prepared statement
stmt.execute('Bob Smith', 'bob@example.com', 28)
stmt.execute('Carol Davis', 'carol@example.com', 35)

# Query data
results = client.query("SELECT * FROM users ORDER BY created_at DESC")
results.each do |row|
  puts "#{row['name']} (#{row['email']}) - Age: #{row['age']}"
end

# Close statement and connection
stmt.close
client.close

MySQL Connection Management

require 'mysql2'

class MySQLConnectionManager
  def initialize(config)
    @config = config
    @connections = []
    @mutex = Mutex.new
    @max_connections = config[:pool_size] || 5
  end

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

  def execute(sql, *params)
    with_connection do |client|
      if params.empty?
        client.query(sql)
      else
        stmt = client.prepare(sql)
        result = stmt.execute(*params)
        stmt.close
        result
      end
    end
  end

  def transaction
    with_connection do |client|
      client.query("START TRANSACTION")
      begin
        result = yield client
        client.query("COMMIT")
        result
      rescue => e
        client.query("ROLLBACK")
        raise e
      end
    end
  end

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

  private

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

  def release_connection(connection)
    @mutex.synchronize do
      if @connections.size < @max_connections && connection.ping
        @connections.push(connection)
      else
        connection.close
      end
    end
  end

  def create_connection
    Mysql2::Client.new(@config)
  end
end

# Usage
mysql_config = {
  host: 'localhost',
  database: 'myapp_development',
  username: 'root',
  password: 'password',
  encoding: 'utf8mb4',
  pool_size: 10,
  reconnect: true
}

db = MySQLConnectionManager.new(mysql_config)

# Execute queries
users = db.execute("SELECT * FROM users WHERE active = ?", true)
users.each { |user| puts user['name'] }

# Use transactions
db.transaction do |client|
  stmt1 = client.prepare("UPDATE users SET age = ? WHERE id = ?")
  stmt1.execute(29, 1)

  stmt2 = client.prepare("INSERT INTO user_activities (user_id, activity) VALUES (?, ?)")
  stmt2.execute(1, 'profile_updated')

  stmt1.close
  stmt2.close
end

MySQL Performance Optimization

require 'mysql2'

class OptimizedMySQL
  def initialize(config)
    @client = Mysql2::Client.new(config.merge(
      read_timeout: 60,
      write_timeout: 60,
      connect_timeout: 10,
      reconnect: true,
      local_infile: false,  # Security
      secure_auth: true
    ))

    optimize_connection
  end

  def bulk_insert(table, columns, data)
    return if data.empty?

    placeholders = (['?'] * columns.length).join(', ')
    sql = "INSERT INTO #{table} (#{columns.join(', ')}) VALUES (#{placeholders})"

    stmt = @client.prepare(sql)

    # Use transaction for bulk operations
    @client.query("START TRANSACTION")

    begin
      data.each do |row|
        stmt.execute(*row)
      end
      @client.query("COMMIT")
    rescue => e
      @client.query("ROLLBACK")
      raise e
    ensure
      stmt.close
    end
  end

  def paginated_query(sql, page: 1, per_page: 100)
    offset = (page - 1) * per_page
    paginated_sql = "#{sql} LIMIT #{per_page} OFFSET #{offset}"

    {
      data: @client.query(paginated_sql).to_a,
      page: page,
      per_page: per_page,
      total: get_count(sql)
    }
  end

  def explain_query(sql)
    explain_result = @client.query("EXPLAIN #{sql}")
    explain_result.each do |row|
      puts "Table: #{row['table']}, Type: #{row['type']}, Key: #{row['key']}"
      puts "Rows: #{row['rows']}, Extra: #{row['Extra']}"
      puts "---"
    end
  end

  def get_table_info(table_name)
    {
      columns: get_columns(table_name),
      indexes: get_indexes(table_name),
      size: get_table_size(table_name)
    }
  end

  def close
    @client.close
  end

  private

  def optimize_connection
    # Optimize MySQL settings for this connection
    @client.query("SET SESSION sql_mode = 'STRICT_TRANS_TABLES,NO_ZERO_DATE,NO_ZERO_IN_DATE,ERROR_FOR_DIVISION_BY_ZERO'")
    @client.query("SET SESSION innodb_lock_wait_timeout = 50")
    @client.query("SET SESSION query_cache_type = ON")
  end

  def get_count(sql)
    # Extract approximate count for pagination
    count_sql = sql.gsub(/SELECT.*FROM/i, 'SELECT COUNT(*) as count FROM')
                   .gsub(/ORDER BY.*$/i, '')
                   .gsub(/LIMIT.*$/i, '')

    result = @client.query(count_sql)
    result.first['count']
  end

  def get_columns(table_name)
    result = @client.query("DESCRIBE #{table_name}")
    result.map do |row|
      {
        name: row['Field'],
        type: row['Type'],
        null: row['Null'] == 'YES',
        default: row['Default'],
        key: row['Key']
      }
    end
  end

  def get_indexes(table_name)
    result = @client.query("SHOW INDEX FROM #{table_name}")
    result.group_by { |row| row['Key_name'] }.map do |key_name, rows|
      {
        name: key_name,
        unique: rows.first['Non_unique'] == '0',
        columns: rows.map { |row| row['Column_name'] }
      }
    end
  end

  def get_table_size(table_name)
    result = @client.query(<<~SQL)
      SELECT
        ROUND(((data_length + index_length) / 1024 / 1024), 2) AS 'size_mb',
        table_rows
      FROM information_schema.tables
      WHERE table_schema = DATABASE() AND table_name = '#{table_name}'
    SQL

    result.first
  end
end

# Usage
mysql = OptimizedMySQL.new(mysql_config)

# Bulk insert example
users_data = [
  ['User 1', 'user1@example.com', 25],
  ['User 2', 'user2@example.com', 30],
  ['User 3', 'user3@example.com', 35]
]

mysql.bulk_insert('users', ['name', 'email', 'age'], users_data)

# Paginated queries
page1 = mysql.paginated_query("SELECT * FROM users ORDER BY created_at DESC", page: 1, per_page: 10)
puts "Page 1: #{page1[:data].length} users, Total: #{page1[:total]}"

# Query analysis
mysql.explain_query("SELECT * FROM users WHERE email = 'alice@example.com'")

# Table information
info = mysql.get_table_info('users')
puts "Table size: #{info[:size]['size_mb']} MB"
puts "Row count: #{info[:size]['table_rows']}"

mysql.close

Database-Agnostic Abstraction Layer

Create a unified interface that works with both PostgreSQL and MySQL.

Universal Database Adapter

require 'pg'
require 'mysql2'

class DatabaseAdapter
  class << self
    def create(type, config)
      case type.to_sym
      when :postgresql
        PostgreSQLAdapter.new(config)
      when :mysql
        MySQLAdapter.new(config)
      else
        raise ArgumentError, "Unsupported database type: #{type}"
      end
    end
  end

  def initialize(config)
    @config = config
    @connection = nil
  end

  def connect
    @connection = create_connection
  end

  def disconnect
    @connection&.close
    @connection = nil
  end

  def connected?
    !@connection.nil? && connection_alive?
  end

  def execute(sql, params = [])
    raise NotImplementedError, "Subclasses must implement execute"
  end

  def execute_single(sql, params = [])
    result = execute(sql, params)
    extract_first_row(result)
  end

  def transaction
    raise NotImplementedError, "Subclasses must implement transaction"
  end

  def escape_identifier(name)
    "\"#{name}\""
  end

  def limit_clause(limit, offset = 0)
    "LIMIT #{limit}" + (offset > 0 ? " OFFSET #{offset}" : "")
  end

  protected

  def create_connection
    raise NotImplementedError, "Subclasses must implement create_connection"
  end

  def connection_alive?
    raise NotImplementedError, "Subclasses must implement connection_alive?"
  end

  def extract_first_row(result)
    raise NotImplementedError, "Subclasses must implement extract_first_row"
  end
end

class PostgreSQLAdapter < DatabaseAdapter
  def execute(sql, params = [])
    connect unless connected?

    if params.empty?
      @connection.exec(sql)
    else
      @connection.exec_params(sql, params)
    end
  end

  def transaction
    connect unless connected?

    @connection.transaction do
      yield self
    end
  end

  def escape_identifier(name)
    "\"#{name}\""
  end

  def placeholder(index)
    "$#{index + 1}"
  end

  protected

  def create_connection
    PG.connect(@config)
  end

  def connection_alive?
    begin
      @connection.exec('SELECT 1')
      true
    rescue PG::Error
      false
    end
  end

  def extract_first_row(result)
    row = result.first
    result.clear
    row
  end
end

class MySQLAdapter < DatabaseAdapter
  def execute(sql, params = [])
    connect unless connected?

    if params.empty?
      @connection.query(sql)
    else
      stmt = @connection.prepare(sql)
      result = stmt.execute(*params)
      stmt.close
      result
    end
  end

  def transaction
    connect unless connected?

    @connection.query("START TRANSACTION")
    begin
      result = yield self
      @connection.query("COMMIT")
      result
    rescue => e
      @connection.query("ROLLBACK")
      raise e
    end
  end

  def escape_identifier(name)
    "`#{name}`"
  end

  def placeholder(index)
    "?"
  end

  protected

  def create_connection
    Mysql2::Client.new(@config)
  end

  def connection_alive?
    @connection.ping
  end

  def extract_first_row(result)
    result.first
  end
end

# Usage example
class DatabaseManager
  def initialize(db_type, config)
    @adapter = DatabaseAdapter.create(db_type, config)
  end

  def create_users_table
    sql = <<~SQL
      CREATE TABLE IF NOT EXISTS users (
        id #{id_column},
        name VARCHAR(100) NOT NULL,
        email VARCHAR(255) UNIQUE NOT NULL,
        created_at #{timestamp_column}
      )
    SQL

    @adapter.execute(sql)
  end

  def insert_user(name, email)
    sql = "INSERT INTO users (name, email) VALUES (#{@adapter.placeholder(0)}, #{@adapter.placeholder(1)})"
    @adapter.execute(sql, [name, email])
  end

  def find_user_by_email(email)
    sql = "SELECT * FROM users WHERE email = #{@adapter.placeholder(0)}"
    @adapter.execute_single(sql, [email])
  end

  def find_users_paginated(page: 1, per_page: 10)
    offset = (page - 1) * per_page
    sql = "SELECT * FROM users ORDER BY created_at DESC #{@adapter.limit_clause(per_page, offset)}"
    @adapter.execute(sql)
  end

  def update_user_in_transaction(id, name, email)
    @adapter.transaction do |db|
      # Update user
      sql1 = "UPDATE users SET name = #{db.placeholder(0)}, email = #{db.placeholder(1)} WHERE id = #{db.placeholder(2)}"
      db.execute(sql1, [name, email, id])

      # Log the change
      sql2 = "INSERT INTO user_logs (user_id, action, timestamp) VALUES (#{db.placeholder(0)}, #{db.placeholder(1)}, #{db.placeholder(2)})"
      db.execute(sql2, [id, 'updated', Time.now])
    end
  end

  def close
    @adapter.disconnect
  end

  private

  def id_column
    case @adapter
    when PostgreSQLAdapter
      'SERIAL PRIMARY KEY'
    when MySQLAdapter
      'INT AUTO_INCREMENT PRIMARY KEY'
    end
  end

  def timestamp_column
    case @adapter
    when PostgreSQLAdapter
      'TIMESTAMP DEFAULT CURRENT_TIMESTAMP'
    when MySQLAdapter
      'TIMESTAMP DEFAULT CURRENT_TIMESTAMP'
    end
  end
end

# Example usage with different databases
pg_config = {
  host: 'localhost',
  port: 5432,
  dbname: 'myapp_development',
  user: 'postgres',
  password: 'password'
}

mysql_config = {
  host: 'localhost',
  port: 3306,
  database: 'myapp_development',
  username: 'root',
  password: 'password'
}

# Use PostgreSQL
pg_manager = DatabaseManager.new(:postgresql, pg_config)
pg_manager.create_users_table
pg_manager.insert_user('Alice', 'alice@example.com')

# Use MySQL
mysql_manager = DatabaseManager.new(:mysql, mysql_config)
mysql_manager.create_users_table
mysql_manager.insert_user('Bob', 'bob@example.com')

# Both work the same way!
alice = pg_manager.find_user_by_email('alice@example.com')
bob = mysql_manager.find_user_by_email('bob@example.com')

pg_manager.close
mysql_manager.close

Database Adapter Best Practices

Use connection pooling to manage database connections efficiently
Always use parameterized queries to prevent SQL injection attacks
Implement proper connection timeout and retry logic
Use transactions for operations that require consistency
Monitor connection health and implement reconnection logic
Set appropriate character encoding (UTF-8/UTF8MB4) for international support
Use prepared statements for frequently executed queries
Implement query logging and performance monitoring
Handle database-specific features gracefully in abstraction layers
Always close connections and prepared statements to avoid resource leaks

SQLite3 Gem Sequel ORM Basics

Quick Navigation

Read Topic

Watch Video Tutorial

Video Tutorial

Watch and learn postgresql & mysql adapters

Pro Tip: After reading through the content above, watch this video to reinforce your understanding and see the concepts in action!

Ruby Language

Rails Framework

Production Ready

PostgreSQL & MySQL Adapters