File Coverage

lib/Data/Record/Serialize/Encode/dbi.pm

Criterion	Covered	Total	%
statement	133	182	73.0
branch	39	78	50.0
condition	10	23	43.4
subroutine	26	32	81.2
pod	5	6	83.3
total	213	321	66.3

package Data::Record::Serialize::Encode::dbi;

# ABSTRACT:  store a record in a database

use v5.12;
use Moo::Role;

use Data::Record::Serialize::Error {
    errors => [
        qw( param
          connect
          schema
          create
          insert
          sqlite_backend
        ),
    ],
  },
  -all;

our $VERSION = '1.08';

use Data::Record::Serialize::Types -types;

use SQL::Translator;
use SQL::Translator::Schema;
use Types::Standard -types;
use Types::Common::String qw( NonEmptySimpleStr );

use List::Util 1.33 qw[ pairmap any ];

use DBI;

use namespace::clean;







has dsn => (
    is       => 'ro',
    required => 1,
    coerce   => sub {
        my $arg = 'ARRAY' eq ref $_[0] ? $_[0] : [ $_[0] ];
        my @dsn;
        my @args;
        for my $el ( @{$arg} ) {

            my $ref = ref $el;
            unless ( $ref eq 'ARRAY' || $ref eq 'HASH' ) {
                push( @dsn, $el );
                next;
            }

            my @arr = $ref eq 'ARRAY' ? @{$el} : %{$el};

            push @args, pairmap { join( q{=}, $a, $b ) } @arr;
        }
        my $args = join( q{;}, @args );
        push @dsn, $args if length $args;
        unshift @dsn, 'dbi' unless $dsn[0] =~ /^dbi/;
        return join( q{:}, @dsn );
    },
);

has _cached => (
    is       => 'ro',
    default  => 0,
    init_arg => 'cached',
);








has table => (
    is       => 'ro',
    isa      => Str,
    required => 1,
);











has schema => (
    is  => 'ro',
    isa => Maybe [NonEmptySimpleStr],
);







has drop_table => (
    is      => 'ro',
    isa     => Bool,
    default => 0,
);







has create_output_dir => (
    is      => 'ro',
    isa     => Bool,
    default => 0,
);









has create_table => (
    is      => 'ro',
    isa     => Bool,
    default => 1,
);







has primary => (
    is      => 'ro',
    isa     => ArrayOfStr,
    coerce  => 1,
    default => sub { [] },
);







has db_user => (
    is      => 'ro',
    isa     => Str,
    default => q{},
);







has db_pass => (
    is      => 'ro',
    isa     => Str,
    default => q{},
);

has _sth => (
    is       => 'rwp',
    init_arg => undef,
);

has _dbh => (
    is        => 'rwp',
    init_arg  => undef,
    clearer   => 1,
    predicate => 1,
);







has column_defs => (
    is       => 'rwp',
    lazy     => 1,
    clearer  => 1,
    init_arg => undef,
    builder  => sub {
        my $self = shift;

        my @column_defs;
        for my $field ( @{ $self->output_fields } ) {
            push @column_defs,
              join( q{ },
                $field,
                $self->output_types->{$field},
                ( 'primary key' )x!!( $self->primary eq $field ) );
        }

        return join ', ', @column_defs;
    },
);







has batch => (
    is      => 'ro',
    isa     => Int,
    default => 100,
    coerce  => sub { $_[0] > 1 ? $_[0] : 0 },
);







has dbitrace => ( is => 'ro', );













has queue => (
    is       => 'ro',
    init_arg => undef,
    default  => sub { [] },
);







has quote_identifiers => (
    is      => 'ro',
    isa     => Bool,
    default => !!0,
);

around '_build__nullified' => sub {
    my $orig = shift;
    my $self = $_[0];

    my $nullified = $self->$orig( @_ );

    # defer to the caller
    return $nullified if $self->has_nullify;

    # add all of the numeric fields
    [ @{ $self->numeric_fields } ];

};

my %MapTypes = (
    Pg      => { S => 'text', N => 'real', I => 'integer', B => 'boolean' },
    SQLite  => { S => 'text', N => 'real', I => 'integer', B => 'integer' },
    Default => { S => 'text', N => 'real', I => 'integer', B => 'integer' },
);

sub _map_types {
    $MapTypes{ $_[0]->_dbi_driver } // $MapTypes{Default};
}









sub to_bool { $_[0] ? 1 : 0 }

sub _table_exists {
    my $self = shift;

    # DBD::Sybase doesn't filter, so need to search
    my $matches
      = $self->_dbh->table_info( q{%}, $self->schema, $self->table, 'TABLE' )->fetchall_arrayref;

    # if $self->table needs to be quoted, table_info returns a quoted
    # string (at least for DBD::Pg), so need an extra check...
    return
      any { $_->[2] eq $self->table or $_->[2] eq $self->_dbh->quote_identifier( $self->table ) }
      @{$matches};
}

sub _fq_table_name {
    my $self = shift;
    defined $self->schema ? $self->schema . q{.} . $self->table : $self->table;
}

