Merge 'DBIx-Class-current' into 'trunk'
[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 $label unless $self->{quote_char};
153   return $self->SUPER::_quote($label);
154 }
155
156 sub _RowNum {
157    my $self = shift;
158    my $c;
159    $_[0] =~ s/SELECT (.*?) FROM/
160      'SELECT '.join(', ', map { $_.' AS col'.++$c } split(', ', $1)).' FROM'/e;
161    $self->SUPER::_RowNum(@_);
162 }
163
164 # Accessor for setting limit dialect. This is useful
165 # for JDBC-bridge among others where the remote SQL-dialect cannot
166 # be determined by the name of the driver alone.
167 #
168 sub limit_dialect {
169     my $self = shift;
170     $self->{limit_dialect} = shift if @_;
171     return $self->{limit_dialect};
172 }
173
174 sub quote_char {
175     my $self = shift;
176     $self->{quote_char} = shift if @_;
177     return $self->{quote_char};
178 }
179
180 sub name_sep {
181     my $self = shift;
182     $self->{name_sep} = shift if @_;
183     return $self->{name_sep};
184 }
185
186
187
188
189 package DBIx::Class::Storage::DBI::DebugCallback;
190
191 sub print {
192   my ($self, $string) = @_;
193   $string =~ m/^(\w+)/;
194   ${$self}->($1, $string);
195 }
196
197 } # End of BEGIN block
198
199 use base qw/DBIx::Class/;
200
201 __PACKAGE__->load_components(qw/AccessorGroup/);
202
203 __PACKAGE__->mk_group_accessors('simple' =>
204   qw/connect_info _dbh _sql_maker _conn_pid _conn_tid debug debugfh
205      cursor on_connect_do transaction_depth/);
206
207 sub new {
208   my $new = bless({}, ref $_[0] || $_[0]);
209   $new->cursor("DBIx::Class::Storage::DBI::Cursor");
210   $new->transaction_depth(0);
211   if (defined($ENV{DBIX_CLASS_STORAGE_DBI_DEBUG}) &&
212      ($ENV{DBIX_CLASS_STORAGE_DBI_DEBUG} =~ /=(.+)$/)) {
213     $new->debugfh(IO::File->new($1, 'w')) || $new->throw_exception("Cannot open trace file $1");
214   } else {
215     $new->debugfh(IO::File->new('>&STDERR'));
216   }
217   $new->debug(1) if $ENV{DBIX_CLASS_STORAGE_DBI_DEBUG};
218   return $new;
219 }
220
221 sub throw_exception {
222   my ($self, $msg) = @_;
223   croak($msg);
224 }
225
226 =head1 NAME 
227
228 DBIx::Class::Storage::DBI - DBI storage handler
229
230 =head1 SYNOPSIS
231
232 =head1 DESCRIPTION
233
234 This class represents the connection to the database
235
236 =head1 METHODS
237
238 =cut
239
240 =head2 on_connect_do
241
242 Executes the sql statements given as a listref on every db connect.
243
244 =head2 debug
245
246 Causes SQL trace information to be emitted on C<debugfh> filehandle
247 (or C<STDERR> if C<debugfh> has not specifically been set).
248
249 =head2 debugfh
250
251 Sets or retrieves the filehandle used for trace/debug output.  This
252 should be an IO::Handle compatible object (only the C<print> method is
253 used).  Initially set to be STDERR - although see information on the
254 L<DBIX_CLASS_STORAGE_DBI_DEBUG> environment variable.
255
256 =head2 debugcb
257
258 Sets a callback to be executed each time a statement is run; takes a sub
259 reference. Overrides debugfh. Callback is executed as $sub->($op, $info)
260 where $op is SELECT/INSERT/UPDATE/DELETE and $info is what would normally
261 be printed.
262
263 =cut
264
265 sub debugcb {
266   my ($self, $cb) = @_;
267   my $cb_obj = bless(\$cb, 'DBIx::Class::Storage::DBI::DebugCallback');
268   $self->debugfh($cb_obj);
269 }
270
271 sub disconnect {
272   my ($self) = @_;
273
274   if( $self->connected ) {
275     $self->_dbh->rollback unless $self->_dbh->{AutoCommit};
276     $self->_dbh->disconnect;
277     $self->_dbh(undef);
278   }
279 }
280
281 sub connected {
282   my ($self) = @_;
283
284   if(my $dbh = $self->_dbh) {
285       if(defined $self->_conn_tid && $self->_conn_tid != threads->tid) {
286           $self->_sql_maker(undef);
287           return $self->_dbh(undef);
288       }
289       elsif($self->_conn_pid != $$) {
290           $self->_dbh->{InactiveDestroy} = 1;
291           $self->_sql_maker(undef);
292           return $self->_dbh(undef)
293       }
294       return ($dbh->FETCH('Active') && $dbh->ping);
295   }
296
297   return 0;
298 }
299
300 sub ensure_connected {
301   my ($self) = @_;
302
303   unless ($self->connected) {
304     $self->_populate_dbh;
305   }
306 }
307
308 sub dbh {
309   my ($self) = @_;
310
311   $self->ensure_connected;
312   return $self->_dbh;
313 }
314
315 sub sql_maker {
316   my ($self) = @_;
317   unless ($self->_sql_maker) {
318     $self->_sql_maker(new DBIC::SQL::Abstract( limit_dialect => $self->dbh ));
319   }
320   return $self->_sql_maker;
321 }
322
323 sub _populate_dbh {
324   my ($self) = @_;
325   my @info = @{$self->connect_info || []};
326   $self->_dbh($self->_connect(@info));
327   my $driver = $self->_dbh->{Driver}->{Name};
328   eval "require DBIx::Class::Storage::DBI::${driver}";
329   unless ($@) {
330     bless $self, "DBIx::Class::Storage::DBI::${driver}";
331   }
332   # if on-connect sql statements are given execute them
333   foreach my $sql_statement (@{$self->on_connect_do || []}) {
334     $self->_dbh->do($sql_statement);
335   }
336
337   $self->_conn_pid($$);
338   $self->_conn_tid(threads->tid) if $INC{'threads.pm'};
339 }
340
341 sub _connect {
342   my ($self, @info) = @_;
343
344   $self->throw_exception("You failed to provide any connection info")
345       if !@info;
346
347   my ($old_connect_via, $dbh);
348
349   if ($INC{'Apache/DBI.pm'} && $ENV{MOD_PERL}) {
350       $old_connect_via = $DBI::connect_via;
351       $DBI::connect_via = 'connect';
352   }
353
354   if(ref $info[0] eq 'CODE') {
355       $dbh = &{$info[0]};
356   }
357   else {
358       $dbh = DBI->connect(@info);
359   }
360
361   $DBI::connect_via = $old_connect_via if $old_connect_via;
362
363   $self->throw_exception("DBI Connection failed: $DBI::errstr")
364       unless $dbh;
365
366   $dbh;
367 }
368
369 =head2 txn_begin
370
371 Calls begin_work on the current dbh.
372
373 See L<DBIx::Class::Schema> for the txn_do() method, which allows for
374 an entire code block to be executed transactionally.
375
376 =cut
377
378 sub txn_begin {
379   my $self = shift;
380   $self->dbh->begin_work
381     if $self->{transaction_depth}++ == 0 and $self->dbh->{AutoCommit};
382 }
383
384 =head2 txn_commit
385
386 Issues a commit against the current dbh.
387
388 =cut
389
390 sub txn_commit {
391   my $self = shift;
392   if ($self->{transaction_depth} == 0) {
393     $self->dbh->commit unless $self->dbh->{AutoCommit};
394   }
395   else {
396     $self->dbh->commit if --$self->{transaction_depth} == 0;    
397   }
398 }
399
400 =head2 txn_rollback
401
402 Issues a rollback against the current dbh. A nested rollback will
403 throw a L<DBIx::Class::Storage::NESTED_ROLLBACK_EXCEPTION> exception,
404 which allows the rollback to propagate to the outermost transaction.
405
406 =cut
407
408 sub txn_rollback {
409   my $self = shift;
410
411   eval {
412     if ($self->{transaction_depth} == 0) {
413       $self->dbh->rollback unless $self->dbh->{AutoCommit};
414     }
415     else {
416       --$self->{transaction_depth} == 0 ?
417         $self->dbh->rollback :
418         die DBIx::Class::Storage::NESTED_ROLLBACK_EXCEPTION->new;
419     }
420   };
421
422   if ($@) {
423     my $error = $@;
424     my $exception_class = "DBIx::Class::Storage::NESTED_ROLLBACK_EXCEPTION";
425     $error =~ /$exception_class/ and $self->throw_exception($error);
426     $self->{transaction_depth} = 0;          # ensure that a failed rollback
427     $self->throw_exception($error);          # resets the transaction depth
428   }
429 }
430
431 sub _execute {
432   my ($self, $op, $extra_bind, $ident, @args) = @_;
433   my ($sql, @bind) = $self->sql_maker->$op($ident, @args);
434   unshift(@bind, @$extra_bind) if $extra_bind;
435   if ($self->debug) {
436       my @debug_bind = map { defined $_ ? qq{`$_'} : q{`NULL'} } @bind;
437       $self->debugfh->print("$sql: " . join(', ', @debug_bind) . "\n");
438   }
439   my $sth = $self->sth($sql,$op);
440   $self->throw_exception("no sth generated via sql: $sql") unless $sth;
441   @bind = map { ref $_ ? ''.$_ : $_ } @bind; # stringify args
442   my $rv;
443   if ($sth) {  
444     $rv = $sth->execute(@bind) or $self->throw_exception("Error executing '$sql': " . $sth->errstr);
445   } else { 
446     $self->throw_exception("'$sql' did not generate a statement.");
447   }
448   return (wantarray ? ($rv, $sth, @bind) : $rv);
449 }
450
451 sub insert {
452   my ($self, $ident, $to_insert) = @_;
453   $self->throw_exception( "Couldn't insert ".join(', ', map "$_ => $to_insert->{$_}", keys %$to_insert)." into ${ident}" )
454     unless ($self->_execute('insert' => [], $ident, $to_insert));
455   return $to_insert;
456 }
457
458 sub update {
459   return shift->_execute('update' => [], @_);
460 }
461
462 sub delete {
463   return shift->_execute('delete' => [], @_);
464 }
465
466 sub _select {
467   my ($self, $ident, $select, $condition, $attrs) = @_;
468   my $order = $attrs->{order_by};
469   if (ref $condition eq 'SCALAR') {
470     $order = $1 if $$condition =~ s/ORDER BY (.*)$//i;
471   }
472   if (exists $attrs->{group_by} || $attrs->{having}) {
473     $order = { group_by => $attrs->{group_by},
474                having => $attrs->{having},
475                ($order ? (order_by => $order) : ()) };
476   }
477   my @args = ('select', $attrs->{bind}, $ident, $select, $condition, $order);
478   if ($attrs->{software_limit} ||
479       $self->sql_maker->_default_limit_syntax eq "GenericSubQ") {
480         $attrs->{software_limit} = 1;
481   } else {
482     push @args, $attrs->{rows}, $attrs->{offset};
483   }
484   return $self->_execute(@args);
485 }
486
487 sub select {
488   my $self = shift;
489   my ($ident, $select, $condition, $attrs) = @_;
490   return $self->cursor->new($self, \@_, $attrs);
491 }
492
493 # Need to call finish() to work round broken DBDs
494
495 sub select_single {
496   my $self = shift;
497   my ($rv, $sth, @bind) = $self->_select(@_);
498   my @row = $sth->fetchrow_array;
499   $sth->finish();
500   return @row;
501 }
502
503 sub sth {
504   my ($self, $sql) = @_;
505   # 3 is the if_active parameter which avoids active sth re-use
506   return $self->dbh->prepare_cached($sql, {}, 3);
507 }
508
509 =head2 columns_info_for
510
511 Returns database type info for a given table columns.
512
513 =cut
514
515 sub columns_info_for {
516     my ($self, $table) = @_;
517
518     if ($self->dbh->can('column_info')) {
519         my %result;
520         my $old_raise_err = $self->dbh->{RaiseError};
521         my $old_print_err = $self->dbh->{PrintError};
522         $self->dbh->{RaiseError} = 1;
523         $self->dbh->{PrintError} = 0;
524         eval {
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} ? 1 : 0;
532                 $column_info{default_value} = $info->{COLUMN_DEF};
533                 $result{$info->{COLUMN_NAME}} = \%column_info;
534             }
535         };
536         $self->dbh->{RaiseError} = $old_raise_err;
537         $self->dbh->{PrintError} = $old_print_err;
538         return \%result if !$@;
539     }
540
541     my %result;
542     my $sth = $self->dbh->prepare("SELECT * FROM $table WHERE 1=0");
543     $sth->execute;
544     my @columns = @{$sth->{NAME_lc}};
545     for my $i ( 0 .. $#columns ){
546         my %column_info;
547         my $type_num = $sth->{TYPE}->[$i];
548         my $type_name;
549         if(defined $type_num && $self->dbh->can('type_info')) {
550             my $type_info = $self->dbh->type_info($type_num);
551             $type_name = $type_info->{TYPE_NAME} if $type_info;
552         }
553         $column_info{data_type} = $type_name ? $type_name : $type_num;
554         $column_info{size} = $sth->{PRECISION}->[$i];
555         $column_info{is_nullable} = $sth->{NULLABLE}->[$i] ? 1 : 0;
556         $result{$columns[$i]} = \%column_info;
557     }
558
559     return \%result;
560 }
561
562 sub last_insert_id {
563   my ($self, $row) = @_;
564     
565   return $self->dbh->func('last_insert_rowid');
566
567 }
568
569 sub sqlt_type { shift->dbh->{Driver}->{Name} }
570
571 sub deployment_statements {
572   my ($self, $schema, $type, $sqltargs) = @_;
573   $type ||= $self->sqlt_type;
574   eval "use SQL::Translator";
575   $self->throw_exception("Can't deploy without SQL::Translator: $@") if $@;
576   eval "use SQL::Translator::Parser::DBIx::Class;";
577   $self->throw_exception($@) if $@; 
578   eval "use SQL::Translator::Producer::${type};";
579   $self->throw_exception($@) if $@;
580   my $tr = SQL::Translator->new(%$sqltargs);
581   SQL::Translator::Parser::DBIx::Class::parse( $tr, $schema );
582   return "SQL::Translator::Producer::${type}"->can('produce')->($tr);
583 }
584
585 sub deploy {
586   my ($self, $schema, $type, $sqltargs) = @_;
587   foreach(split(";\n", $self->deployment_statements($schema, $type, $sqltargs))) {
588       $self->debugfh->print("$_\n") if $self->debug;
589           $self->dbh->do($_) or warn "SQL was:\n $_";
590   } 
591 }
592
593 sub DESTROY { shift->disconnect }
594
595 1;
596
597 =head1 ENVIRONMENT VARIABLES
598
599 =head2 DBIX_CLASS_STORAGE_DBI_DEBUG
600
601 If C<DBIX_CLASS_STORAGE_DBI_DEBUG> is set then SQL trace information
602 is produced (as when the L<debug> method is set).
603
604 If the value is of the form C<1=/path/name> then the trace output is
605 written to the file C</path/name>.
606
607 =head1 AUTHORS
608
609 Matt S. Trout <mst@shadowcatsystems.co.uk>
610
611 Andy Grundman <andy@hybridized.org>
612
613 =head1 LICENSE
614
615 You may distribute this code under the same terms as Perl itself.
616
617 =cut
618