File Coverage

blib/lib/Footprintless/Plugin/Database/AbstractProvider.pm

Criterion	Covered	Total	%
statement	110	159	69.1
branch	24	56	42.8
condition	2	13	15.3
subroutine	25	34	73.5
pod	14	15	93.3
total	175	277	63.1

use strict;
use warnings;

package Footprintless::Plugin::Database::AbstractProvider;
$Footprintless::Plugin::Database::AbstractProvider::VERSION = '1.01';
# ABSTRACT: A base class for database providers
# PODNAME: Footprintless::Plugin::Database::AbstractProvider

use parent qw(Footprintless::MixableBase);

use overload q{""} => 'to_string', fallback => 1;

use Carp;
use DBI;
use Footprintless::Mixins qw(
    _entity
);
use Log::Any;

my $logger = Log::Any->get_logger();

sub backup {
    die("abstract method invocation");
}

sub begin_transaction {
    my ($self) = @_;
    croak("not connected") unless ( $self->{connection} );

    $self->{connection}->begin_work();

    return $self;
}

sub client {
    my ( $self, %options ) = @_;

    my $in_file;
    eval {
        my $in_handle = delete( $options{in_handle} );
        if ( $options{in_file} ) {
            open( $in_file, '<', delete( $options{in_file} ) )
                || croak("invalid in_file: $!");
        }
        if ( $options{in_string} ) {
            my $string = delete( $options{in_string} );
            open( $in_file, '<', \$string )
                || croak("invalid in_string: $!");
        }
        $self->_connect_tunnel();

        my $local_in = $in_handle || $in_file;
        local (*STDIN) = $local_in if ($local_in);

        require Footprintless::Plugin::Database::SqlShellAdapter;
        Footprintless::Plugin::Database::SqlShellAdapter::sql_shell( $self->_connection_string(),
            $self->{username}, $self->{password}, @{ $options{client_options} } );
    };
    my $error = $@;
    $self->disconnect();
    if ($in_file) {
        close($in_file);
    }

    croak($error) if ($error);
}

sub commit_transaction {
    my ($self) = @_;
    croak("not connected") unless ( $self->{connection} );

    $self->{connection}->commit();

    return $self;
}

sub connect {
    my ($self) = @_;

    return if ( $self->{connection} );

    $self->_connect_tunnel();

    my ( $hostname, $port ) = $self->_hostname_port();

    $logger->debugf( 'connecting to %s', $self->to_string() );
    $self->{connection} = DBI->connect( $self->_connection_string(),
        $self->{username}, $self->{password}, { RaiseError => 1, AutoCommit => 1 } )
        || croak("unable to connect to $hostname on port $port: $@");
    $logger->tracef('connected');

    return $self;
}

sub _connect_tunnel {
    my ($self) = @_;

    return if ( $self->{tunnel} );

    if ( $self->{tunnel_hostname} ) {
        $logger->debugf( 'opening tunnel through %s', $self->{tunnel_hostname} );
        $self->{tunnel} = $self->{factory}->tunnel(
            $self->{coordinate},
            destination_hostname => $self->{tunnel_destination_hostname} || $self->{hostname},
            destination_port => $self->{port}
        );
        $self->{tunnel}->open();
    }

    return $self;
}

sub _connection_string {
    die("abstract method invocation");
}

sub DESTROY {
    my ($self) = @_;
    $self->disconnect();
}

sub disconnect {
    my ($self) = @_;

    if ( $self->{connection} ) {
        $logger->debugf( 'disconnecting from %s', $self->to_string() );
        $self->{connection}->disconnect();
        delete( $self->{connection} );
    }

    if ( $self->{tunnel} ) {
        $logger->debug('closing tunnel');
        $self->{tunnel}->close();
        delete( $self->{tunnel} );
    }

    return $self;
}

sub execute {
    my ( $self, $query ) = @_;

    my $result;
    $self->_process_sql(
        $query,
        sub {
            $result = $_[1];
        }
    );
    return $result;
}

