File Coverage

blib/lib/DBIx/Brev.pm

Criterion	Covered	Total	%
statement	66	184	35.8
branch	18	108	16.6
condition	8	39	20.5
subroutine	14	30	46.6
pod	9	16	56.2
total	115	377	30.5

package DBIx::Brev;

use strict;
use warnings;

use DBI;

our $use_config  = eval q{use Config::General;1};
our $use_sqlsplit = eval q{use SQL::SplitStatement;1};
our $use_connector = eval q{use DBIx::Connector;1};

use Scalar::Util qw(looks_like_number);

our $VERSION = '0.02';

use base 'Exporter';

our @EXPORT = our @EXPORT_OK = qw(
sql_exec
sql_value
sql_query
sql_map
sql_hash
sql_query_hash
sql_in
inserts
db_use
dbc
quote
);

my ($dbh,$dbc);

sub dbc {
    $dbc = $_[0] if @_;
    $dbh = $dbc unless $use_connector;
    $dbc;
}

sub dbh {
    $dbh = $_[0] if @_;
    $dbc = $dbh unless $use_connector;
    $dbh;
}

sub shift_params(&\@) {
    my ($predicate,$params) = @_;
    local $_ = shift(@$params);
    my $p = $_;
    if (eval {$predicate->($_) && 1}) {
        return $p;
    } else {
        unshift @$params,$p;
        return;
    }
}

my %config;
{
my $config_loaded;
sub load_config {
    #return if $config_loaded; $config_loaded = 1;
    return unless $use_config;
    my $config = shift_params {ref($_) eq 'HASH'} @_;
    if ($config) {
         %config = %$config;
        return;
    }
    my $mswin = $^O eq 'MSWin32';
    my @path = $mswin?map( {exists $ENV{$_}?$ENV{$_}:()} qw(
        USERPROFILE HOME ALLUSERSPROFILE APPDATA ProgramData SYSTEMROOT WINDIR
    )) : ($ENV{HOME},'/etc');
    my $fd = $mswin?q{\\}:q{/};
    my ($config_file) = grep defined && -f,@_,$ENV{DBI_CONF},map $_.$fd.q{dbi.conf},@path;
    %config = Config::General->new($config_file)->getall if $config_file;
}
}

my %dbc; # cache of dbx connections for fast switching between handles
sub db_use {
    my ($db_alias,%options) = @_;
    my @connect = ($db_alias,@options{qw(username password)});
    my $options = $options{options};
    $options ||= {RaiseError => 1,AutoCommit => 1};
    # subroutine changes default dbc if it is called in void context or $dbc is undefined
    my $keep_default = $dbc && defined(wantarray);
    my $connection_mode = delete $options->{connection_mode} || 'fixup';
    my ($local_dbc) = $dbc{$db_alias};
    unless ($local_dbc) {
        load_config() if $use_config && keys(%config)==0;
        my ($alias,$mode) = split /:/,$db_alias;
        my $databases = $config{database};
        die "wrong config" unless $databases;
        my @keys = qw(data_source username password);
        @connect = @{$databases->{$alias}}{map $mode."_$_",@keys} if exists $databases->{$alias} && $mode;
        @connect = @{$databases->{$db_alias}}{@keys} if exists $databases->{$db_alias};
        push @connect,$options;
        $local_dbc = $use_connector?DBIx::Connector->new(@connect):DBI->connect(@connect);
        $local_dbc->mode($connection_mode) if $use_connector;
        $dbc{$db_alias} = $dbc;
    }
    return $local_dbc if $keep_default;
    dbc($local_dbc);
}

sub quote {
    die "connect to db first!" unless $dbc;
    ($use_connector?$dbc->dbh():$dbh)->quote(@_);
}

sub import { my ($class,$db_alias,@options) = @_;
    db_use($db_alias,@options) if $db_alias;
    $class->export_to_level(1);
}

sub shift_connection(\@) {
    my $c = shift;
    my $local_dbc = shift_params {UNIVERSAL::isa($_,$use_connector?'DBIx::Connector':'DBI::db')} @$c;
    unless ($local_dbc) {
        my $db_alias = shift_params {!ref($_) && !m{\s}} @$c;
        if ($db_alias) {
            my $db_options = shift_params {ref($_) eq 'HASH'} @$c;
            $local_dbc = db_use($db_alias,$db_options);
        }
    }
    return $local_dbc || $dbc;
}

sub get_sth
{
    my $dbc = shift_connection(@_);
    my ($sql,@bind_values) = @_;
    my $executed;
    my $sth;
    if ($use_connector) {
        $dbc->run(sub {$executed = ($sth = $_->prepare($sql)) && $sth->execute(@bind_values);});
    } else {
        $executed = ($sth = $dbc->prepare($sql)) && $sth->execute(@bind_values);
    }
    my $err = $@ or $sth->errstr;
    die "$err\n[$sql]" if $err;
    unless  ($executed) {
        die "Error:$DBI::errstr\nSQL::$sql\n";
    };
    return $sth;
}

