class Lutaml::Qea::Infrastructure::SchemaReader

columns = reader.columns(“t_object”)
tables = reader.tables
reader = SchemaReader.new(db_connection)
@example Read schema information
including table names, column definitions, and metadata.
This class is responsible for introspecting the database schema,
QEA SQLite database.
SchemaReader reads database schema information from a

def column_names(table_name)

Returns:
  • (Array) - List of column names

Parameters:
  • table_name (String) -- The table name 
def column_names(table_name)
  columns(table_name).map { |col| col["name"] }
end

def columns(table_name)

Returns:
  • (Array) - Array of column information hashes

Parameters:
  • table_name (String) -- The table name 
def columns(table_name)
  @database.execute("PRAGMA table_info(#{table_name})")
end

def indexes(table_name)

Returns:
  • (Array) - Array of index information

Parameters:
  • table_name (String) -- The table name 
def indexes(table_name)
  @database.execute(
    "SELECT name, sql FROM sqlite_master " \
    "WHERE type='index' AND tbl_name=?",
    table_name,
  )
end

def initialize(database)

Raises:
  • (ArgumentError) - if database is nil

Parameters:
  • database (SQLite3::Database) -- The database connection 
def initialize(database)
  raise ArgumentError, "database cannot be nil" if database.nil?
  @database = database
end

def primary_key(table_name)

Returns:
  • (String, nil) - The primary key column name,

Parameters:
  • table_name (String) -- The table name 
def primary_key(table_name)
  pk_column = columns(table_name).find { |col| col["pk"] == 1 }
  pk_column&.fetch("name", nil)
end

def row_count(table_name)

Returns:
  • (Integer) - Number of rows in the table

Parameters:
  • table_name (String) -- The table name 
def row_count(table_name)
  result = @database.execute(
    "SELECT COUNT(*) as count FROM #{table_name}",
  )
  result.first["count"]
end

def statistics

Returns:
  • (Hash) - Hash mapping table names to row counts 
def statistics
  tables.to_h do |table_name|
    [table_name, row_count(table_name)]
  end
end

def table_exists?(table_name)

Returns:
  • (Boolean) - true if table exists

Parameters:
  • table_name (String) -- The table name to check 
def table_exists?(table_name)
  result = @database.execute(
    "SELECT name FROM sqlite_master WHERE type='table' AND name=?",
    table_name,
  )
  !result.empty?
end

def table_schema(table_name)

Returns:
  • (String, nil) - The CREATE TABLE SQL statement,

Parameters:
  • table_name (String) -- The table name 
def table_schema(table_name)
  result = @database.execute(
    "SELECT sql FROM sqlite_master WHERE type='table' AND name=?",
    table_name,
  )
  result.first&.fetch("sql", nil)
end

def tables(exclude_system: true)

Returns:
  • (Array) - List of table names

Parameters:
  • exclude_system (Boolean) -- 
def tables(exclude_system: true)
  query = "SELECT name FROM sqlite_master WHERE type='table'"
  query += " AND name NOT LIKE 'sqlite_%'" if exclude_system
  query += " ORDER BY name"
  @database.execute(query).map { |row| row["name"] }
end