File Coverage

blib/lib/Mojo/mysql/Database/Role/LoadDataInfile.pm

Criterion	Covered	Total	%
statement	27	122	22.1
branch	7	70	10.0
condition	3	61	4.9
subroutine	6	20	30.0
pod	2	2	100.0
total	45	275	16.3

package Mojo::mysql::Database::Role::LoadDataInfile;
use Mojo::Base -role;
use Mojo::File 'tempfile';
use Mojo::Util ();

our $VERSION = '0.01';

my $text_csv_package;
BEGIN {
    if (not $ENV{MOJO_MYSQL_DATABASE_ROLE_LOAD_DATA_INFILE_NO_XS} and eval { require Text::CSV_XS; 1 }) {
        Text::CSV_XS->import('csv');
        $text_csv_package = 'Text::CSV_XS';
    } else {
        require Text::CSV_PP;
        Text::CSV_PP->import('csv');
        $text_csv_package = 'Text::CSV_PP';
    }
}

sub import {
    my $class = shift;
    if (grep { $_ eq '-no_apply' } @_) {
        return if @_ == 1;
        Carp::croak 'no other options may be provided with -no_apply';
    }

    my %options = @_;
    my $database_class = delete $options{database_class} || 'Mojo::mysql::Database';

    Carp::croak 'unknown options provided to import: ' . Mojo::Util::dumper(\%options) if %options;

    require Role::Tiny;
    Role::Tiny->apply_roles_to_package($database_class, 'Mojo::mysql::Database::Role::LoadDataInfile');
}

sub load_data_infile {
    my $self = shift;
    my $cb = ref $_[-1] eq 'CODE' ? pop : undef;

    my (
        $low_priority,
        $concurrent,
        $replace,
        $ignore,
        $table,
        $partition,
        $character_set,
        $tempfile_open_mode,
        $set,
        $rows,
        $columns,
        $headers) = _parse_options(@_);

    my $tempfile = _write_temp_file($tempfile_open_mode, $rows, $headers);

    my $query = _build_query($low_priority, $concurrent, $tempfile, $replace, $ignore, $table, $partition, $character_set, $columns, $set);

    if ($cb) {
        my $cb_wrapper = sub {
            no warnings 'void';
            $tempfile;
            $cb->(@_);
        };

        return $self->query($query, $cb_wrapper);
    } else {
        return $self->query($query);
    }
}

sub load_data_infile_p {
    my $promise = Mojo::Promise->new;

    shift->load_data_infile(@_ => sub { $_[1] ? $promise->reject($_[1]) : $promise->resolve($_[2]) });

    return $promise;
}