sub sql_query
{
    my $sth = &get_sth;
    my $r = $sth->fetchall_arrayref;
    if (@$r && (@{$r->[0]}==1)) {
        $_ = $_->[0] for @$r; #scalarize rows if row is single-dimension array
    }
    return wantarray?@$r:$r;
}

sub sql_in {
    my $dbc = shift_connection(@_);
    my $dbh = $use_connector?$dbc->dbh():$dbc;
    my $list = join ",",map {looks_like_number($_)?$_:$dbh->quote($_)} sql_query($dbc,@_);
    sprintf(" in (%s)",$list || 'NULL');
}

sub sql_value
{
    my $sth = &get_sth;
    my @row_ary  = $sth->fetchrow_array;
    wantarray()?@row_ary:$row_ary[0];
}

sub sql_hash
{
    my $sth = &get_sth;
    my $hash  = $sth->fetchrow_hashref;
    return undef unless $hash;
    wantarray()?%$hash:$hash;
}

sub sql_query_hash
{
    my $sth = &get_sth;
    my @result;
    while (my $row  = $sth->fetchrow_hashref) { push @result,$row; }
    wantarray()?@result:\@result;
}

sub sql_map(&@)
{
    my $callback = shift;
    my $sth = get_sth(@_);
    my @result = ();
    my $wantresult = defined wantarray;
    local $_;
    while (defined($_ = $sth->fetch)) {
        # do copy because fetch uses the same buffer
        $_ = @$_>1?[@$_]:$_->[0];
        my @r = $callback->($_);
        last unless @r;
        if ($wantresult) {
            push @result, @r;
        }
    }
    return unless $wantresult;
    return wantarray?@result:\@result;
}

sub sql_exec
{
    my $dbc = shift_connection(@_);
    my ($attr) = grep ref($_) eq 'HASH', @_,{};
    my ($sql_code,@bind_values) = grep ref($_) ne 'HASH', @_;
    my $rows_affected = 0;
    my $dbh = $use_connector?$dbc->dbh():$dbc;
    if ($use_sqlsplit) {
        my $splitter_options = delete $attr->{splitter_options}||{};
        my $no_commit = delete $attr->{no_commit};
        my $splitter = SQL::SplitStatement->new($splitter_options);
        my ( $statements, $placeholders ) = $splitter->split_with_placeholders( $sql_code );
        $dbh->{AutoCommit} = 0;  # enable transactions, if possible
        $dbh->{RaiseError} = 1;
        die $@ unless eval {
            for my $statement (@$statements) {
                my $placeholders_count = shift(@$placeholders);
                my @sbind_values = splice @bind_values, 0, $placeholders_count;
                $rows_affected += $dbh->do($statement,$attr,@sbind_values);
            }
            $dbh->commit unless $no_commit;
            1;
        };
    } else {
        $rows_affected += $dbh->do($sql_code,$attr,@bind_values);
    }
    return $rows_affected;
}


sub inserts {
    my $dbc = shift_connection(@_);
    my ($sprintf_sql,$data,%opts) = @_;
    $sprintf_sql .= ' %s' unless $sprintf_sql =~ m{%s};
    # set default step 500 for smooth sqlite experience (see http://stackoverflow.com/questions/15858466/limit-on-multiple-rows-insert)
    my $step = $opts{step} || 500; # split batch into bunches of $step records
    my $delay = ($opts{delay} || 0) / 1000; # delay between statements in milliseconds
    my $last = $#$data;
    my $offset = 0;
    my $updated = 0;
    # make fixup for dbh
    #$dbc->run(fixup => sub {$_->do(q{SELECT 1});});
    # do all inserts in one transaction
    my $dbh = $use_connector?$dbc->dbh():$dbc;
    $dbh->{AutoCommit} = 0;  # enable transactions, if possible
    $dbh->{RaiseError} = 1;
    die $@ unless eval {
        while ($offset <= $last) {
            my $limit = $offset + $step - 1;
            $limit = $last if $limit > $last;
            my $records = join "\nunion all\n",
                map sprintf("select %s",
                    ref($_) eq 'ARRAY'?
                    join(",", map $dbh->quote($_), @$_)
                    :
                    $_
                ), @{$data}[$offset..$limit];
            $offset += $step;
            sleep($delay) if $updated;
            my $time = time;
            my $sql = sprintf($sprintf_sql,$records);
            $updated += $dbh->do($sql);
            # make next delay twice time of sql_exec if it was not explicitly specified
            $delay = (time - $time) * 2 unless exists $opts{delay};
        }
        $dbh->commit;
        1;
    };
    return $updated;
}


1;
__END__