fork tests/code improved, ithreads tests/code added, version bumped
[dbsrgits/DBIx-Class.git] / lib / DBIx / Class / Storage / DBI.pm
1 package DBIx::Class::Storage::DBI;
2
3 use base 'DBIx::Class::Storage';
4
5 use strict;
6 use warnings;
7 use DBI;
8 use SQL::Abstract::Limit;
9 use DBIx::Class::Storage::DBI::Cursor;
10 use IO::File;
11 use Carp::Clan qw/DBIx::Class/;
12
13 BEGIN {
14
15 package DBIC::SQL::Abstract; # Would merge upstream, but nate doesn't reply :(
16
17 use base qw/SQL::Abstract::Limit/;
18
19 sub select {
20   my ($self, $table, $fields, $where, $order, @rest) = @_;
21   @rest = (-1) unless defined $rest[0];
22   local $self->{having_bind} = [];
23   my ($sql, @ret) = $self->SUPER::select($table,
24                       $self->_recurse_fields($fields), $where, $order, @rest);
25   return wantarray ? ($sql, @ret, @{$self->{having_bind}}) : $sql;
26 }
27
28 sub _emulate_limit {
29   my $self = shift;
30   if ($_[3] == -1) {
31     return $_[1].$self->_order_by($_[2]);
32   } else {
33     return $self->SUPER::_emulate_limit(@_);
34   }
35 }
36
37 sub _recurse_fields {
38   my ($self, $fields) = @_;
39   my $ref = ref $fields;
40   return $self->_quote($fields) unless $ref;
41   return $$fields if $ref eq 'SCALAR';
42
43   if ($ref eq 'ARRAY') {
44     return join(', ', map { $self->_recurse_fields($_) } @$fields);
45   } elsif ($ref eq 'HASH') {
46     foreach my $func (keys %$fields) {
47       return $self->_sqlcase($func)
48         .'( '.$self->_recurse_fields($fields->{$func}).' )';
49     }
50   }
51 }
52
53 sub _order_by {
54   my $self = shift;
55   my $ret = '';
56   my @extra;
57   if (ref $_[0] eq 'HASH') {
58     if (defined $_[0]->{group_by}) {
59       $ret = $self->_sqlcase(' group by ')
60                .$self->_recurse_fields($_[0]->{group_by});
61     }
62     if (defined $_[0]->{having}) {
63       my $frag;
64       ($frag, @extra) = $self->_recurse_where($_[0]->{having});
65       push(@{$self->{having_bind}}, @extra);
66       $ret .= $self->_sqlcase(' having ').$frag;
67     }
68     if (defined $_[0]->{order_by}) {
69       $ret .= $self->SUPER::_order_by($_[0]->{order_by});
70     }
71   } elsif(ref $_[0] eq 'SCALAR') {
72     $ret = $self->_sqlcase(' order by ').${ $_[0] };
73   } else {
74     $ret = $self->SUPER::_order_by(@_);
75   }
76   return $ret;
77 }
78
79 sub _order_directions {
80   my ($self, $order) = @_;
81   $order = $order->{order_by} if ref $order eq 'HASH';
82   return $self->SUPER::_order_directions($order);
83 }
84
85 sub _table {
86   my ($self, $from) = @_;
87   if (ref $from eq 'ARRAY') {
88     return $self->_recurse_from(@$from);
89   } elsif (ref $from eq 'HASH') {
90     return $self->_make_as($from);
91   } else {
92     return $from;
93   }
94 }
95
96 sub _recurse_from {
97   my ($self, $from, @join) = @_;
98   my @sqlf;
99   push(@sqlf, $self->_make_as($from));
100   foreach my $j (@join) {
101     my ($to, $on) = @$j;
102
103     # check whether a join type exists
104     my $join_clause = '';
105     if (ref($to) eq 'HASH' and exists($to->{-join_type})) {
106       $join_clause = ' '.uc($to->{-join_type}).' JOIN ';
107     } else {
108       $join_clause = ' JOIN ';
109     }
110     push(@sqlf, $join_clause);
111
112     if (ref $to eq 'ARRAY') {
113       push(@sqlf, '(', $self->_recurse_from(@$to), ')');
114     } else {
115       push(@sqlf, $self->_make_as($to));
116     }
117     push(@sqlf, ' ON ', $self->_join_condition($on));
118   }
119   return join('', @sqlf);
120 }
121
122 sub _make_as {
123   my ($self, $from) = @_;
124   return join(' ', map { (ref $_ eq 'SCALAR' ? $$_ : $self->_quote($_)) }
125                            reverse each %{$self->_skip_options($from)});
126 }
127
128 sub _skip_options {
129   my ($self, $hash) = @_;
130   my $clean_hash = {};
131   $clean_hash->{$_} = $hash->{$_}
132     for grep {!/^-/} keys %$hash;
133   return $clean_hash;
134 }
135
136 sub _join_condition {
137   my ($self, $cond) = @_;
138   if (ref $cond eq 'HASH') {
139     my %j;
140     for (keys %$cond) { my $x = '= '.$self->_quote($cond->{$_}); $j{$_} = \$x; };
141     return $self->_recurse_where(\%j);
142   } elsif (ref $cond eq 'ARRAY') {
143     return join(' OR ', map { $self->_join_condition($_) } @$cond);
144   } else {
145     die "Can't handle this yet!";
146   }
147 }
148
149 sub _quote {
150   my ($self, $label) = @_;
151   return '' unless defined $label;
152   return "*" if $label eq '*';
153   return $label unless $self->{quote_char};
154   if(ref $self->{quote_char} eq "ARRAY"){
155     return $self->{quote_char}->[0] . $label . $self->{quote_char}->[1]
156       if !defined $self->{name_sep};
157     my $sep = $self->{name_sep};
158     return join($self->{name_sep},
159         map { $self->{quote_char}->[0] . $_ . $self->{quote_char}->[1]  }
160        split(/\Q$sep\E/,$label));
161   }
162   return $self->SUPER::_quote($label);
163 }
164
165 sub _RowNum {
166    my $self = shift;
167    my $c;
168    $_[0] =~ s/SELECT (.*?) FROM/
169      'SELECT '.join(', ', map { $_.' AS col'.++$c } split(', ', $1)).' FROM'/e;
170    $self->SUPER::_RowNum(@_);
171 }
172
173 # Accessor for setting limit dialect. This is useful
174 # for JDBC-bridge among others where the remote SQL-dialect cannot
175 # be determined by the name of the driver alone.
176 #
177 sub limit_dialect {
178     my $self = shift;
179     $self->{limit_dialect} = shift if @_;
180     return $self->{limit_dialect};
181 }
182
183 sub quote_char {
184     my $self = shift;
185     $self->{quote_char} = shift if @_;
186     return $self->{quote_char};
187 }
188
189 sub name_sep {
190     my $self = shift;
191     $self->{name_sep} = shift if @_;
192     return $self->{name_sep};
193 }
194
195
196
197
198 package DBIx::Class::Storage::DBI::DebugCallback;
199
200 sub print {
201   my ($self, $string) = @_;
202   $string =~ m/^(\w+)/;
203   ${$self}->($1, $string);
204 }
205
206 } # End of BEGIN block
207
208 use base qw/DBIx::Class/;
209
210 __PACKAGE__->load_components(qw/AccessorGroup/);
211
212 __PACKAGE__->mk_group_accessors('simple' =>
213   qw/connect_info _dbh _sql_maker _conn_pid _conn_tid debug debugfh
214      cursor on_connect_do transaction_depth/);
215
216 sub new {
217   my $new = bless({}, ref $_[0] || $_[0]);
218   $new->cursor("DBIx::Class::Storage::DBI::Cursor");
219   $new->transaction_depth(0);
220   if (defined($ENV{DBIX_CLASS_STORAGE_DBI_DEBUG}) &&
221      ($ENV{DBIX_CLASS_STORAGE_DBI_DEBUG} =~ /=(.+)$/)) {
222     $new->debugfh(IO::File->new($1, 'w')) || $new->throw_exception("Cannot open trace file $1");
223   } else {
224     $new->debugfh(IO::File->new('>&STDERR'));
225   }
226   $new->debug(1) if $ENV{DBIX_CLASS_STORAGE_DBI_DEBUG};
227   return $new;
228 }
229
230 sub throw_exception {
231   my ($self, $msg) = @_;
232   croak($msg);
233 }
234
235 =head1 NAME 
236
237 DBIx::Class::Storage::DBI - DBI storage handler
238
239 =head1 SYNOPSIS
240
241 =head1 DESCRIPTION
242
243 This class represents the connection to the database
244
245 =head1 METHODS
246
247 =cut
248
249 =head2 on_connect_do
250
251 Executes the sql statements given as a listref on every db connect.
252
253 =head2 debug
254
255 Causes SQL trace information to be emitted on C<debugfh> filehandle
256 (or C<STDERR> if C<debugfh> has not specifically been set).
257
258 =head2 debugfh
259
260 Sets or retrieves the filehandle used for trace/debug output.  This
261 should be an IO::Handle compatible object (only the C<print> method is
262 used).  Initially set to be STDERR - although see information on the
263 L<DBIX_CLASS_STORAGE_DBI_DEBUG> environment variable.
264
265 =head2 debugcb
266
267 Sets a callback to be executed each time a statement is run; takes a sub
268 reference. Overrides debugfh. Callback is executed as $sub->($op, $info)
269 where $op is SELECT/INSERT/UPDATE/DELETE and $info is what would normally
270 be printed.
271
272 =cut
273
274 sub debugcb {
275   my ($self, $cb) = @_;
276   my $cb_obj = bless(\$cb, 'DBIx::Class::Storage::DBI::DebugCallback');
277   $self->debugfh($cb_obj);
278 }
279
280 sub disconnect {
281   my ($self) = @_;
282
283   if( $self->connected ) {
284     $self->_dbh->rollback unless $self->_dbh->{AutoCommit};
285     $self->_dbh->disconnect;
286     $self->_dbh(undef);
287   }
288 }
289
290 sub connected {
291   my ($self) = @_;
292
293   if(my $dbh = $self->_dbh) {
294       if(defined $self->_conn_tid && $self->_conn_tid != threads->tid) {
295           $self->_sql_maker(undef);
296           return $self->_dbh(undef);
297       }
298       elsif($self->_conn_pid != $$) {
299           $self->_dbh->{InactiveDestroy} = 1;
300           $self->_sql_maker(undef);
301           return $self->_dbh(undef)
302       }
303       return ($dbh->FETCH('Active') && $dbh->ping);
304   }
305
306   return 0;
307 }
308
309 sub ensure_connected {
310   my ($self) = @_;
311
312   unless ($self->connected) {
313     $self->_populate_dbh;
314   }
315 }
316
317 sub dbh {
318   my ($self) = @_;
319
320   $self->ensure_connected;
321   return $self->_dbh;
322 }
323
324 sub sql_maker {
325   my ($self) = @_;
326   unless ($self->_sql_maker) {
327     $self->_sql_maker(new DBIC::SQL::Abstract( limit_dialect => $self->dbh ));
328   }
329   return $self->_sql_maker;
330 }
331
332 sub _populate_dbh {
333   my ($self) = @_;
334   my @info = @{$self->connect_info || []};
335   $self->_dbh($self->_connect(@info));
336   my $driver = $self->_dbh->{Driver}->{Name};
337   eval "require DBIx::Class::Storage::DBI::${driver}";
338   unless ($@) {
339     bless $self, "DBIx::Class::Storage::DBI::${driver}";
340   }
341   # if on-connect sql statements are given execute them
342   foreach my $sql_statement (@{$self->on_connect_do || []}) {
343     $self->_dbh->do($sql_statement);
344   }
345
346   $self->_conn_pid($$);
347   $self->_conn_tid(threads->tid) if $INC{'threads.pm'};
348 }
349
350 sub _connect {
351   my ($self, @info) = @_;
352
353   my ($old_connect_via, $dbh);
354
355   if ($INC{'Apache/DBI.pm'} && $ENV{MOD_PERL}) {
356       $old_connect_via = $DBI::connect_via;
357       $DBI::connect_via = 'connect';
358   }
359
360   if(ref $info[0] eq 'CODE') {
361       $dbh = &{$info[0]};
362   }
363   else {
364       $dbh = DBI->connect(@info);
365   }
366
367   $DBI::connect_via = $old_connect_via if $old_connect_via;
368
369   $self->throw_exception("DBI Connection failed: $DBI::errstr")
370       unless $dbh;
371
372   $dbh;
373 }
374
375 =head2 txn_begin
376
377 Calls begin_work on the current dbh.
378
379 See L<DBIx::Class::Schema> for the txn_do() method, which allows for
380 an entire code block to be executed transactionally.
381
382 =cut
383
384 sub txn_begin {
385   my $self = shift;
386   $self->dbh->begin_work
387     if $self->{transaction_depth}++ == 0 and $self->dbh->{AutoCommit};
388 }
389
390 =head2 txn_commit
391
392 Issues a commit against the current dbh.
393
394 =cut
395
396 sub txn_commit {
397   my $self = shift;
398   if ($self->{transaction_depth} == 0) {
399     $self->dbh->commit unless $self->dbh->{AutoCommit};
400   }
401   else {
402     $self->dbh->commit if --$self->{transaction_depth} == 0;    
403   }
404 }
405
406 =head2 txn_rollback
407
408 Issues a rollback against the current dbh. A nested rollback will
409 throw a L<DBIx::Class::Storage::NESTED_ROLLBACK_EXCEPTION> exception,
410 which allows the rollback to propagate to the outermost transaction.
411
412 =cut
413
414 sub txn_rollback {
415   my $self = shift;
416
417   eval {
418     if ($self->{transaction_depth} == 0) {
419       $self->dbh->rollback unless $self->dbh->{AutoCommit};
420     }
421     else {
422       --$self->{transaction_depth} == 0 ?
423         $self->dbh->rollback :
424         die DBIx::Class::Storage::NESTED_ROLLBACK_EXCEPTION->new;
425     }
426   };
427
428   if ($@) {
429     my $error = $@;
430     my $exception_class = "DBIx::Class::Storage::NESTED_ROLLBACK_EXCEPTION";
431     $error =~ /$exception_class/ and $self->throw_exception($error);
432     $self->{transaction_depth} = 0;          # ensure that a failed rollback
433     $self->throw_exception($error);          # resets the transaction depth
434   }
435 }
436
437 sub _execute {
438   my ($self, $op, $extra_bind, $ident, @args) = @_;
439   my ($sql, @bind) = $self->sql_maker->$op($ident, @args);
440   unshift(@bind, @$extra_bind) if $extra_bind;
441   if ($self->debug) {
442       my @debug_bind = map { defined $_ ? qq{`$_'} : q{`NULL'} } @bind;
443       $self->debugfh->print("$sql: " . join(', ', @debug_bind) . "\n");
444   }
445   my $sth = $self->sth($sql,$op);
446   $self->throw_exception("no sth generated via sql: $sql") unless $sth;
447   @bind = map { ref $_ ? ''.$_ : $_ } @bind; # stringify args
448   my $rv;
449   if ($sth) {  
450     $rv = $sth->execute(@bind) or $self->throw_exception("Error executing '$sql': " . $sth->errstr);
451   } else { 
452     $self->throw_exception("'$sql' did not generate a statement.");
453   }
454   return (wantarray ? ($rv, $sth, @bind) : $rv);
455 }
456
457 sub insert {
458   my ($self, $ident, $to_insert) = @_;
459   $self->throw_exception( "Couldn't insert ".join(', ', map "$_ => $to_insert->{$_}", keys %$to_insert)." into ${ident}" )
460     unless ($self->_execute('insert' => [], $ident, $to_insert));
461   return $to_insert;
462 }
463
464 sub update {
465   return shift->_execute('update' => [], @_);
466 }
467
468 sub delete {
469   return shift->_execute('delete' => [], @_);
470 }
471
472 sub _select {
473   my ($self, $ident, $select, $condition, $attrs) = @_;
474   my $order = $attrs->{order_by};
475   if (ref $condition eq 'SCALAR') {
476     $order = $1 if $$condition =~ s/ORDER BY (.*)$//i;
477   }
478   if (exists $attrs->{group_by} || $attrs->{having}) {
479     $order = { group_by => $attrs->{group_by},
480                having => $attrs->{having},
481                ($order ? (order_by => $order) : ()) };
482   }
483   my @args = ('select', $attrs->{bind}, $ident, $select, $condition, $order);
484   if ($attrs->{software_limit} ||
485       $self->sql_maker->_default_limit_syntax eq "GenericSubQ") {
486         $attrs->{software_limit} = 1;
487   } else {
488     push @args, $attrs->{rows}, $attrs->{offset};
489   }
490   return $self->_execute(@args);
491 }
492
493 sub select {
494   my $self = shift;
495   my ($ident, $select, $condition, $attrs) = @_;
496   return $self->cursor->new($self, \@_, $attrs);
497 }
498
499 # Need to call finish() to work round broken DBDs
500
501 sub select_single {
502   my $self = shift;
503   my ($rv, $sth, @bind) = $self->_select(@_);
504   my @row = $sth->fetchrow_array;
505   $sth->finish();
506   return @row;
507 }
508
509 sub sth {
510   my ($self, $sql) = @_;
511   # 3 is the if_active parameter which avoids active sth re-use
512   return $self->dbh->prepare_cached($sql, {}, 3);
513 }
514
515 =head2 columns_info_for
516
517 Returns database type info for a given table columns.
518
519 =cut
520
521 sub columns_info_for {
522     my ($self, $table) = @_;
523     my %result;
524     if ( $self->dbh->can( 'column_info' ) ){
525         my $sth = $self->dbh->column_info( undef, undef, $table, '%' );
526         $sth->execute();
527         while ( my $info = $sth->fetchrow_hashref() ){
528             my %column_info;
529             $column_info{data_type} = $info->{TYPE_NAME};
530             $column_info{size} = $info->{COLUMN_SIZE};
531             $column_info{is_nullable} = $info->{NULLABLE};
532             $result{$info->{COLUMN_NAME}} = \%column_info;
533         }
534     } else {
535         my $sth = $self->dbh->prepare("SELECT * FROM $table WHERE 1=0");
536         $sth->execute;
537         my @columns = @{$sth->{NAME}};
538         for my $i ( 0 .. $#columns ){
539             $result{$columns[$i]}{data_type} = $sth->{TYPE}->[$i];
540         }
541     }
542     return \%result;
543 }
544
545 sub last_insert_id {
546   my ($self, $row) = @_;
547     
548   return $self->dbh->func('last_insert_rowid');
549
550 }
551
552 sub sqlt_type { shift->dbh->{Driver}->{Name} }
553
554 sub deployment_statements {
555   my ($self, $schema, $type, $sqltargs) = @_;
556   $type ||= $self->sqlt_type;
557   eval "use SQL::Translator";
558   $self->throw_exception("Can't deploy without SQL::Translator: $@") if $@;
559   eval "use SQL::Translator::Parser::DBIx::Class;";
560   $self->throw_exception($@) if $@; 
561   eval "use SQL::Translator::Producer::${type};";
562   $self->throw_exception($@) if $@;
563   my $tr = SQL::Translator->new(%$sqltargs);
564   SQL::Translator::Parser::DBIx::Class::parse( $tr, $schema );
565   return "SQL::Translator::Producer::${type}"->can('produce')->($tr);
566 }
567
568 sub deploy {
569   my ($self, $schema, $type, $sqltargs) = @_;
570   foreach(split(";\n", $self->deployment_statements($schema, $type, $sqltargs))) {
571       $self->debugfh->print("$_\n") if $self->debug;
572           $self->dbh->do($_) or warn "SQL was:\n $_";
573   } 
574 }
575
576 sub DESTROY { shift->disconnect }
577
578 1;
579
580 =head1 ENVIRONMENT VARIABLES
581
582 =head2 DBIX_CLASS_STORAGE_DBI_DEBUG
583
584 If C<DBIX_CLASS_STORAGE_DBI_DEBUG> is set then SQL trace information
585 is produced (as when the L<debug> method is set).
586
587 If the value is of the form C<1=/path/name> then the trace output is
588 written to the file C</path/name>.
589
590 =head1 AUTHORS
591
592 Matt S. Trout <mst@shadowcatsystems.co.uk>
593
594 Andy Grundman <andy@hybridized.org>
595
596 =head1 LICENSE
597
598 You may distribute this code under the same terms as Perl itself.
599
600 =cut
601