sub _write_temp_file {
    my ($tempfile_open_mode, $rows, $headers) = @_;

    my $tempfile = tempfile();
    my $temp_fh = $tempfile->open($tempfile_open_mode);
    csv(in => $rows, out => $temp_fh, sep_char => "\t", quote_char => q{"}, eol => "\n", headers => $headers) or Carp::croak $text_csv_package->error_diag;
    close $temp_fh;

    return $tempfile;
}

sub _parse_options {
    my %options = @_;

    my @parsed_options = (
        _parse_low_priority_and_concurrent(\%options),
        _parse_replace_and_ignore(\%options),
        _parse_table(\%options),
        _parse_partition(\%options),
        _parse_character_set_and_tempfile_open_mode(\%options),
        _parse_set(\%options),
        _parse_rows_and_columns_and_headers(\%options),
    );

    Carp::croak 'unknown options provided: ' . Mojo::Util::dumper(\%options) if %options;

    return @parsed_options;
}

sub _parse_low_priority_and_concurrent {
    my ($options) = @_;

    if ($options->{low_priority} and $options->{concurrent}) {
        Carp::croak 'cannot set both low_priority and concurrent';
    }

    my $low_priority = delete $options->{low_priority} ? 'LOW_PRIORITY' : '';
    my $concurrent = delete $options->{concurrent} ? 'CONCURRENT' : '';

    return $low_priority, $concurrent;
}

sub _parse_replace_and_ignore {
    my ($options) = @_;

    if ($options->{replace} and $options->{ignore}) {
        Carp::croak 'cannot set both replace and ignore';
    }
    my $replace = delete $options->{replace} ? 'REPLACE' : '';
    my $ignore = delete $options->{ignore} ? 'IGNORE' : '';

    return $replace, $ignore;
}

sub _parse_table {
    my $table = delete shift->{table};
    Carp::croak 'table required for load_data_infile' unless defined $table and $table ne '';

    return $table;

}

sub _parse_partition {
    my ($options) = @_;

    my $partition = delete $options->{partition};
    my $ref = ref $partition // '';
    Carp::croak 'partition must be an arrayref if provided' if $partition and $ref ne 'ARRAY';

    return $partition && @$partition ? 'PARTITION (' . join(',', map "`$_`", @$partition) . ')' : '';
}

sub _parse_character_set_and_tempfile_open_mode {
    my ($options) = @_;

    if ($options->{character_set} xor $options->{tempfile_open_mode}) {
        Carp::croak 'character_set and tempfile_open_mode must both be set when one is';
    }

    my $character_set = delete $options->{character_set} || 'utf8';
    my $tempfile_open_mode = delete $options->{tempfile_open_mode} || '>:encoding(UTF-8)';

    return $character_set, $tempfile_open_mode;
}

sub _parse_set {
    my ($options) = @_;

    my $set = delete $options->{set};
    my $ref = ref $set // '';
    Carp::croak 'set must be an arrayref if provided' if $set and $ref ne 'ARRAY';

    return
        $set && @$set
            ? 'SET ' . join ',', map {
                    Carp::croak 'hashrefs passed to set must have only exactly key and value' unless keys %$_ == 1;
                    my ($column, $expression) = %$_;
                    "`$column`=$expression"
                } @$set
            : '';
}

sub _parse_rows_and_columns_and_headers {
    my ($options) = @_;

    my $rows = delete $options->{rows} || Carp::croak 'rows required for load_data_infile';
    Carp::croak 'rows must be an arrayref' unless ref $rows eq 'ARRAY';
    Carp::croak 'rows cannot be empty' unless @$rows;

    my $hashes_in_columns_allowed;
    if (ref $rows->[0] eq 'ARRAY') {
        Carp::croak 'columns required when rows contains arrayrefs'
            unless $options->{columns} and @{$options->{columns}};
    } else {
        $hashes_in_columns_allowed = 1;
    }

    Carp::croak 'columns array cannot be empty' if $options->{columns} and not @{$options->{columns}};
    my $columns = delete $options->{columns} || [keys %{$rows->[0]}];
    my @headers = map {
        my $header;
        if (ref $_ and ref $_ eq 'HASH') {
            Carp::croak 'cannot provide hashes in columns when rows contains arrayrefs' unless $hashes_in_columns_allowed;
            Carp::croak 'hashrefs passed to columns must have only one key and value' unless keys %$_ == 1;
            ($header) = keys %$_;
        } else {
            $header = $_;
        }

        Carp::croak 'columns elements cannot be undef or an empty string' unless defined $header and $header ne '';

        $header;
    } @$columns;
    $columns = join ',', map {
        my $column;
        if (ref $_ and ref $_ eq 'HASH') {
            Carp::croak 'cannot provide hashes in columns when rows contains arrayrefs' unless $hashes_in_columns_allowed;
            Carp::croak 'hashrefs passed to columns must have only exactly key and value' unless keys %$_ == 1;
            ($column) = values %$_;
        } else {
            $column = $_;
        }

        Carp::croak 'columns elements cannot be undef or an empty string' unless defined $column and $column ne '';

        "`$column`";
    } @$columns;

    return $rows, $columns, \@headers;
}

sub _build_query {
    my ($low_priority, $concurrent, $tempfile, $replace, $ignore, $table, $partition, $character_set, $columns, $set) = @_;

    return qq{
        LOAD DATA $low_priority $concurrent LOCAL INFILE '$tempfile'
        $replace $ignore INTO TABLE `$table`
        $partition
        CHARACTER SET '$character_set'
        FIELDS TERMINATED BY '\\t' OPTIONALLY ENCLOSED BY '"'
        LINES TERMINATED BY '\\n'
        IGNORE 1 LINES
        ($columns)
        $set
    };
}

1;
__END__