Class: DataDrain::Engine

Inherits:
Object
  • Object
show all
Defined in:
lib/data_drain/engine.rb

Overview

Motor principal de extracción y purga de datos (DataDrain).

Orquesta el flujo ETL desde PostgreSQL hacia un Data Lake analítico delegando la interacción del almacenamiento al adaptador configurado.

Instance Method Summary collapse

Constructor Details

#initialize(options) ⇒ Engine

Inicializa una nueva instancia del motor de extracción.

Parameters:

  • options (Hash) —

    Diccionario de configuración para la extracción.

Options Hash (options):

  • :start_date (Time, DateTime, Date) —

    Fecha y hora de inicio.

  • :end_date (Time, DateTime, Date) —

    Fecha y hora de fin.

  • :table_name (String) —

    Nombre de la tabla en PostgreSQL.

  • :folder_name (String) — default: Opcional —

    Nombre de la carpeta destino.

  • :select_sql (String) — default: Opcional —

    Sentencia SELECT personalizada.

  • :partition_keys (Array<String, Symbol>) —

    Columnas para particionar.

  • :primary_key (String) — default: Opcional —

    Clave primaria para borrado. Por defecto ‘id’.

  • :where_clause (String) — default: Opcional —

    Condición SQL extra.



23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
 
# File 'lib/data_drain/engine.rb', line 23

def initialize(options)
  @start_date     = options.fetch(:start_date).beginning_of_day
  @end_date       = options.fetch(:end_date).end_of_day
  @table_name     = options.fetch(:table_name)
  @folder_name    = options.fetch(:folder_name, @table_name)
  @select_sql     = options.fetch(:select_sql, "*")
  @partition_keys = options.fetch(:partition_keys)
  @primary_key    = options.fetch(:primary_key, "id")
  @where_clause   = options[:where_clause]
  @bucket         = options[:bucket]

  @config  = DataDrain.configuration
  @logger  = @config.logger
  @adapter = DataDrain::Storage.adapter

  database = DuckDB::Database.open(":memory:")
  @duckdb  = database.connect
end

Instance Method Details

#call ⇒ Boolean

Ejecuta el flujo completo del motor: Setup, Conteo, Exportación, Verificación y Purga.

Returns:

  • (Boolean) —

    true si el proceso finalizó con éxito, false si falló la integridad.



45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
 
# File 'lib/data_drain/engine.rb', line 45

def call
  @logger.info "[DataDrain Engine] 🚀 Preparando '#{@table_name}' (#{@start_date.to_date} a #{@end_date.to_date})..."

  setup_duckdb

  @pg_count = get_postgres_count

  if @pg_count.zero?
    @logger.info "[DataDrain Engine] ⏭️ No hay registros que cumplan las condiciones."
    return true
  end

  @logger.info "[DataDrain Engine] 📦 Exportando #{@pg_count} registros a Parquet..."
  export_to_parquet

  if verify_integrity
    purge_from_postgres
    @logger.info "[DataDrain Engine] ✅ Proceso completado exitosamente para '#{@table_name}'."
    true
  else
    @logger.error "[DataDrain Engine] ❌ ERROR de integridad en '#{@table_name}'. Abortando purga."
    false
  end
end