File Coverage

blib/lib/DBIx/Lite/ResultSet.pm

Criterion	Covered	Total	%
statement	227	348	65.2
branch	68	148	45.9
condition	28	77	36.3
subroutine	40	62	64.5
pod	33	33	100.0
total	396	668	59.2

package DBIx::Lite::ResultSet;
$DBIx::Lite::ResultSet::VERSION = '0.33';
use strict;
use warnings;

use Carp qw(croak);
use Clone qw(clone);
use Data::Page;
use List::MoreUtils qw(uniq firstval);
use vars qw($AUTOLOAD);
$Carp::Internal{$_}++ for __PACKAGE__;

sub _new {
    my $class = shift;
    my (%params) = @_;

    my $self = {
        joins           => [],
        where           => [],
        select          => ['me.*'],
        rows_per_page   => 10,
    };

    # required arguments
    for (qw(dbix_lite table)) {
        $self->{$_} = delete $params{$_} or croak "$_ argument needed";
    }

    # optional arguments
    for (grep exists($params{$_}), qw(joins where select group_by having order_by
        limit offset for rows_per_page page cur_table with from)) {
        $self->{$_} = delete $params{$_};
    }
    $self->{cur_table} //= $self->{table};

    !%params
        or croak "Unknown options: " . join(', ', keys %params);

    bless $self, $class;
    $self;
}

# create setters
for my $methname (qw(group_by having order_by limit offset rows_per_page page from)) {
    no strict 'refs';
    *$methname = sub {
        my $self = shift;

        # we always return a new object for easy chaining
        my $new_self = $self->_clone;

        # set new values
        $new_self->{$methname} = $methname =~ /^(group_by|order_by|from)$/ ? [@_] : $_[0];
        $new_self->{pager}->current_page($_[0]) if $methname eq 'page' && $new_self->{pager};

        # return object
        $new_self;
    };
}

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

    return $self->for('UPDATE');
}

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

    my $new_self = $self->_clone;
    $new_self->{for} = $for;
    $new_self;
}

# return a clone of this object
sub _clone {
    my $self = shift;
    (ref $self)->_new(
        # clone all members except for some which we copy by reference
        map { $_ => /^(?:dbix_lite|table|cur_table)$/ ? $self->{$_} : clone($self->{$_}) }
            grep !/^(?:sth|pager)$/, keys %$self,
    );
}

sub select {
    my $self = shift;

    my $new_self = $self->_clone;
    $new_self->{select} = @_ ? [@_] : undef;

    $new_self;
}

sub select_also {
    my $self = shift;
    return $self->select(@{$self->{select}}, @_);
}

sub with {
    my $self = shift;
    my %with = @_;

    croak "with() requires a hash of scalarrefs or refs to arrayrefs"
        if grep { !ref($_) eq 'SCALAR' } values %with;

    my $new_self = $self->_clone;
    $new_self->{with} = %with ? {%with} : undef;

    $new_self;
}

sub with_also {
    my $self = shift;
    return $self->with(%{$self->{with}}, @_);
}