has _dsn_components => (
    is       => 'lazy',
    init_arg => undef,
    builder  => sub {
        my %dsn;
        @dsn{ 'scheme', 'driver', 'attr_string', 'attr_hash', 'driver_dsn' } = DBI->parse_dsn( $_[0]->dsn )
          or error( 'param', 'unable to parse DSN: ', $_[0]->dsn );

        my @driver_dsn = split( /;/, $dsn{driver_dsn} );
        my %driver_dsn = map { split( /=/, $_, 2 ) } @driver_dsn;
        $dsn{driver_dsn} = \%driver_dsn;
        \%dsn;
    },
);

sub _dbi_driver {
    $_[0]->_dsn_components->{driver};
}

my %producer = (
    DB2       => 'DB2',
    MySQL     => 'mysql',
    Oracle    => 'Oracle',
    Pg        => 'PostgreSQL',
    SQLServer => 'SQLServer',
    SQLite    => 'SQLite',
    Sybase    => 'Sybase',
);

has _producer => (
    is       => 'lazy',
    init_arg => undef,
    builder  => sub {
        my $dbi_driver = $_[0]->_dbi_driver;
        $producer{$dbi_driver} || $dbi_driver;
    },
);

has _is_file_based => (
    is       => 'lazy',
    init_arg => undef,
    builder  => sub {
        my $dbi_driver = $_[0]->_dbi_driver;
        return $dbi_driver eq 'SQLite';
    },
);







sub setup {    ## no critic (Subroutines::ProhibitExcessComplexity)
    my $self = shift;

    return if $self->_has_dbh;

    my %attr = (
        AutoCommit                   => !$self->batch,
        RaiseError                   => 1,
        PrintError                   => 0,
        ( 'private_' . __PACKAGE__ ) => __FILE__ . __LINE__,
    );


    my $dbd = $self->_dbi_driver;

    if ( $dbd eq 'Sybase' ) {
        $attr{syb_quoted_identifier} = 1;
    }

    my $connect = $self->_cached ? 'connect_cached' : 'connect';

    if ( $self->create_output_dir && $self->_is_file_based ) {
        my $dbname = $self->_dsn_components->{driver_dsn}{dbname}
          // error( 'param', "unable to parse dbname from DSN (@{[ $self->dsn ]}) to create output dir" );

        require Path::Tiny;
        my $dir = Path::Tiny::path( $dbname )->parent;
        eval { $dir->mkdir; } or error( '::create', "unable to create output directory '$dir': $@" );
    }

    $self->_set__dbh( DBI->$connect( $self->dsn, $self->db_user, $self->db_pass, \%attr ) )
      or error( 'connect', 'error connecting to ', $self->dsn, "\n" );

    $self->_dbh->trace( $self->dbitrace )
      if $self->dbitrace;

    if ( $self->drop_table || ( $self->create_table && !( my $table_exists = $self->_table_exists ) ) )
    {
        # in case the first half of the conditional in the above if
        # statement is true, the call to $self->_table_exists isn't
        # made, so $table_exists is not defined.
        $table_exists //= $self->_table_exists;

        ## no critic (Variables::ProhibitUnusedVarsStricter)
        my @guards = $dbd eq 'Sybase' ? _patch_sqlt_producer_sybase() : ();

        my $tr = SQL::Translator->new(
            from => sub {
                my $schema = $_[0]->schema;
                my $table  = $schema->add_table( name => $self->_fq_table_name )
                  or error( 'schema', $schema->error );

                for my $field_name ( @{ $self->output_fields } ) {
                    $table->add_field(
                        name      => $field_name,
                        data_type => $self->output_types->{$field_name} ) or error( 'schema', $table->error );
                }

                if ( @{ $self->primary } ) {
                    $table->primary_key( @{ $self->primary } )
                      or error( 'schema', $table->error );
                }

                1;
            },
            to                => $self->_producer,
            producer_args     => { no_transaction => 1 },
            add_drop_table    => $self->drop_table && $table_exists,
            quote_identifiers => $self->quote_identifiers,
            no_comments       => 1,
        );

        my @sql = $tr->translate;
        defined $sql[0] or error( 'schema', $tr->error );

        _do_sql( $self->_dbh, 'create', @sql );
        $self->_dbh->commit if $self->batch;
    }

    my $sql = sprintf(
        'insert into %s (%s) values (%s)',
        $self->_dbh->quote_identifier( undef, $self->schema, $self->table ),
        join( q{,}, @{ $self->output_fields } ),
        join( q{,}, ( q{?} ) x @{ $self->output_fields } ),
    );

    $self->_set__sth( $self->_dbh->prepare( $sql ) );

    return;
}