sub get_schema {
    return $_[0]->{schema};
}

sub _hostname_port {
    my ($self) = @_;

    my ( $hostname, $port );
    if ( $self->{tunnel} ) {
        $hostname = $self->{tunnel}->get_local_hostname() || 'localhost';
        $port = $self->{tunnel}->get_local_port();
    }
    else {
        $hostname = $self->{hostname};
        $port     = $self->{port};
    }

    return ( $hostname eq 'localhost' ? '127.0.0.1' : $hostname, $port );
}

sub _init {
    my ( $self, %options ) = @_;

    my $entity = $self->_entity( $self->{coordinate} );

    $self->{backup}                      = $entity->{backup};
    $self->{database}                    = $entity->{database};
    $self->{hostname}                    = $entity->{hostname} || 'localhost';
    $self->{password}                    = $entity->{password};
    $self->{port}                        = $entity->{port};
    $self->{schema}                      = $entity->{schema};
    $self->{tunnel_destination_hostname} = $entity->{tunnel_destination_hostname};
    $self->{tunnel_hostname}             = $entity->{tunnel_hostname};
    $self->{tunnel_username}             = $entity->{tunnel_username};
    $self->{username}                    = $entity->{username};

    return $self;
}

sub _process_sql {
    my ( $self, $query, $statement_handler ) = @_;

    my ( $sql, $parameters ) =
        ref($query) eq 'HASH'
        ? ( $query->{sql}, $query->{parameters} )
        : ($query);
    eval {
        if ( $logger->is_trace() ) {
            $logger->trace(
                "$self->{hostname}: '$sql'",
                (   $parameters
                    ? ( ",[" . join( ',', @$parameters ) . "]" )
                    : ''
                )
            );
        }
        my $statement_handle = $self->{connection}->prepare_cached($sql);
        &{$statement_handler}(
            $statement_handle,
            (   defined($parameters)
                ? $statement_handle->execute(@$parameters)
                : $statement_handle->execute()
            )
        );
        $statement_handle->finish();
    };
    if ($@) {
        croak("query failed: $@");
    }
}

sub query {
    my ( $self, $query, $result_handler ) = @_;

    $self->_process_sql(
        $query,
        sub {
            my ( $statement_handle, $execute_result ) = @_;

            while ( my @row = $statement_handle->fetchrow_array() ) {
                &{$result_handler}(@row);
            }
        }
    );
    return;
}

sub query_for_list {
    my ( $self, $query, $row_mapper ) = @_;

    my @results = ();
    $self->query(
        $query,
        sub {
            if ($row_mapper) {
                push( @results, &{$row_mapper}(@_) );
            }
            else {
                push( @results, \@_ );
            }
        }
    );
    return wantarray() ? @results : \@results;
}

sub query_for_map {
    my ( $self, $query, $row_mapper ) = @_;

    my %results = ();
    $self->query(
        $query,
        sub {
            if ($row_mapper) {
                my $key_value_pair = &{$row_mapper}(@_);
                $results{ $key_value_pair->[0] } = $key_value_pair->[1];
            }
            else {
                $results{ $_[0] } = \@_;
            }
        }
    );
    return wantarray() ? %results : \%results;
}

sub query_for_scalar {
    my ( $self, $query, $row_mapper ) = @_;

    my $result;
    $self->_process_sql(
        $query,
        sub {
            my ( $statement_handle, $execute_result ) = @_;
            if ( my @row = $statement_handle->fetchrow_array() ) {
                if ($row_mapper) {
                    $result = &$row_mapper(@row);
                }
                else {
                    $result = $row[0];
                }
            }
        }
    );
    return $result;
}

sub restore {
    die("abstract method invocation");
}

sub rollback_transaction {
    my ($self) = @_;
    croak("not connected") unless ( $self->{connection} );

    $self->{connection}->rollback();

    return $self;
}

sub to_string {
    my ($self) = @_;
    return "{schema=>'$self->{schema}',hostname=>'$self->{hostname}',port=>$self->{port}}";
}

1;

__END__