bab19eeff2afe22ce0d491b08f2099fd447ce8b8
[dbsrgits/DBIx-Class.git] / lib / DBIx / Class / ResultSet.pm
1 package DBIx::Class::ResultSet;
2
3 use strict;
4 use warnings;
5 use overload
6         '0+'     => \&count,
7         'bool'   => sub { 1; },
8         fallback => 1;
9 use Data::Page;
10 use Storable;
11 use Scalar::Util qw/weaken/;
12
13 use base qw/DBIx::Class/;
14 __PACKAGE__->load_components(qw/AccessorGroup/);
15 __PACKAGE__->mk_group_accessors('simple' => qw/result_source result_class/);
16
17 =head1 NAME
18
19 DBIx::Class::ResultSet - Responsible for fetching and creating resultset.
20
21 =head1 SYNOPSIS
22
23   my $rs   = $schema->resultset('User')->search(registered => 1);
24   my @rows = $schema->resultset('CD')->search(year => 2005);
25
26 =head1 DESCRIPTION
27
28 The resultset is also known as an iterator. It is responsible for handling
29 queries that may return an arbitrary number of rows, e.g. via L</search>
30 or a C<has_many> relationship.
31
32 In the examples below, the following table classes are used:
33
34   package MyApp::Schema::Artist;
35   use base qw/DBIx::Class/;
36   __PACKAGE__->load_components(qw/Core/);
37   __PACKAGE__->table('artist');
38   __PACKAGE__->add_columns(qw/artistid name/);
39   __PACKAGE__->set_primary_key('artistid');
40   __PACKAGE__->has_many(cds => 'MyApp::Schema::CD');
41   1;
42
43   package MyApp::Schema::CD;
44   use base qw/DBIx::Class/;
45   __PACKAGE__->load_components(qw/Core/);
46   __PACKAGE__->table('cd');
47   __PACKAGE__->add_columns(qw/cdid artist title year/);
48   __PACKAGE__->set_primary_key('cdid');
49   __PACKAGE__->belongs_to(artist => 'MyApp::Schema::Artist');
50   1;
51
52 =head1 METHODS
53
54 =head2 new 
55
56 =over 4
57
58 =item Arguments: $source, \%$attrs
59
60 =item Return Value: $rs
61
62 =item 
63
64 =back
65
66 The resultset constructor. Takes a source object (usually a
67 L<DBIx::Class::ResultSourceProxy::Table>) and an attribute hash (see
68 L</ATTRIBUTES> below).  Does not perform any queries -- these are
69 executed as needed by the other methods.
70
71 Generally you won't need to construct a resultset manually.  You'll
72 automatically get one from e.g. a L</search> called in scalar context:
73
74   my $rs = $schema->resultset('CD')->search({ title => '100th Window' });
75
76 IMPORTANT: If called on an object, proxies to new_result instead so
77
78   my $cd = $schema->resultset('CD')->new({ title => 'Spoon' });
79
80 will return a CD object, not a ResultSet.
81
82 =cut
83
84 sub new {
85   my $class = shift;
86   return $class->new_result(@_) if ref $class;
87   
88   my ($source, $attrs) = @_;
89   weaken $source;
90   $attrs = Storable::dclone($attrs || {}); # { %{ $attrs || {} } };
91   #use Data::Dumper; warn Dumper($attrs);
92   my $alias = ($attrs->{alias} ||= 'me');
93   
94   $attrs->{columns} ||= delete $attrs->{cols} if $attrs->{cols};
95   delete $attrs->{as} if $attrs->{columns};
96   $attrs->{columns} ||= [ $source->columns ] unless $attrs->{select};
97   $attrs->{select} = [
98     map { m/\./ ? $_ : "${alias}.$_" } @{delete $attrs->{columns}}
99   ] if $attrs->{columns};
100   $attrs->{as} ||= [
101     map { m/^\Q$alias.\E(.+)$/ ? $1 : $_ } @{$attrs->{select}}
102   ];
103   if (my $include = delete $attrs->{include_columns}) {
104     push(@{$attrs->{select}}, @$include);
105     push(@{$attrs->{as}}, map { m/([^.]+)$/; $1; } @$include);
106   }
107   #use Data::Dumper; warn Dumper(@{$attrs}{qw/select as/});
108
109   $attrs->{from} ||= [ { $alias => $source->from } ];
110   $attrs->{seen_join} ||= {};
111   my %seen;
112   if (my $join = delete $attrs->{join}) {
113     foreach my $j (ref $join eq 'ARRAY' ? @$join : ($join)) {
114       if (ref $j eq 'HASH') {
115         $seen{$_} = 1 foreach keys %$j;
116       } else {
117         $seen{$j} = 1;
118       }
119     }
120     push(@{$attrs->{from}}, $source->resolve_join(
121       $join, $attrs->{alias}, $attrs->{seen_join})
122     );
123   }
124   
125   $attrs->{group_by} ||= $attrs->{select} if delete $attrs->{distinct};
126   $attrs->{order_by} = [ $attrs->{order_by} ] if
127     $attrs->{order_by} and !ref($attrs->{order_by});
128   $attrs->{order_by} ||= [];
129
130   my $collapse = $attrs->{collapse} || {};
131   if (my $prefetch = delete $attrs->{prefetch}) {
132     my @pre_order;
133     foreach my $p (ref $prefetch eq 'ARRAY' ? @$prefetch : ($prefetch)) {
134       if ( ref $p eq 'HASH' ) {
135         foreach my $key (keys %$p) {
136           push(@{$attrs->{from}}, $source->resolve_join($p, $attrs->{alias}))
137             unless $seen{$key};
138         }
139       } else {
140         push(@{$attrs->{from}}, $source->resolve_join($p, $attrs->{alias}))
141             unless $seen{$p};
142       }
143       my @prefetch = $source->resolve_prefetch(
144            $p, $attrs->{alias}, {}, \@pre_order, $collapse);
145       push(@{$attrs->{select}}, map { $_->[0] } @prefetch);
146       push(@{$attrs->{as}}, map { $_->[1] } @prefetch);
147     }
148     push(@{$attrs->{order_by}}, @pre_order);
149   }
150   $attrs->{collapse} = $collapse;
151 #  use Data::Dumper; warn Dumper($collapse) if keys %{$collapse};
152
153   if ($attrs->{page}) {
154     $attrs->{rows} ||= 10;
155     $attrs->{offset} ||= 0;
156     $attrs->{offset} += ($attrs->{rows} * ($attrs->{page} - 1));
157   }
158
159   bless {
160     result_source => $source,
161     result_class => $attrs->{result_class} || $source->result_class,
162     cond => $attrs->{where},
163     from => $attrs->{from},
164     collapse => $collapse,
165     count => undef,
166     page => delete $attrs->{page},
167     pager => undef,
168     attrs => $attrs
169   }, $class;
170 }
171
172 =head2 search
173
174 =over 4
175
176 =item Arguments: $cond, \%attrs?
177
178 =item Return Value: $resultset (scalar context), @row_objs (list context)
179
180 =back
181
182   my @cds    = $cd_rs->search({ year => 2001 }); # "... WHERE year = 2001"
183   my $new_rs = $cd_rs->search({ year => 2005 });
184
185   my $new_rs = $cd_rs->search([ { year => 2005 }, { year => 2004 } ]);
186                  # year = 2005 OR year = 2004
187
188 If you need to pass in additional attributes but no additional condition,
189 call it as C<search(undef, \%attrs);>.
190
191   # "SELECT name, artistid FROM $artist_table"
192   my @all_artists = $schema->resultset('Artist')->search(undef, {
193     columns => [qw/name artistid/],
194   });
195
196 =cut
197
198 sub search {
199   my $self = shift;
200
201   my $rs;
202   if( @_ ) {
203     
204     my $attrs = { %{$self->{attrs}} };
205     my $having = delete $attrs->{having};
206     $attrs = { %$attrs, %{ pop(@_) } } if @_ > 1 and ref $_[$#_] eq 'HASH';
207
208     my $where = (@_
209                   ? ((@_ == 1 || ref $_[0] eq "HASH")
210                       ? shift
211                       : ((@_ % 2)
212                           ? $self->throw_exception(
213                               "Odd number of arguments to search")
214                           : {@_}))
215                   : undef());
216     if (defined $where) {
217       $attrs->{where} = (defined $attrs->{where}
218                 ? { '-and' =>
219                     [ map { ref $_ eq 'ARRAY' ? [ -or => $_ ] : $_ }
220                         $where, $attrs->{where} ] }
221                 : $where);
222     }
223
224     if (defined $having) {
225       $attrs->{having} = (defined $attrs->{having}
226                 ? { '-and' =>
227                     [ map { ref $_ eq 'ARRAY' ? [ -or => $_ ] : $_ }
228                         $having, $attrs->{having} ] }
229                 : $having);
230     }
231
232     $rs = (ref $self)->new($self->result_source, $attrs);
233   }
234   else {
235     $rs = $self;
236     $rs->reset;
237   }
238   return (wantarray ? $rs->all : $rs);
239 }
240
241 =head2 search_literal
242
243 =over 4
244
245 =item Arguments: $sql_fragment, @bind_values
246
247 =item Return Value: $resultset (scalar context), @row_objs (list context)
248
249 =back
250
251   my @cds   = $cd_rs->search_literal('year = ? AND title = ?', qw/2001 Reload/);
252   my $newrs = $artist_rs->search_literal('name = ?', 'Metallica');
253
254 Pass a literal chunk of SQL to be added to the conditional part of the
255 resultset query.
256
257 =cut
258
259 sub search_literal {
260   my ($self, $cond, @vals) = @_;
261   my $attrs = (ref $vals[$#vals] eq 'HASH' ? { %{ pop(@vals) } } : {});
262   $attrs->{bind} = [ @{$self->{attrs}{bind}||[]}, @vals ];
263   return $self->search(\$cond, $attrs);
264 }
265
266 =head2 find
267
268 =over 4
269
270 =item Arguments: (@values | \%cols), \%attrs?
271
272 =item Return Value: $row_object
273
274 =back
275
276 Finds a row based on its primary key or unique constraint. For example:
277
278   my $cd = $schema->resultset('CD')->find(5);
279
280 Also takes an optional C<key> attribute, to search by a specific key or unique
281 constraint. For example:
282
283   my $cd = $schema->resultset('CD')->find(
284     {
285       artist => 'Massive Attack',
286       title  => 'Mezzanine',
287     },
288     { key => 'artist_title' }
289   );
290
291 See also L</find_or_create> and L</update_or_create>.
292
293 =cut
294
295 sub find {
296   my ($self, @vals) = @_;
297   my $attrs = (@vals > 1 && ref $vals[$#vals] eq 'HASH' ? pop(@vals) : {});
298
299   my @cols = $self->result_source->primary_columns;
300   if (exists $attrs->{key}) {
301     my %uniq = $self->result_source->unique_constraints;
302     $self->throw_exception(
303       "Unknown key $attrs->{key} on '" . $self->result_source->name . "'"
304     ) unless exists $uniq{$attrs->{key}};
305     @cols = @{ $uniq{$attrs->{key}} };
306   }
307   #use Data::Dumper; warn Dumper($attrs, @vals, @cols);
308   $self->throw_exception(
309     "Can't find unless a primary key or unique constraint is defined"
310   ) unless @cols;
311
312   my $query;
313   if (ref $vals[0] eq 'HASH') {
314     $query = { %{$vals[0]} };
315   } elsif (@cols == @vals) {
316     $query = {};
317     @{$query}{@cols} = @vals;
318   } else {
319     $query = {@vals};
320   }
321   foreach my $key (grep { ! m/\./ } keys %$query) {
322     $query->{"$self->{attrs}{alias}.$key"} = delete $query->{$key};
323   }
324   #warn Dumper($query);
325   
326   if (keys %$attrs) {
327       my $rs = $self->search($query,$attrs);
328       return keys %{$rs->{collapse}} ? $rs->next : $rs->single;
329   } else {
330       return keys %{$self->{collapse}} ?
331         $self->search($query)->next :
332         $self->single($query);
333   }
334 }
335
336 =head2 search_related
337
338 =over 4
339
340 =item Arguments: $cond, \%attrs?
341
342 =item Return Value: $new_resultset
343
344 =back
345
346   $new_rs = $cd_rs->search_related('artist', {
347     name => 'Emo-R-Us',
348   });
349
350 Search the specified relationship, optionally specify a condition and
351 attributes for matching records. See L</ATTRIBUTES> for more information.
352
353 =cut
354
355 sub search_related {
356   return shift->related_resultset(shift)->search(@_);
357 }
358
359 =head2 cursor
360
361 =over 4
362
363 =item Arguments: none
364
365 =item Return Value: $cursor
366
367 =back
368
369 Returns a storage-driven cursor to the given resultset. See
370 L<DBIx::Class::Cursor> for more information.
371
372 =cut
373
374 sub cursor {
375   my ($self) = @_;
376   my $attrs = { %{$self->{attrs}} };
377   return $self->{cursor}
378     ||= $self->result_source->storage->select($self->{from}, $attrs->{select},
379           $attrs->{where},$attrs);
380 }
381
382 =head2 single
383
384 =over 4
385
386 =item Arguments: $cond?
387
388 =item Return Value: $row_object?
389
390 =back
391
392   my $cd = $schema->resultset('CD')->single({ year => 2001 });
393
394 Inflates the first result without creating a cursor if the resultset has
395 any records in it; if not returns nothing. Used by find() as an optimisation.
396
397 =cut
398
399 sub single {
400   my ($self, $where) = @_;
401   my $attrs = { %{$self->{attrs}} };
402   if ($where) {
403     if (defined $attrs->{where}) {
404       $attrs->{where} = {
405         '-and' => 
406             [ map { ref $_ eq 'ARRAY' ? [ -or => $_ ] : $_ }
407                $where, delete $attrs->{where} ]
408       };
409     } else {
410       $attrs->{where} = $where;
411     }
412   }
413   my @data = $self->result_source->storage->select_single(
414           $self->{from}, $attrs->{select},
415           $attrs->{where},$attrs);
416   return (@data ? $self->_construct_object(@data) : ());
417 }
418
419
420 =head2 search_like
421
422 =over 4
423
424 =item Arguments: $cond, \%attrs?
425
426 =item Return Value: $resultset (scalar context), @row_objs (list context)
427
428 =back
429
430   # WHERE title LIKE '%blue%'
431   $cd_rs = $rs->search_like({ title => '%blue%'});
432
433 Perform a search, but use C<LIKE> instead of C<=> as the condition. Note
434 that this is simply a convenience method. You most likely want to use
435 L</search> with specific operators.
436
437 For more information, see L<DBIx::Class::Manual::Cookbook>.
438
439 =cut
440
441 sub search_like {
442   my $class = shift;
443   my $attrs = (@_ > 1 && ref $_[$#_] eq 'HASH' ? pop(@_) : {});
444   my $query = ref $_[0] eq 'HASH' ? { %{shift()} }: {@_};
445   $query->{$_} = { 'like' => $query->{$_} } for keys %$query;
446   return $class->search($query, { %$attrs });
447 }
448
449 =head2 slice
450
451 =over 4
452
453 =item Arguments: $first, $last
454
455 =item Return Value: $resultset (scalar context), @row_objs (list context)
456
457 =back
458
459 Returns a resultset or object list representing a subset of elements from the
460 resultset slice is called on.  Indexes are from 0 - i.e. to get the first
461 three records, call
462
463   my ($one, $two, $three) = $rs->slice(0, 2);
464
465 =cut
466
467 sub slice {
468   my ($self, $min, $max) = @_;
469   my $attrs = { %{ $self->{attrs} || {} } };
470   $attrs->{offset} ||= 0;
471   $attrs->{offset} += $min;
472   $attrs->{rows} = ($max ? ($max - $min + 1) : 1);
473   my $slice = (ref $self)->new($self->result_source, $attrs);
474   return (wantarray ? $slice->all : $slice);
475 }
476
477 =head2 next
478
479 =over 4
480
481 =item Arguments: none
482
483 =item Return Value: $result?
484
485 =back
486
487 Returns the next element in the resultset (C<undef> is there is none).
488
489 Can be used to efficiently iterate over records in the resultset:
490
491   my $rs = $schema->resultset('CD')->search;
492   while (my $cd = $rs->next) {
493     print $cd->title;
494   }
495
496 =cut
497
498 sub next {
499   my ($self) = @_;
500   if (@{$self->{all_cache} || []}) {
501     $self->{all_cache_position} ||= 0;
502     return $self->{all_cache}->[$self->{all_cache_position}++];
503   }
504   if ($self->{attrs}{cache}) {
505     $self->{all_cache_position} = 1;
506     return ($self->all)[0];
507   }
508   my @row = (exists $self->{stashed_row} ?
509                @{delete $self->{stashed_row}} :
510                $self->cursor->next
511   );
512 #  warn Dumper(\@row); use Data::Dumper;
513   return unless (@row);
514   return $self->_construct_object(@row);
515 }
516
517 sub _construct_object {
518   my ($self, @row) = @_;
519   my @as = @{ $self->{attrs}{as} };
520   
521   my $info = $self->_collapse_result(\@as, \@row);
522   
523   my $new = $self->result_class->inflate_result($self->result_source, @$info);
524   
525   $new = $self->{attrs}{record_filter}->($new)
526     if exists $self->{attrs}{record_filter};
527   return $new;
528 }
529
530 sub _collapse_result {
531   my ($self, $as, $row, $prefix) = @_;
532
533   my %const;
534
535   my @copy = @$row;
536   foreach my $this_as (@$as) {
537     my $val = shift @copy;
538     if (defined $prefix) {
539       if ($this_as =~ m/^\Q${prefix}.\E(.+)$/) {
540         my $remain = $1;
541         $remain =~ /^(?:(.*)\.)?([^.]+)$/;
542         $const{$1||''}{$2} = $val;
543       }
544     } else {
545       $this_as =~ /^(?:(.*)\.)?([^.]+)$/;
546       $const{$1||''}{$2} = $val;
547     }
548   }
549
550   my $info = [ {}, {} ];
551   foreach my $key (keys %const) {
552     if (length $key) {
553       my $target = $info;
554       my @parts = split(/\./, $key);
555       foreach my $p (@parts) {
556         $target = $target->[1]->{$p} ||= [];
557       }
558       $target->[0] = $const{$key};
559     } else {
560       $info->[0] = $const{$key};
561     }
562   }
563
564   my @collapse;
565   if (defined $prefix) {
566     @collapse = map {
567         m/^\Q${prefix}.\E(.+)$/ ? ($1) : ()
568     } keys %{$self->{collapse}}
569   } else {
570     @collapse = keys %{$self->{collapse}};
571   };
572
573   if (@collapse) {
574     my ($c) = sort { length $a <=> length $b } @collapse;
575     my $target = $info;
576     foreach my $p (split(/\./, $c)) {
577       $target = $target->[1]->{$p} ||= [];
578     }
579     my $c_prefix = (defined($prefix) ? "${prefix}.${c}" : $c);
580     my @co_key = @{$self->{collapse}{$c_prefix}};
581     my %co_check = map { ($_, $target->[0]->{$_}); } @co_key;
582     my $tree = $self->_collapse_result($as, $row, $c_prefix);
583     my (@final, @raw);
584     while ( !(grep {
585                 !defined($tree->[0]->{$_}) ||
586                 $co_check{$_} ne $tree->[0]->{$_}
587               } @co_key) ) {
588       push(@final, $tree);
589       last unless (@raw = $self->cursor->next);
590       $row = $self->{stashed_row} = \@raw;
591       $tree = $self->_collapse_result($as, $row, $c_prefix);
592       #warn Data::Dumper::Dumper($tree, $row);
593     }
594     @$target = @final;
595   }
596
597   return $info;
598 }
599
600 =head2 result_source
601
602 =over 4
603
604 =item Arguments: $result_source?
605
606 =item Return Value: $result_source
607
608 =back
609
610 An accessor for the primary ResultSource object from which this ResultSet
611 is derived.
612
613 =cut
614
615
616 =head2 count
617
618 =over 4
619
620 =item Arguments: ($cond, \%attrs?)?
621
622 =item Return Value: $count
623
624 =back
625
626 Performs an SQL C<COUNT> with the same query as the resultset was built
627 with to find the number of elements. If passed arguments, does a search
628 on the resultset and counts the results of that.
629
630 Note: When using C<count> with C<group_by>, L<DBIX::Class> emulates C<GROUP BY>
631 using C<COUNT( DISTINCT( columns ) )>. Some databases (notably SQLite) do
632 not support C<DISTINCT> with multiple columns. If you are using such a
633 database, you should only use columns from the main table in your C<group_by>
634 clause.
635
636 =cut
637
638 sub count {
639   my $self = shift;
640   return $self->search(@_)->count if @_ and defined $_[0];
641   return scalar @{ $self->get_cache } if @{ $self->get_cache };
642
643   my $count = $self->_count;
644   return 0 unless $count;
645
646   $count -= $self->{attrs}{offset} if $self->{attrs}{offset};
647   $count = $self->{attrs}{rows} if
648     $self->{attrs}{rows} and $self->{attrs}{rows} < $count;
649   return $count;
650 }
651
652 sub _count { # Separated out so pager can get the full count
653   my $self = shift;
654   my $select = { count => '*' };
655   my $attrs = { %{ $self->{attrs} } };
656   if (my $group_by = delete $attrs->{group_by}) {
657     delete $attrs->{having};
658     my @distinct = (ref $group_by ?  @$group_by : ($group_by));
659     # todo: try CONCAT for multi-column pk
660     my @pk = $self->result_source->primary_columns;
661     if (@pk == 1) {
662       foreach my $column (@distinct) {
663         if ($column =~ qr/^(?:\Q$attrs->{alias}.\E)?$pk[0]$/) {
664           @distinct = ($column);
665           last;
666         }
667       } 
668     }
669
670     $select = { count => { distinct => \@distinct } };
671     #use Data::Dumper; die Dumper $select;
672   }
673
674   $attrs->{select} = $select;
675   $attrs->{as} = [qw/count/];
676
677   # offset, order by and page are not needed to count. record_filter is cdbi
678   delete $attrs->{$_} for qw/rows offset order_by page pager record_filter/;
679         
680   my ($count) = (ref $self)->new($self->result_source, $attrs)->cursor->next;
681   return $count;
682 }
683
684 =head2 count_literal
685
686 =over 4
687
688 =item Arguments: $sql_fragment, @bind_values
689
690 =item Return Value: $count
691
692 =back
693
694 Counts the results in a literal query. Equivalent to calling L</search_literal>
695 with the passed arguments, then L</count>.
696
697 =cut
698
699 sub count_literal { shift->search_literal(@_)->count; }
700
701 =head2 all
702
703 =over 4
704
705 =item Arguments: none
706
707 =item Return Value: @objects
708
709 =back
710
711 Returns all elements in the resultset. Called implicitly if the resultset
712 is returned in list context.
713
714 =cut
715
716 sub all {
717   my ($self) = @_;
718   return @{ $self->get_cache } if @{ $self->get_cache };
719
720   my @obj;
721
722   if (keys %{$self->{collapse}}) {
723       # Using $self->cursor->all is really just an optimisation.
724       # If we're collapsing has_many prefetches it probably makes
725       # very little difference, and this is cleaner than hacking
726       # _construct_object to survive the approach
727     $self->cursor->reset;
728     my @row = $self->cursor->next;
729     while (@row) {
730       push(@obj, $self->_construct_object(@row));
731       @row = (exists $self->{stashed_row}
732                ? @{delete $self->{stashed_row}}
733                : $self->cursor->next);
734     }
735   } else {
736     @obj = map { $self->_construct_object(@$_) } $self->cursor->all;
737   }
738
739   $self->set_cache(\@obj) if $self->{attrs}{cache};
740   return @obj;
741 }
742
743 =head2 reset
744
745 =over 4
746
747 =item Arguments: none
748
749 =item Return Value: $self
750
751 =back
752
753 Resets the resultset's cursor, so you can iterate through the elements again.
754
755 =cut
756
757 sub reset {
758   my ($self) = @_;
759   $self->{all_cache_position} = 0;
760   $self->cursor->reset;
761   return $self;
762 }
763
764 =head2 first
765
766 =over 4
767
768 =item Arguments: none
769
770 =item Return Value: $object?
771
772 =back
773
774 Resets the resultset and returns an object for the first result (if the
775 resultset contains anything).
776
777 =cut
778
779 sub first {
780   return $_[0]->reset->next;
781 }
782
783 =head2 update
784
785 =over 4
786
787 =item Arguments: \%values
788
789 =item Return Value: $storage_rv
790
791 =back
792
793 Sets the specified columns in the resultset to the supplied values in a
794 single query. Return value will be true if the update succeeded or false
795 if no records were updated; exact type of success value is storage-dependent.
796
797 =cut
798
799 sub update {
800   my ($self, $values) = @_;
801   $self->throw_exception("Values for update must be a hash")
802     unless ref $values eq 'HASH';
803   return $self->result_source->storage->update(
804     $self->result_source->from, $values, $self->{cond}
805   );
806 }
807
808 =head2 update_all
809
810 =over 4
811
812 =item Arguments: \%values
813
814 =item Return Value: 1
815
816 =back
817
818 Fetches all objects and updates them one at a time.  Note that C<update_all>
819 will run cascade triggers while L</update> will not.
820
821 =cut
822
823 sub update_all {
824   my ($self, $values) = @_;
825   $self->throw_exception("Values for update must be a hash")
826     unless ref $values eq 'HASH';
827   foreach my $obj ($self->all) {
828     $obj->set_columns($values)->update;
829   }
830   return 1;
831 }
832
833 =head2 delete
834
835 =over 4
836
837 =item Arguments: none
838
839 =item Return Value: 1
840
841 =back
842
843 Deletes the contents of the resultset from its result source. Note that this
844 will not run cascade triggers. See L</delete_all> if you need triggers to run.
845
846 =cut
847
848 sub delete {
849   my ($self) = @_;
850   my $del = {};
851
852   if (!ref($self->{cond})) {
853
854     # No-op. No condition, we're deleting everything
855
856   } elsif (ref $self->{cond} eq 'ARRAY') {
857
858     $del = [ map { my %hash;
859       foreach my $key (keys %{$_}) {
860         $key =~ /([^.]+)$/;
861         $hash{$1} = $_->{$key};
862       }; \%hash; } @{$self->{cond}} ];
863
864   } elsif (ref $self->{cond} eq 'HASH') {
865
866     if ((keys %{$self->{cond}})[0] eq '-and') {
867
868       $del->{-and} = [ map { my %hash;
869         foreach my $key (keys %{$_}) {
870           $key =~ /([^.]+)$/;
871           $hash{$1} = $_->{$key};
872         }; \%hash; } @{$self->{cond}{-and}} ];
873
874     } else {
875
876       foreach my $key (keys %{$self->{cond}}) {
877         $key =~ /([^.]+)$/;
878         $del->{$1} = $self->{cond}{$key};
879       }
880     }
881
882   } else {
883     $self->throw_exception(
884       "Can't delete on resultset with condition unless hash or array"
885     );
886   }
887
888   $self->result_source->storage->delete($self->result_source->from, $del);
889   return 1;
890 }
891
892 =head2 delete_all
893
894 =over 4
895
896 =item Arguments: none
897
898 =item Return Value: 1
899
900 =back
901
902 Fetches all objects and deletes them one at a time.  Note that C<delete_all>
903 will run cascade triggers while L</delete> will not.
904
905 =cut
906
907 sub delete_all {
908   my ($self) = @_;
909   $_->delete for $self->all;
910   return 1;
911 }
912
913 =head2 pager
914
915 =over 4
916
917 =item Arguments: none
918
919 =item Return Value: $pager
920
921 =back
922
923 Return Value a L<Data::Page> object for the current resultset. Only makes
924 sense for queries with a C<page> attribute.
925
926 =cut
927
928 sub pager {
929   my ($self) = @_;
930   my $attrs = $self->{attrs};
931   $self->throw_exception("Can't create pager for non-paged rs")
932     unless $self->{page};
933   $attrs->{rows} ||= 10;
934   return $self->{pager} ||= Data::Page->new(
935     $self->_count, $attrs->{rows}, $self->{page});
936 }
937
938 =head2 page
939
940 =over 4
941
942 =item Arguments: $page_number
943
944 =item Return Value: $rs
945
946 =back
947
948 Returns a resultset for the $page_number page of the resultset on which page
949 is called, where each page contains a number of rows equal to the 'rows'
950 attribute set on the resultset, or 10 by default
951
952 =cut
953
954 sub page {
955   my ($self, $page) = @_;
956   my $attrs = { %{$self->{attrs}} };
957   $attrs->{page} = $page;
958   return (ref $self)->new($self->result_source, $attrs);
959 }
960
961 =head2 new_result
962
963 =over 4
964
965 =item Arguments: \%vals
966
967 =item Return Value: $object
968
969 =back
970
971 Creates an object in the resultset's result class and returns it.
972
973 =cut
974
975 sub new_result {
976   my ($self, $values) = @_;
977   $self->throw_exception( "new_result needs a hash" )
978     unless (ref $values eq 'HASH');
979   $self->throw_exception(
980     "Can't abstract implicit construct, condition not a hash"
981   ) if ($self->{cond} && !(ref $self->{cond} eq 'HASH'));
982   my %new = %$values;
983   my $alias = $self->{attrs}{alias};
984   foreach my $key (keys %{$self->{cond}||{}}) {
985     $new{$1} = $self->{cond}{$key} if ($key =~ m/^(?:\Q${alias}.\E)?([^.]+)$/);
986   }
987   my $obj = $self->result_class->new(\%new);
988   $obj->result_source($self->result_source) if $obj->can('result_source');
989   return $obj;
990 }
991
992 =head2 create
993
994 =over 4
995
996 =item Arguments: \%vals
997
998 =item Return Value: $object
999
1000 =back
1001
1002 Inserts a record into the resultset and returns the object representing it.
1003
1004 Effectively a shortcut for C<< ->new_result(\%vals)->insert >>.
1005
1006 =cut
1007
1008 sub create {
1009   my ($self, $attrs) = @_;
1010   $self->throw_exception( "create needs a hashref" )
1011     unless ref $attrs eq 'HASH';
1012   return $self->new_result($attrs)->insert;
1013 }
1014
1015 =head2 find_or_create
1016
1017 =over 4
1018
1019 =item Arguments: \%vals, \%attrs?
1020
1021 =item Return Value: $object
1022
1023 =back
1024
1025   $class->find_or_create({ key => $val, ... });
1026
1027 Searches for a record matching the search condition; if it doesn't find one,
1028 creates one and returns that instead.
1029
1030   my $cd = $schema->resultset('CD')->find_or_create({
1031     cdid   => 5,
1032     artist => 'Massive Attack',
1033     title  => 'Mezzanine',
1034     year   => 2005,
1035   });
1036
1037 Also takes an optional C<key> attribute, to search by a specific key or unique
1038 constraint. For example:
1039
1040   my $cd = $schema->resultset('CD')->find_or_create(
1041     {
1042       artist => 'Massive Attack',
1043       title  => 'Mezzanine',
1044     },
1045     { key => 'artist_title' }
1046   );
1047
1048 See also L</find> and L</update_or_create>.
1049
1050 =cut
1051
1052 sub find_or_create {
1053   my $self     = shift;
1054   my $attrs    = (@_ > 1 && ref $_[$#_] eq 'HASH' ? pop(@_) : {});
1055   my $hash     = ref $_[0] eq 'HASH' ? shift : {@_};
1056   my $exists   = $self->find($hash, $attrs);
1057   return defined $exists ? $exists : $self->create($hash);
1058 }
1059
1060 =head2 update_or_create
1061
1062 =over 4
1063
1064 =item Arguments: \%col_values, { key => $unique_constraint }?
1065
1066 =item Return Value: $object
1067
1068 =back
1069
1070   $class->update_or_create({ col => $val, ... });
1071
1072 First, search for an existing row matching one of the unique constraints
1073 (including the primary key) on the source of this resultset.  If a row is
1074 found, update it with the other given column values.  Otherwise, create a new
1075 row.
1076
1077 Takes an optional C<key> attribute to search on a specific unique constraint.
1078 For example:
1079
1080   # In your application
1081   my $cd = $schema->resultset('CD')->update_or_create(
1082     {
1083       artist => 'Massive Attack',
1084       title  => 'Mezzanine',
1085       year   => 1998,
1086     },
1087     { key => 'artist_title' }
1088   );
1089
1090 If no C<key> is specified, it searches on all unique constraints defined on the
1091 source, including the primary key.
1092
1093 If the C<key> is specified as C<primary>, search only on the primary key.
1094
1095 See also L</find> and L</find_or_create>.
1096
1097 =cut
1098
1099 sub update_or_create {
1100   my $self = shift;
1101   my $attrs = (@_ > 1 && ref $_[$#_] eq 'HASH' ? pop(@_) : {});
1102   my $hash = ref $_[0] eq 'HASH' ? shift : {@_};
1103
1104   my %unique_constraints = $self->result_source->unique_constraints;
1105   my @constraint_names   = (exists $attrs->{key}
1106                             ? ($attrs->{key})
1107                             : keys %unique_constraints);
1108
1109   my @unique_hashes;
1110   foreach my $name (@constraint_names) {
1111     my @unique_cols = @{ $unique_constraints{$name} };
1112     my %unique_hash =
1113       map  { $_ => $hash->{$_} }
1114       grep { exists $hash->{$_} }
1115       @unique_cols;
1116
1117     push @unique_hashes, \%unique_hash
1118       if (scalar keys %unique_hash == scalar @unique_cols);
1119   }
1120
1121   if (@unique_hashes) {
1122     my $row = $self->single(\@unique_hashes);
1123     if (defined $row) {
1124       $row->set_columns($hash);
1125       $row->update;
1126       return $row;
1127     }
1128   }
1129
1130   return $self->create($hash);
1131 }
1132
1133 =head2 get_cache
1134
1135 =over 4
1136
1137 =item Arguments: none
1138
1139 =item Return Value: \@cache_objects?
1140
1141 =back
1142
1143 Gets the contents of the cache for the resultset if the cache is set
1144
1145 =cut
1146
1147 sub get_cache {
1148   shift->{all_cache} || [];
1149 }
1150
1151 =head2 set_cache
1152
1153 =over 4
1154
1155 =item Arguments: \@cache_objects
1156
1157 =item Return Value: \@cache_objects
1158
1159 =back
1160
1161 Sets the contents of the cache for the resultset. Expects an arrayref
1162 of objects of the same class as those produced by the resultset. Note that
1163 if the cache is set the resultset will return the cached objects rather
1164 than re-querying the database even if the cache attr is not set.
1165
1166 =cut
1167
1168 sub set_cache {
1169   my ( $self, $data ) = @_;
1170   $self->throw_exception("set_cache requires an arrayref")
1171     if ref $data ne 'ARRAY';
1172   my $result_class = $self->result_class;
1173   foreach( @$data ) {
1174     $self->throw_exception(
1175       "cannot cache object of type '$_', expected '$result_class'"
1176     ) if ref $_ ne $result_class;
1177   }
1178   $self->{all_cache} = $data;
1179 }
1180
1181 =head2 clear_cache
1182
1183 =over 4
1184
1185 =item Arguments: none
1186
1187 =item Return Value: []
1188
1189 =back
1190
1191 Clears the cache for the resultset.
1192
1193 =cut
1194
1195 sub clear_cache {
1196   shift->set_cache([]);
1197 }
1198
1199 =head2 related_resultset
1200
1201 =over 4
1202
1203 =item Arguments: $relationship_name
1204
1205 =item Return Value: $resultset
1206
1207 =back
1208
1209 Returns a related resultset for the supplied relationship name.
1210
1211   $artist_rs = $schema->resultset('CD')->related_resultset('Artist');
1212
1213 =cut
1214
1215 sub related_resultset {
1216   my ( $self, $rel ) = @_;
1217   $self->{related_resultsets} ||= {};
1218   return $self->{related_resultsets}{$rel} ||= do {
1219       #warn "fetching related resultset for rel '$rel'";
1220       my $rel_obj = $self->result_source->relationship_info($rel);
1221       $self->throw_exception(
1222         "search_related: result source '" . $self->result_source->name .
1223         "' has no such relationship ${rel}")
1224         unless $rel_obj; #die Dumper $self->{attrs};
1225
1226       my $rs = $self->search(undef, { join => $rel });
1227       my $alias = defined $rs->{attrs}{seen_join}{$rel}
1228                     && $rs->{attrs}{seen_join}{$rel} > 1
1229                   ? join('_', $rel, $rs->{attrs}{seen_join}{$rel})
1230                   : $rel;
1231
1232       $self->result_source->schema->resultset($rel_obj->{class}
1233            )->search( undef,
1234              { %{$rs->{attrs}},
1235                alias => $alias,
1236                select => undef,
1237                as => undef }
1238            );
1239   };
1240 }
1241
1242 =head2 throw_exception
1243
1244 See L<DBIx::Class::Schema/throw_exception> for details.
1245
1246 =cut
1247
1248 sub throw_exception {
1249   my $self=shift;
1250   $self->result_source->schema->throw_exception(@_);
1251 }
1252
1253 # XXX: FIXME: Attributes docs need clearing up
1254
1255 =head1 ATTRIBUTES
1256
1257 The resultset takes various attributes that modify its behavior. Here's an
1258 overview of them:
1259
1260 =head2 order_by
1261
1262 =over 4
1263
1264 =item Value: ($order_by | \@order_by)
1265
1266 Which column(s) to order the results by. This is currently passed
1267 through directly to SQL, so you can give e.g. C<year DESC> for a
1268 descending order on the column `year'.
1269
1270 =head2 columns
1271
1272 =over 4
1273
1274 =item Value: \@columns
1275
1276 =back
1277
1278 Shortcut to request a particular set of columns to be retrieved.  Adds
1279 C<me.> onto the start of any column without a C<.> in it and sets C<select>
1280 from that, then auto-populates C<as> from C<select> as normal. (You may also
1281 use the C<cols> attribute, as in earlier versions of DBIC.)
1282
1283 =head2 include_columns
1284
1285 =over 4
1286
1287 =item Value: \@columns
1288
1289 =back
1290
1291 Shortcut to include additional columns in the returned results - for example
1292
1293   $schema->resultset('CD')->search(undef, {
1294     include_columns => ['artist.name'],
1295     join => ['artist']
1296   });
1297
1298 would return all CDs and include a 'name' column to the information
1299 passed to object inflation
1300
1301 =head2 select
1302
1303 =over 4
1304
1305 =item Value: \@select_columns
1306
1307 =back
1308
1309 Indicates which columns should be selected from the storage. You can use
1310 column names, or in the case of RDBMS back ends, function or stored procedure
1311 names:
1312
1313   $rs = $schema->resultset('Employee')->search(undef, {
1314     select => [
1315       'name',
1316       { count => 'employeeid' },
1317       { sum => 'salary' }
1318     ]
1319   });
1320
1321 When you use function/stored procedure names and do not supply an C<as>
1322 attribute, the column names returned are storage-dependent. E.g. MySQL would
1323 return a column named C<count(employeeid)> in the above example.
1324
1325 =head2 as
1326
1327 =over 4
1328
1329 =item Value: \@inflation_names
1330
1331 =back
1332
1333 Indicates column names for object inflation. This is used in conjunction with
1334 C<select>, usually when C<select> contains one or more function or stored
1335 procedure names:
1336
1337   $rs = $schema->resultset('Employee')->search(undef, {
1338     select => [
1339       'name',
1340       { count => 'employeeid' }
1341     ],
1342     as => ['name', 'employee_count'],
1343   });
1344
1345   my $employee = $rs->first(); # get the first Employee
1346
1347 If the object against which the search is performed already has an accessor
1348 matching a column name specified in C<as>, the value can be retrieved using
1349 the accessor as normal:
1350
1351   my $name = $employee->name();
1352
1353 If on the other hand an accessor does not exist in the object, you need to
1354 use C<get_column> instead:
1355
1356   my $employee_count = $employee->get_column('employee_count');
1357
1358 You can create your own accessors if required - see
1359 L<DBIx::Class::Manual::Cookbook> for details.
1360
1361 =head2 join
1362
1363 =over 4
1364
1365 =item Value: ($rel_name | \@rel_names | \%rel_names)
1366
1367 =back
1368
1369 Contains a list of relationships that should be joined for this query.  For
1370 example:
1371
1372   # Get CDs by Nine Inch Nails
1373   my $rs = $schema->resultset('CD')->search(
1374     { 'artist.name' => 'Nine Inch Nails' },
1375     { join => 'artist' }
1376   );
1377
1378 Can also contain a hash reference to refer to the other relation's relations.
1379 For example:
1380
1381   package MyApp::Schema::Track;
1382   use base qw/DBIx::Class/;
1383   __PACKAGE__->table('track');
1384   __PACKAGE__->add_columns(qw/trackid cd position title/);
1385   __PACKAGE__->set_primary_key('trackid');
1386   __PACKAGE__->belongs_to(cd => 'MyApp::Schema::CD');
1387   1;
1388
1389   # In your application
1390   my $rs = $schema->resultset('Artist')->search(
1391     { 'track.title' => 'Teardrop' },
1392     {
1393       join     => { cd => 'track' },
1394       order_by => 'artist.name',
1395     }
1396   );
1397
1398 If the same join is supplied twice, it will be aliased to <rel>_2 (and
1399 similarly for a third time). For e.g.
1400
1401   my $rs = $schema->resultset('Artist')->search({
1402     'cds.title'   => 'Down to Earth',
1403     'cds_2.title' => 'Popular',
1404   }, {
1405     join => [ qw/cds cds/ ],
1406   });
1407
1408 will return a set of all artists that have both a cd with title 'Down
1409 to Earth' and a cd with title 'Popular'.
1410
1411 If you want to fetch related objects from other tables as well, see C<prefetch>
1412 below.
1413
1414 =head2 prefetch
1415
1416 =over 4
1417
1418 =item Value: ($rel_name | \@rel_names | \%rel_names)
1419
1420 =back
1421
1422 Contains one or more relationships that should be fetched along with the main 
1423 query (when they are accessed afterwards they will have already been
1424 "prefetched").  This is useful for when you know you will need the related
1425 objects, because it saves at least one query:
1426
1427   my $rs = $schema->resultset('Tag')->search(
1428     undef,
1429     {
1430       prefetch => {
1431         cd => 'artist'
1432       }
1433     }
1434   );
1435
1436 The initial search results in SQL like the following:
1437
1438   SELECT tag.*, cd.*, artist.* FROM tag
1439   JOIN cd ON tag.cd = cd.cdid
1440   JOIN artist ON cd.artist = artist.artistid
1441
1442 L<DBIx::Class> has no need to go back to the database when we access the
1443 C<cd> or C<artist> relationships, which saves us two SQL statements in this
1444 case.
1445
1446 Simple prefetches will be joined automatically, so there is no need
1447 for a C<join> attribute in the above search. If you're prefetching to
1448 depth (e.g. { cd => { artist => 'label' } or similar), you'll need to
1449 specify the join as well.
1450
1451 C<prefetch> can be used with the following relationship types: C<belongs_to>,
1452 C<has_one> (or if you're using C<add_relationship>, any relationship declared
1453 with an accessor type of 'single' or 'filter').
1454
1455 =head2 from
1456
1457 =over 4
1458
1459 =item Value: \@from_clause
1460
1461 =back
1462
1463 The C<from> attribute gives you manual control over the C<FROM> clause of SQL
1464 statements generated by L<DBIx::Class>, allowing you to express custom C<JOIN>
1465 clauses.
1466
1467 NOTE: Use this on your own risk.  This allows you to shoot off your foot!
1468 C<join> will usually do what you need and it is strongly recommended that you
1469 avoid using C<from> unless you cannot achieve the desired result using C<join>.
1470
1471 In simple terms, C<from> works as follows:
1472
1473     [
1474         { <alias> => <table>, -join-type => 'inner|left|right' }
1475         [] # nested JOIN (optional)
1476         { <table.column> => <foreign_table.foreign_key> }
1477     ]
1478
1479     JOIN
1480         <alias> <table>
1481         [JOIN ...]
1482     ON <table.column> = <foreign_table.foreign_key>
1483
1484 An easy way to follow the examples below is to remember the following:
1485
1486     Anything inside "[]" is a JOIN
1487     Anything inside "{}" is a condition for the enclosing JOIN
1488
1489 The following examples utilize a "person" table in a family tree application.
1490 In order to express parent->child relationships, this table is self-joined:
1491
1492     # Person->belongs_to('father' => 'Person');
1493     # Person->belongs_to('mother' => 'Person');
1494
1495 C<from> can be used to nest joins. Here we return all children with a father,
1496 then search against all mothers of those children:
1497
1498   $rs = $schema->resultset('Person')->search(
1499       undef,
1500       {
1501           alias => 'mother', # alias columns in accordance with "from"
1502           from => [
1503               { mother => 'person' },
1504               [
1505                   [
1506                       { child => 'person' },
1507                       [
1508                           { father => 'person' },
1509                           { 'father.person_id' => 'child.father_id' }
1510                       ]
1511                   ],
1512                   { 'mother.person_id' => 'child.mother_id' }
1513               ],
1514           ]
1515       },
1516   );
1517
1518   # Equivalent SQL:
1519   # SELECT mother.* FROM person mother
1520   # JOIN (
1521   #   person child
1522   #   JOIN person father
1523   #   ON ( father.person_id = child.father_id )
1524   # )
1525   # ON ( mother.person_id = child.mother_id )
1526
1527 The type of any join can be controlled manually. To search against only people
1528 with a father in the person table, we could explicitly use C<INNER JOIN>:
1529
1530     $rs = $schema->resultset('Person')->search(
1531         undef,
1532         {
1533             alias => 'child', # alias columns in accordance with "from"
1534             from => [
1535                 { child => 'person' },
1536                 [
1537                     { father => 'person', -join-type => 'inner' },
1538                     { 'father.id' => 'child.father_id' }
1539                 ],
1540             ]
1541         },
1542     );
1543
1544     # Equivalent SQL:
1545     # SELECT child.* FROM person child
1546     # INNER JOIN person father ON child.father_id = father.id
1547
1548 =head2 page
1549
1550 =over 4
1551
1552 =item Value: $page
1553
1554 =back
1555
1556 Makes the resultset paged and specifies the page to retrieve. Effectively
1557 identical to creating a non-pages resultset and then calling ->page($page)
1558 on it.
1559
1560 =head2 rows
1561
1562 =over 4
1563
1564 =item Value: $rows
1565
1566 =back
1567
1568 Specifes the maximum number of rows for direct retrieval or the number of
1569 rows per page if the page attribute or method is used.
1570
1571 =head2 group_by
1572
1573 =over 4
1574
1575 =item Value: \@columns
1576
1577 =back
1578
1579 A arrayref of columns to group by. Can include columns of joined tables.
1580
1581   group_by => [qw/ column1 column2 ... /]
1582
1583 =head2 distinct
1584
1585 =over 4
1586
1587 =item Value: (0 | 1)
1588
1589 =back
1590
1591 Set to 1 to group by all columns.
1592
1593 =head2 cache
1594
1595 Set to 1 to cache search results. This prevents extra SQL queries if you
1596 revisit rows in your ResultSet:
1597
1598   my $resultset = $schema->resultset('Artist')->search( undef, { cache => 1 } );
1599   
1600   while( my $artist = $resultset->next ) {
1601     ... do stuff ...
1602   }
1603
1604   $rs->first; # without cache, this would issue a query 
1605
1606 By default, searches are not cached.
1607
1608 For more examples of using these attributes, see
1609 L<DBIx::Class::Manual::Cookbook>.
1610
1611 =cut
1612
1613 1;