# SQL::Translator::Producer::Sybase has behaviors and bugs.
sub _patch_sqlt_producer_sybase {

    require SQL::Translator::Producer::Sybase;
    require Monkey::Patch;
    my %scope;

    my @guards;

    # Behavior: It tries to be 'helpful' and renames a table if it
    # collides with a table it has already seen.  There's no way to
    # turn that off. In this case, we may dropping/creating a table
    # multiple times, and having the table name change is not useful.
    # We turn its concept of "global name space" into a local one for
    # tables/indices/constraints. args[2] is the namespace hash, and
    # mk_name is currently passed a non-empty entry *only* for fields,
    # so we make sure to only give it a hash when its not specified
    # (and thus not a field), as we don't want to mix field names and
    # the table/indices/constraints level names.

    push @guards, Monkey::Patch::patch_package(
        'SQL::Translator::Producer::Sybase',
        'mk_name',
        sub {
            my ( $orig, @args ) = @_;
            $args[2] ||= \%scope;
            $orig->( @args );
        } );

    # Bug: add_drop_table attribute doesn't add the drop table
    # command.  future proof by chcking if the Sybase producer
    # implements it correctly and only monkey-patching it if it
    # doesn't.
    my $need_to_add_drop_table = do {
        my $tr = SQL::Translator->new(
            from => sub {
                require SQL::Translator::Schema::Table;
                my $table = SQL::Translator::Schema::Table->new( name => 'foo' );
                $table->add_field( name => 'foo_id', date_type => 'integer' );
                $_[0]->schema->add_table( $table );
                1;
            },
            to             => 'Sybase',
            add_drop_table => 1,
            no_comments    => 1,
        );
        $tr->translate !~ /DROP TABLE/;
    };

    if ( $need_to_add_drop_table ) {
        push @guards, Monkey::Patch::patch_package(
            'SQL::Translator::Producer::Sybase',
            'produce',
            sub {
                my ( $orig, @args ) = @_;

                my $translator = $args[0];

                return $orig->( @args ) unless $translator->add_drop_table;

                if ( wantarray ) {    ## no critic (Community::Wantarray)
                    my @orig = $orig->( @args );
                    return map { /CREATE TABLE (\S+) [(]/ ? ( qq[DROP TABLE $1\n\n], $_ ) : $_ } @orig;
                }
                else {
                    my $output = $orig->( @args );
                    $output =~ s/CREATE TABLE\s+(\S+)\s+[(]/DROP TABLE $1\n$&/g;
                    return $output;
                }

            } );

    }

    return @guards;
}

sub _do_sql {
    my ( $dbh, $subsys, @sql ) = @_;

    my $in_txn = !$dbh->{AutoCommit};

    my $statement;
    eval {
        $dbh->begin_work unless $in_txn;
        while ( $statement = shift @sql ) {
            die if !defined $dbh->do( $statement );
        }
        $dbh->commit unless $in_txn;
        1;
    } or do {
        my $e = $@;
        $dbh->rollback unless $in_txn;
        $statement =~ s/\n/ /g;
        error( $subsys, { msg => $e, payload => $statement } );
    };

}




















































sub flush {
    my $self = shift;

    return 1 unless $self->_has_dbh;

    my $queue = $self->queue;

    if ( @{$queue} ) {
        my $last;

        my $ret   = eval { $self->_sth->execute( @$last ) while $last = shift @{$queue}; 1; };
        my $error = $@;

        $self->_dbh->commit;

        if ( !defined $ret ) {
            unshift @{$queue}, $last;

            my %query;
            @query{ @{ $self->output_fields } } = @$last;
            error( 'insert', { msg => "Transaction aborted: $error", payload => \%query } );
        }
    }

    1;
}


















sub send {
    my $self = shift;

    if ( $self->batch ) {
        push @{ $self->queue }, [ @{ $_[0] }{ @{ $self->output_fields } } ];
        $self->flush
          if @{ $self->queue } == $self->batch;

    }
    else {
        eval { $self->_sth->execute( @{ $_[0] }{ @{ $self->output_fields } } ); 1; }
          or error( 'insert', { msg => "record insertion failed: $@", payload => $_[0] } );
    }
}


after '_trigger_output_fields' => sub {
    $_[0]->clear_column_defs;
};

after '_trigger_output_types' => sub {
    $_[0]->clear_column_defs;
};





















sub close {
    my $self = shift;

    return 1 unless $self->_has_dbh;

    $self->flush if $self->batch;
    $self->_dbh->disconnect;
    $self->_clear_dbh;

    1;
}












sub DEMOLISH {
    my $self = shift;

    warnings::warnif( 'Data::Record::Serialize::Encode::dbi::queue',
        __PACKAGE__ . ': record queue is not empty in object destruction' )
      if @{ $self->queue };

    $self->_dbh->disconnect
      if $self->_has_dbh;

}


# these are required by the Sink/Encode interfaces but should never be
# called in the ordinary run of things.








with 'Data::Record::Serialize::Role::EncodeAndSink';

1;

#
# This file is part of Data-Record-Serialize-Encode-dbi
#
# This software is Copyright (c) 2017 by Smithsonian Astrophysical Observatory.
#
# This is free software, licensed under:
#
#   The GNU General Public License, Version 3, June 2007
#

__END__