sub pager {
    my $self = shift;
    if (!$self->{pager}) {
        $self->{pager} ||= Data::Page->new;
        $self->{pager}->total_entries($self->page(undef)->count);
        $self->{pager}->entries_per_page($self->{rows_per_page} // $self->{pager}->total_entries);
        $self->{pager}->current_page($self->{page});
    }
    return $self->{pager};
}

sub search {
    my $self = shift;
    my ($where) = @_;

    my $new_self = $self->_clone;
    push @{$new_self->{where}}, $where if defined $where;
    $new_self;
}

sub clear_search {
    my $self = shift;

    my $new_self = $self->_clone;
    @{$new_self->{where}} = ();
    $new_self;
}

sub find {
    my $self = shift;
    my ($where) = @_;

    # if user did not supply a search hashref, we assume the supplied
    # value(s) are the key(s) of the primary key column(s) defined for
    # this table
    if (!ref $where && (my @pk = $self->{table}->pk)) {
        # prepend table alias to all pk columns
        $_ =~ s/^[^.]+$/me\.$&/ for @pk;

        $where = { map +(shift(@pk) => $_), @_ };
    }
    return $self->search($where)->single;
}

sub where_sql {
    my $self = shift;

    my ($sql, @bind) = $self->{dbix_lite}->{abstract}->where({ -and => $self->{where} });
    return ($sql, @bind);
}

sub select_sql {
    my $self = shift;

    # prepare names of columns to be selected
    my @cols = ();
    my $cur_table_prefix = $self->_table_alias($self->{cur_table}{name}, 'select');
    foreach my $col (grep defined $_, @{$self->{select}}) {
        # check whether user specified an alias
        my ($expr, $as) = ref $col eq 'ARRAY' ? @$col : ($col, undef);

        # prepend table alias if column name doesn't contain one already
        $expr =~ s/^[^.]+$/$cur_table_prefix\.$&/ if !ref($expr) && !$self->{from};

        # explode the expression if it's a scalar ref
        if (ref $expr eq 'SCALAR') {
            $expr = $$expr;
        }

        # build the column definition according to the SQL::Abstract::More syntax
        push @cols, $expr . ($as ? "|$as" : "");
    }

    # joins
    my @joins = ();
    foreach my $join (@{$self->{joins}}) {
        # get table name and alias if any
        my ($table_name, $table_alias) = @{$join->{table}};
        my $left_table_prefix = $self->_table_alias($join->{cur_table}{name}, 'select');

        # prepare join conditions
        my %cond = ();
        while (my ($col1, $col2) = each %{$join->{condition}}) {
            # in case they have no explicit table alias,
            # $col1 is supposed to belong to the current table, and
            # $col2 is supposed to belong to the joined table

            # prepend table alias to the column of the first table
            $col1 =~ s/^[^.]+$/$left_table_prefix.$&/;

            # in case user supplied the table name as table alias, replace it
            # with the proper one (such as "me.")
            $col1 =~ s/^$join->{cur_table}{name}\./$left_table_prefix./;

            # prepend table alias to the column of the second table
            $col2 = ($table_alias || $self->{dbix_lite}->_quote($table_name)) . ".$col2"
                unless ref $col2 || $col2 =~ /\./;

            # in case the second item is a scalar reference (literal SQL)
            # or a hashref (search condition), pass it unchanged
            $cond{$col1} = ref($col2) ? $col2 : \ "= $col2";
        }

        # store the join definition according to the SQL::Abstract::More syntax
        push @joins, {
            operator    => $join->{join_type} eq 'inner' ? '<=>' : '=>',
            condition   => \%cond,
        };
        push @joins, $table_name . ($table_alias ? "|$table_alias" : "");
    }

    # from
    my @from = ();
    if ($self->{from}) {
        @from = @{$self->{from}};
    } else {
        @from = (-join => $self->{table}{name} . "|me", @joins);
    }

    # paging overrides limit and offset if any
    if ($self->{page} && defined $self->{rows_per_page}) {
        $self->{limit} = $self->{rows_per_page};
        $self->{offset} = $self->pager->skipped;
    }

    # ordering
    if ($self->{order_by}) {
        $self->{order_by} = [$self->{order_by}]
            unless ref $self->{order_by} eq 'ARRAY';
    }

    my ($sql, @bind) = $self->{dbix_lite}->{abstract}->select(
        -columns    => [ uniq @cols ],
        -from       => [ @from ],
        -where      => { -and => $self->{where} },
        $self->{group_by}   ? (-group_by    => $self->{group_by})   : (),
        $self->{having}     ? (-having      => $self->{having})     : (),
        $self->{order_by}   ? (-order_by    => $self->{order_by})   : (),
        $self->{limit}      ? (-limit       => $self->{limit})      : (),
        $self->{offset}     ? (-offset      => $self->{offset})     : (),
        $self->{for}        ? (-for         => $self->{for})     : (),
    );

    if ($self->{with}) {
        my @with_sql = ();
        foreach my $alias (keys %{$self->{with}}) {
            my $def = $self->{with}{$alias};

            if (ref $$def eq 'ARRAY') {
                my ($wsql, @wbind) = @$$def;
                push @with_sql, sprintf "%s AS (%s)", $alias, $wsql;
                unshift @bind, @wbind;
            } else {
                push @with_sql, sprintf "%s AS (%s)", $alias, $$def;
            }
        }
        $sql = sprintf 'WITH %s %s', join(', ', @with_sql), $sql;
    }

    return ($sql, @bind);
}

sub select_sth {
    my $self = shift;

    my ($sql, @bind) = $self->select_sql;
    return $self->{dbix_lite}->dbh->prepare($sql) || undef, @bind;
}

sub _select_sth_for_object {
    my $self = shift;

    # check whether any of the selected columns is a scalar ref
    my $cur_table_prefix = $self->_table_alias($self->{cur_table}{name}, 'select');
    my $have_scalar_ref = 0;
    my $have_star = 0;
    foreach my $col (grep defined $_, @{$self->{select}}) {
        my $expr = ref($col) eq 'ARRAY' ? $col->[0] : $col;
        if (ref($expr) eq 'SCALAR') {
            $have_scalar_ref = 1;
        } elsif ($expr eq "$cur_table_prefix.*") {
            $have_star = 1;
        }
    }

    # always retrieve our primary key if provided and no col name is a scalar ref
    # also skip this if we are retrieving all columns (me.*)
    if (!$have_scalar_ref && !$have_star && (my @pk = $self->{cur_table}->pk)) {
        # prepend table alias to all pk columns
        $_ =~ s/^[^.]+$/$cur_table_prefix\.$&/ for @pk;

        # append instead of prepend, otherwise get_column() on a non-PK column
        # would return the wrong values
        $self = $self->select_also(@pk);
    }

    return $self->select_sth;
}

sub insert_sql {
    my $self = shift;
    my $insert_cols = shift;
    ref $insert_cols eq 'HASH' or croak "insert_sql() requires a hashref";

    if (@{$self->{joins}}) {
        warn "Attempt to call ->insert() after joining other tables\n";
    }

    if (!%$insert_cols && $self->{dbix_lite}->driver_name eq 'Pg') {
        # Postgres doesn't support the VALUES () syntax
        return sprintf "INSERT INTO %s DEFAULT VALUES",
            $self->{dbix_lite}->_quote($self->{table}{name});
    }

    return $self->{dbix_lite}->{abstract}->insert(
        $self->{table}{name}, $insert_cols,
    );
}

sub insert_sth {
    my $self = shift;
    my $insert_cols = shift;
    ref $insert_cols eq 'HASH' or croak "insert_sth() requires a hashref";

    my ($sql, @bind) = $self->insert_sql($insert_cols);
    return $self->{dbix_lite}->dbh->prepare($sql) || undef, @bind;
}

sub insert {
    my $self = shift;
    my $insert_cols = shift;
    ref $insert_cols eq 'HASH' or croak "insert() requires a hashref";

    # perform operation
    my $res;
    $self->{dbix_lite}->dbh_do(sub {
        my ($sth, @bind) = $self->insert_sth($insert_cols);
        $res = $sth->execute(@bind);
    });
    return undef if !$res;

    # populate the autopk field if any
    if (my $pk = $self->{table}->autopk) {
        $insert_cols = clone $insert_cols;
        $insert_cols->{$pk} = $self->{dbix_lite}->_autopk($self->{table}{name})
            if !exists $insert_cols->{$pk};
    }

    # return a DBIx::Lite::Row object with the inserted values
    return $self->_inflate_row($insert_cols);
}

sub update_sql {
    my $self = shift;
    my $update_cols = shift;
    ref $update_cols eq 'HASH' or croak "update_sql() requires a hashref";

    my $update_where = { -and => $self->{where} };

    if ($self->{cur_table}{name} ne $self->{table}{name}) {
        my @pk = $self->{cur_table}->pk
            or croak "No primary key defined for " . $self->{cur_table}{name} . "; cannot update using relationships";
        @pk == 1
            or croak "Update across relationships is not allowed with multi-column primary keys";

        my $fq_pk = $self->_table_alias($self->{cur_table}{name}, 'update') . "." . $pk[0];
        $update_where = {
            $fq_pk => {
                -in => \[ $self->select($pk[0])->select_sql ],
            },
        };
    }

    return $self->{dbix_lite}->{abstract}->update(
        -table  => $self->_table_alias_expr($self->{cur_table}{name}, 'update'),
        -set    => $update_cols,
        -where  => $update_where,
    );
}

sub update_sth {
    my $self = shift;
    my $update_cols = shift;
    ref $update_cols eq 'HASH' or croak "update_sth() requires a hashref";

    my ($sql, @bind) = $self->update_sql($update_cols);
    return $self->{dbix_lite}->dbh->prepare($sql) || undef, @bind;
}

sub update {
    my $self = shift;
    my $update_cols = shift;
    ref $update_cols eq 'HASH' or croak "update() requires a hashref";

    my $affected_rows;
    $self->{dbix_lite}->dbh_do(sub {
        my ($sth, @bind) = $self->update_sth($update_cols);
        $affected_rows = $sth->execute(@bind);
    });
    return $affected_rows;
}

sub find_or_insert {
    my $self = shift;
    my $cols = shift;
    ref $cols eq 'HASH' or croak "find_or_insert() requires a hashref";

    my $object;
    $self->{dbix_lite}->txn(sub {
        if (!($object = $self->find($cols))) {
            $object = $self->insert($cols);
        }
    });
    return $object;
}

sub delete_sql {
    my $self = shift;

    my $delete_where = { -and => $self->{where} };

    if ($self->{cur_table}{name} ne $self->{table}{name}) {
        my @pk = $self->{cur_table}->pk
            or croak "No primary key defined for " . $self->{cur_table}{name} . "; cannot delete using relationships";
        @pk == 1
            or croak "Delete across relationships is not allowed with multi-column primary keys";

        my $fq_pk = $self->_table_alias($self->{cur_table}{name}, 'delete') . "." . $pk[0];
        $delete_where = {
            $fq_pk => {
                -in => \[ $self->select($pk[0])->select_sql ],
            },
        };
    }

    return $self->{dbix_lite}->{abstract}->delete(
        $self->_table_alias_expr($self->{cur_table}{name}, 'delete'),
        $delete_where,
    );
}

sub delete_sth {
    my $self = shift;

    my ($sql, @bind) = $self->delete_sql;
    return $self->{dbix_lite}->dbh->prepare($sql) || undef, @bind;
}

sub delete {
    my $self = shift;

    my $affected_rows;
    $self->{dbix_lite}->dbh_do(sub {
        my ($sth, @bind) = $self->delete_sth;
        $affected_rows = $sth->execute(@bind);
    });
    return $affected_rows;
}

sub single {
    my $self = shift;

    my $row;
    $self->{dbix_lite}->dbh_do(sub {
        my ($sth, @bind) = $self->_select_sth_for_object;
        $sth->execute(@bind);
        $row = $sth->fetchrow_hashref;
    });
    return $row ? $self->_inflate_row($row) : undef;
}

sub single_value {
    my $self = shift;

    my $value;
    $self->{dbix_lite}->dbh_do(sub {
        my ($sth, @bind) = $self->select_sth;
        $sth->execute(@bind);
        ($value) = $sth->fetchrow_array;
    });
    return $value;
}

sub all {
    my $self = shift;

    my $rows;
    $self->{dbix_lite}->dbh_do(sub {
        my ($sth, @bind) = $self->_select_sth_for_object;
        $sth->execute(@bind);
        $rows = $sth->fetchall_arrayref({});
    });
    return map $self->_inflate_row($_), @$rows;
}

sub next {
    my $self = shift;

    $self->{dbix_lite}->dbh_do(sub {
        ($self->{sth}, my @bind) = $self->_select_sth_for_object;
        $self->{sth}->execute(@bind);
    }) if !$self->{sth};

    my $row = $self->{sth}->fetchrow_hashref or return undef;
    return $self->_inflate_row($row);
}

sub count {
    my $self = shift;

    my $count;
    $self->{dbix_lite}->dbh_do(sub {
        # Postgres throws an error when using ORDER BY clauses with COUNT(*)
        my $count_rs = $self->select(\ "COUNT(*)")->order_by(undef);
        my ($sth, @bind) = $count_rs->select_sth;
        $sth->execute(@bind);
        $count = +($sth->fetchrow_array)[0];
    });
    return $count;
}

sub column_names {
    my $self = shift;

    $self->{dbix_lite}->dbh_do(sub {
        ($self->{sth}, my @bind) = $self->select_sth;
        $self->{sth}->execute(@bind);
    }) if !$self->{sth};

    my $c = $self->{sth}->{NAME};
    return wantarray ? @$c : $c;
}

sub get_column {
    my $self = shift;
    my $column_name = shift or croak "get_column() requires a column name";

    my @values = ();
    $self->{dbix_lite}->dbh_do(sub {
        my $rs = ($self->_clone)->select($column_name);
        my ($sql, @bind) = $rs->select_sql;

        @values = @{$self->{dbix_lite}->dbh->selectcol_arrayref($sql, {}, @bind)};
    });
    return @values;
}

sub inner_join {
    my $self = shift;
    return $self->_join('inner', @_);
}

sub left_join {
    my $self = shift;
    return $self->_join('left', @_);
}

sub _join {
    my $self = shift;
    my ($type, $table_name, $condition, $options) = @_;
    $options ||= {};
    $table_name = [ $table_name, undef ] if ref $table_name ne 'ARRAY';

    my $new_self = $self->_clone;

    # if user asked for duplicate join removal, check whether no joins
    # with the same table alias exist
    if ($options->{prevent_duplicates}) {
        foreach my $join (@{$self->{joins}}) {
            if ((defined($table_name->[1]) && defined $join->{table}[1] && $table_name->[1] eq $join->{table}[1])
                || (!defined($table_name->[1]) && $table_name->[0] eq $join->{table}[0])) {
                return $new_self;
            }
        }
    }

    push @{$new_self->{joins}}, {
        join_type   => $type,
        cur_table   => $self->{cur_table},
        table       => $table_name,
        condition   => $condition,
    };

    $new_self;
}

sub clear_joins {
    my $self = shift;

    my $new_self = $self->_clone;
    @{$new_self->{joins}} = ();
    $new_self;
}

sub _table_alias {
    my $self = shift;
    my ($table_name, $op) = @_;

    my $driver_name = $self->{dbix_lite}->driver_name;

    if ($table_name eq $self->{table}{name}) {
        if ($op eq 'select'
            || ($op eq 'update' && $driver_name =~ /^(?:MySQL|Pg)$/i)
            || ($op eq 'delete' && $driver_name =~ /^Pg$/i)) {
            return 'me';
        }
    }

    return $table_name;
}

sub _table_alias_expr {
    my $self = shift;
    my ($table_name, $op) = @_;

    my $table_alias = $self->_table_alias($table_name, $op);
    if ($table_name eq $table_alias) {
        # foo
        return $table_name;
    } else {
        # foo AS my_foo
        return $self->{dbix_lite}->{abstract}->table_alias($table_name, $table_alias);
    }
}

sub _inflate_row {
    my $self = shift;
    my ($hashref) = @_;

    # get the row package
    my $package = $self->{cur_table}{class} || 'DBIx::Lite::Row';

    # get the constructor, if any
    my $constructor = $self->{cur_table}{class_constructor};
    if (!defined $constructor && $package->can('new')) {
        $constructor = 'new';
    }

    # create the object
    my $object;
    if (defined $constructor) {
        if (ref($constructor) eq 'CODE') {
            $object = $constructor->($hashref);
        } else {
            $object = $package->$constructor;
        }
    } else {
        $object = {};
        bless $object, $package;
    }

    # get the hashref where we are going to store our data
    my $storage;
    if (my $method = $self->{cur_table}{class_storage}) {
        croak "No ${package}::${method} method exists"
            if !$package->can($method);
        $storage = $object->$method;
        croak "${package}::${method}() did not return a hashref"
            if ref($storage) ne 'HASH';
    } else {
        $storage = $object;
    }

    # store our data
    $storage->{dbix_lite}   = $self->{dbix_lite};
    $storage->{table}       = $self->{cur_table};
    $storage->{data}        = $hashref;

    return $object;
}

sub AUTOLOAD {
    my $self = shift or return undef;

    # Get the called method name and trim off the namespace
    (my $method = $AUTOLOAD) =~ s/.*:://;

    if (my $rel = $self->{cur_table}{has_many}{$method}) {
        my $new_self = $self->inner_join($rel->[0], $rel->[1])->select("$method.*");
        $new_self->{cur_table} = $self->{dbix_lite}->schema->table($rel->[0]);
        bless $new_self, $new_self->{cur_table}->resultset_class || __PACKAGE__;
        return $new_self;
    }

    croak "No $method method is provided by this " . ref($self) . " object";
}

sub DESTROY {}

1;

__END__