lib/DBIx/Class/Storage/DBI.pm

   1 package DBIx::Class::Storage::DBI;
   2 # -*- mode: cperl; cperl-indent-level: 2 -*-
   3
   4 use base 'DBIx::Class::Storage';
   5
   6 use strict;
   7 use warnings;
   8 use DBI;
   9 use SQL::Abstract::Limit;
  10 use DBIx::Class::Storage::DBI::Cursor;
  11 use DBIx::Class::Storage::Statistics;
  12 use IO::File;
  13 use Scalar::Util 'blessed';
  14
  15 __PACKAGE__->mk_group_accessors(
  16   'simple' =>
  17     qw/_connect_info _dbh _sql_maker _sql_maker_opts _conn_pid _conn_tid
  18        disable_sth_caching cursor on_connect_do transaction_depth/
  19 );
  20
  21 BEGIN {
  22
  23 package DBIC::SQL::Abstract; # Would merge upstream, but nate doesn't reply :(
  24
  25 use base qw/SQL::Abstract::Limit/;
  26
  27 # This prevents the caching of $dbh in S::A::L, I believe
  28 sub new {
  29   my $self = shift->SUPER::new(@_);
  30
  31   # If limit_dialect is a ref (like a $dbh), go ahead and replace
  32   #   it with what it resolves to:
  33   $self->{limit_dialect} = $self->_find_syntax($self->{limit_dialect})
  34     if ref $self->{limit_dialect};
  35
  36   $self;
  37 }
  38
  39 sub _RowNumberOver {
  40   my ($self, $sql, $order, $rows, $offset ) = @_;
  41
  42   $offset += 1;
  43   my $last = $rows + $offset;
  44   my ( $order_by ) = $self->_order_by( $order );
  45
  46   $sql = <<"";
  47 SELECT * FROM
  48 (
  49    SELECT Q1.*, ROW_NUMBER() OVER( ) AS ROW_NUM FROM (
  50       $sql
  51       $order_by
  52    ) Q1
  53 ) Q2
  54 WHERE ROW_NUM BETWEEN $offset AND $last
  55
  56   return $sql;
  57 }
  58
  59
  60 # While we're at it, this should make LIMIT queries more efficient,
  61 #  without digging into things too deeply
  62 use Scalar::Util 'blessed';
  63 sub _find_syntax {
  64   my ($self, $syntax) = @_;
  65   my $dbhname = blessed($syntax) ?  $syntax->{Driver}{Name} : $syntax;
  66   if(ref($self) && $dbhname && $dbhname eq 'DB2') {
  67     return 'RowNumberOver';
  68   }
  69
  70   $self->{_cached_syntax} ||= $self->SUPER::_find_syntax($syntax);
  71 }
  72
  73 sub select {
  74   my ($self, $table, $fields, $where, $order, @rest) = @_;
  75   $table = $self->_quote($table) unless ref($table);
  76   local $self->{rownum_hack_count} = 1
  77     if (defined $rest[0] && $self->{limit_dialect} eq 'RowNum');
  78   @rest = (-1) unless defined $rest[0];
  79   die "LIMIT 0 Does Not Compute" if $rest[0] == 0;
  80     # and anyway, SQL::Abstract::Limit will cause a barf if we don't first
  81   local $self->{having_bind} = [];
  82   my ($sql, @ret) = $self->SUPER::select(
  83     $table, $self->_recurse_fields($fields), $where, $order, @rest
  84   );
  85   return wantarray ? ($sql, @ret, @{$self->{having_bind}}) : $sql;
  86 }
  87
  88 sub insert {
  89   my $self = shift;
  90   my $table = shift;
  91   $table = $self->_quote($table) unless ref($table);
  92   $self->SUPER::insert($table, @_);
  93 }
  94
  95 sub update {
  96   my $self = shift;
  97   my $table = shift;
  98   $table = $self->_quote($table) unless ref($table);
  99   $self->SUPER::update($table, @_);
 100 }
 101
 102 sub delete {
 103   my $self = shift;
 104   my $table = shift;
 105   $table = $self->_quote($table) unless ref($table);
 106   $self->SUPER::delete($table, @_);
 107 }
 108
 109 sub _emulate_limit {
 110   my $self = shift;
 111   if ($_[3] == -1) {
 112     return $_[1].$self->_order_by($_[2]);
 113   } else {
 114     return $self->SUPER::_emulate_limit(@_);
 115   }
 116 }
 117
 118 sub _recurse_fields {
 119   my ($self, $fields) = @_;
 120   my $ref = ref $fields;
 121   return $self->_quote($fields) unless $ref;
 122   return $$fields if $ref eq 'SCALAR';
 123
 124   if ($ref eq 'ARRAY') {
 125     return join(', ', map {
 126       $self->_recurse_fields($_)
 127       .(exists $self->{rownum_hack_count}
 128          ? ' AS col'.$self->{rownum_hack_count}++
 129          : '')
 130      } @$fields);
 131   } elsif ($ref eq 'HASH') {
 132     foreach my $func (keys %$fields) {
 133       return $self->_sqlcase($func)
 134         .'( '.$self->_recurse_fields($fields->{$func}).' )';
 135     }
 136   }
 137 }
 138
 139 sub _order_by {
 140   my $self = shift;
 141   my $ret = '';
 142   my @extra;
 143   if (ref $_[0] eq 'HASH') {
 144     if (defined $_[0]->{group_by}) {
 145       $ret = $self->_sqlcase(' group by ')
 146                .$self->_recurse_fields($_[0]->{group_by});
 147     }
 148     if (defined $_[0]->{having}) {
 149       my $frag;
 150       ($frag, @extra) = $self->_recurse_where($_[0]->{having});
 151       push(@{$self->{having_bind}}, @extra);
 152       $ret .= $self->_sqlcase(' having ').$frag;
 153     }
 154     if (defined $_[0]->{order_by}) {
 155       $ret .= $self->_order_by($_[0]->{order_by});
 156     }
 157   } elsif (ref $_[0] eq 'SCALAR') {
 158     $ret = $self->_sqlcase(' order by ').${ $_[0] };
 159   } elsif (ref $_[0] eq 'ARRAY' && @{$_[0]}) {
 160     my @order = @{+shift};
 161     $ret = $self->_sqlcase(' order by ')
 162           .join(', ', map {
 163                         my $r = $self->_order_by($_, @_);
 164                         $r =~ s/^ ?ORDER BY //i;
 165                         $r;
 166                       } @order);
 167   } else {
 168     $ret = $self->SUPER::_order_by(@_);
 169   }
 170   return $ret;
 171 }
 172
 173 sub _order_directions {
 174   my ($self, $order) = @_;
 175   $order = $order->{order_by} if ref $order eq 'HASH';
 176   return $self->SUPER::_order_directions($order);
 177 }
 178
 179 sub _table {
 180   my ($self, $from) = @_;
 181   if (ref $from eq 'ARRAY') {
 182     return $self->_recurse_from(@$from);
 183   } elsif (ref $from eq 'HASH') {
 184     return $self->_make_as($from);
 185   } else {
 186     return $from; # would love to quote here but _table ends up getting called
 187                   # twice during an ->select without a limit clause due to
 188                   # the way S::A::Limit->select works. should maybe consider
 189                   # bypassing this and doing S::A::select($self, ...) in
 190                   # our select method above. meantime, quoting shims have
 191                   # been added to select/insert/update/delete here
 192   }
 193 }
 194
 195 sub _recurse_from {
 196   my ($self, $from, @join) = @_;
 197   my @sqlf;
 198   push(@sqlf, $self->_make_as($from));
 199   foreach my $j (@join) {
 200     my ($to, $on) = @$j;
 201
 202     # check whether a join type exists
 203     my $join_clause = '';
 204     my $to_jt = ref($to) eq 'ARRAY' ? $to->[0] : $to;
 205     if (ref($to_jt) eq 'HASH' and exists($to_jt->{-join_type})) {
 206       $join_clause = ' '.uc($to_jt->{-join_type}).' JOIN ';
 207     } else {
 208       $join_clause = ' JOIN ';
 209     }
 210     push(@sqlf, $join_clause);
 211
 212     if (ref $to eq 'ARRAY') {
 213       push(@sqlf, '(', $self->_recurse_from(@$to), ')');
 214     } else {
 215       push(@sqlf, $self->_make_as($to));
 216     }
 217     push(@sqlf, ' ON ', $self->_join_condition($on));
 218   }
 219   return join('', @sqlf);
 220 }
 221
 222 sub _make_as {
 223   my ($self, $from) = @_;
 224   return join(' ', map { (ref $_ eq 'SCALAR' ? $$_ : $self->_quote($_)) }
 225                      reverse each %{$self->_skip_options($from)});
 226 }
 227
 228 sub _skip_options {
 229   my ($self, $hash) = @_;
 230   my $clean_hash = {};
 231   $clean_hash->{$_} = $hash->{$_}
 232     for grep {!/^-/} keys %$hash;
 233   return $clean_hash;
 234 }
 235
 236 sub _join_condition {
 237   my ($self, $cond) = @_;
 238   if (ref $cond eq 'HASH') {
 239     my %j;
 240     for (keys %$cond) {
 241       my $v = $cond->{$_};
 242       if (ref $v) {
 243         # XXX no throw_exception() in this package and croak() fails with strange results
 244         Carp::croak(ref($v) . qq{ reference arguments are not supported in JOINS - try using \"..." instead'})
 245             if ref($v) ne 'SCALAR';
 246         $j{$_} = $v;
 247       }
 248       else {
 249         my $x = '= '.$self->_quote($v); $j{$_} = \$x;
 250       }
 251     };
 252     return scalar($self->_recurse_where(\%j));
 253   } elsif (ref $cond eq 'ARRAY') {
 254     return join(' OR ', map { $self->_join_condition($_) } @$cond);
 255   } else {
 256     die "Can't handle this yet!";
 257   }
 258 }
 259
 260 sub _quote {
 261   my ($self, $label) = @_;
 262   return '' unless defined $label;
 263   return "*" if $label eq '*';
 264   return $label unless $self->{quote_char};
 265   if(ref $self->{quote_char} eq "ARRAY"){
 266     return $self->{quote_char}->[0] . $label . $self->{quote_char}->[1]
 267       if !defined $self->{name_sep};
 268     my $sep = $self->{name_sep};
 269     return join($self->{name_sep},
 270         map { $self->{quote_char}->[0] . $_ . $self->{quote_char}->[1]  }
 271        split(/\Q$sep\E/,$label));
 272   }
 273   return $self->SUPER::_quote($label);
 274 }
 275
 276 sub limit_dialect {
 277     my $self = shift;
 278     $self->{limit_dialect} = shift if @_;
 279     return $self->{limit_dialect};
 280 }
 281
 282 sub quote_char {
 283     my $self = shift;
 284     $self->{quote_char} = shift if @_;
 285     return $self->{quote_char};
 286 }
 287
 288 sub name_sep {
 289     my $self = shift;
 290     $self->{name_sep} = shift if @_;
 291     return $self->{name_sep};
 292 }
 293
 294 } # End of BEGIN block
 295
 296 =head1 NAME
 297
 298 DBIx::Class::Storage::DBI - DBI storage handler
 299
 300 =head1 SYNOPSIS
 301
 302 =head1 DESCRIPTION
 303
 304 This class represents the connection to an RDBMS via L<DBI>.  See
 305 L<DBIx::Class::Storage> for general information.  This pod only
 306 documents DBI-specific methods and behaviors.
 307
 308 =head1 METHODS
 309
 310 =cut
 311
 312 sub new {
 313   my $new = shift->next::method(@_);
 314
 315   $new->cursor("DBIx::Class::Storage::DBI::Cursor");
 316   $new->transaction_depth(0);
 317   $new->_sql_maker_opts({});
 318   $new->{_in_dbh_do} = 0;
 319   $new->{_dbh_gen} = 0;
 320
 321   $new;
 322 }
 323
 324 =head2 connect_info
 325
 326 The arguments of C<connect_info> are always a single array reference.
 327
 328 This is normally accessed via L<DBIx::Class::Schema/connection>, which
 329 encapsulates its argument list in an arrayref before calling
 330 C<connect_info> here.
 331
 332 The arrayref can either contain the same set of arguments one would
 333 normally pass to L<DBI/connect>, or a lone code reference which returns
 334 a connected database handle.
 335
 336 In either case, if the final argument in your connect_info happens
 337 to be a hashref, C<connect_info> will look there for several
 338 connection-specific options:
 339
 340 =over 4
 341
 342 =item on_connect_do
 343
 344 This can be set to an arrayref of literal sql statements, which will
 345 be executed immediately after making the connection to the database
 346 every time we [re-]connect.
 347
 348 =item disable_sth_caching
 349
 350 If set to a true value, this option will disable the caching of
 351 statement handles via L<DBI/prepare_cached>.
 352
 353 =item limit_dialect
 354
 355 Sets the limit dialect. This is useful for JDBC-bridge among others
 356 where the remote SQL-dialect cannot be determined by the name of the
 357 driver alone.
 358
 359 =item quote_char
 360
 361 Specifies what characters to use to quote table and column names. If
 362 you use this you will want to specify L<name_sep> as well.
 363
 364 quote_char expects either a single character, in which case is it is placed
 365 on either side of the table/column, or an arrayref of length 2 in which case the
 366 table/column name is placed between the elements.
 367
 368 For example under MySQL you'd use C<quote_char =E<gt> '`'>, and user SQL Server you'd
 369 use C<quote_char =E<gt> [qw/[ ]/]>.
 370
 371 =item name_sep
 372
 373 This only needs to be used in conjunction with L<quote_char>, and is used to
 374 specify the charecter that seperates elements (schemas, tables, columns) from
 375 each other. In most cases this is simply a C<.>.
 376
 377 =back
 378
 379 These options can be mixed in with your other L<DBI> connection attributes,
 380 or placed in a seperate hashref after all other normal L<DBI> connection
 381 arguments.
 382
 383 Every time C<connect_info> is invoked, any previous settings for
 384 these options will be cleared before setting the new ones, regardless of
 385 whether any options are specified in the new C<connect_info>.
 386
 387 Important note:  DBIC expects the returned database handle provided by
 388 a subref argument to have RaiseError set on it.  If it doesn't, things
 389 might not work very well, YMMV.  If you don't use a subref, DBIC will
 390 force this setting for you anyways.  Setting HandleError to anything
 391 other than simple exception object wrapper might cause problems too.
 392
 393 Examples:
 394
 395   # Simple SQLite connection
 396   ->connect_info([ 'dbi:SQLite:./foo.db' ]);
 397
 398   # Connect via subref
 399   ->connect_info([ sub { DBI->connect(...) } ]);
 400
 401   # A bit more complicated
 402   ->connect_info(
 403     [
 404       'dbi:Pg:dbname=foo',
 405       'postgres',
 406       'my_pg_password',
 407       { AutoCommit => 0 },
 408       { quote_char => q{"}, name_sep => q{.} },
 409     ]
 410   );
 411
 412   # Equivalent to the previous example
 413   ->connect_info(
 414     [
 415       'dbi:Pg:dbname=foo',
 416       'postgres',
 417       'my_pg_password',
 418       { AutoCommit => 0, quote_char => q{"}, name_sep => q{.} },
 419     ]
 420   );
 421
 422   # Subref + DBIC-specific connection options
 423   ->connect_info(
 424     [
 425       sub { DBI->connect(...) },
 426       {
 427           quote_char => q{`},
 428           name_sep => q{@},
 429           on_connect_do => ['SET search_path TO myschema,otherschema,public'],
 430           disable_sth_caching => 1,
 431       },
 432     ]
 433   );
 434
 435 =cut
 436
 437 sub connect_info {
 438   my ($self, $info_arg) = @_;
 439
 440   return $self->_connect_info if !$info_arg;
 441
 442   # Kill sql_maker/_sql_maker_opts, so we get a fresh one with only
 443   #  the new set of options
 444   $self->_sql_maker(undef);
 445   $self->_sql_maker_opts({});
 446
 447   my $info = [ @$info_arg ]; # copy because we can alter it
 448   my $last_info = $info->[-1];
 449   if(ref $last_info eq 'HASH') {
 450     for my $storage_opt (qw/on_connect_do disable_sth_caching/) {
 451       if(my $value = delete $last_info->{$storage_opt}) {
 452         $self->$storage_opt($value);
 453       }
 454     }
 455     for my $sql_maker_opt (qw/limit_dialect quote_char name_sep/) {
 456       if(my $opt_val = delete $last_info->{$sql_maker_opt}) {
 457         $self->_sql_maker_opts->{$sql_maker_opt} = $opt_val;
 458       }
 459     }
 460
 461     # Get rid of any trailing empty hashref
 462     pop(@$info) if !keys %$last_info;
 463   }
 464
 465   $self->_connect_info($info);
 466 }
 467
 468 =head2 on_connect_do
 469
 470 This method is deprecated in favor of setting via L</connect_info>.
 471
 472 =head2 dbh_do
 473
 474 Arguments: $subref, @extra_coderef_args?
 475
 476 Execute the given subref using the new exception-based connection management.
 477
 478 The first two arguments will be the storage object that C<dbh_do> was called
 479 on and a database handle to use.  Any additional arguments will be passed
 480 verbatim to the called subref as arguments 2 and onwards.
 481
 482 Using this (instead of $self->_dbh or $self->dbh) ensures correct
 483 exception handling and reconnection (or failover in future subclasses).
 484
 485 Your subref should have no side-effects outside of the database, as
 486 there is the potential for your subref to be partially double-executed
 487 if the database connection was stale/dysfunctional.
 488
 489 Example:
 490
 491   my @stuff = $schema->storage->dbh_do(
 492     sub {
 493       my ($storage, $dbh, @cols) = @_;
 494       my $cols = join(q{, }, @cols);
 495       $dbh->selectrow_array("SELECT $cols FROM foo");
 496     },
 497     @column_list
 498   );
 499
 500 =cut
 501
 502 sub dbh_do {
 503   my $self = shift;
 504   my $coderef = shift;
 505
 506   ref $coderef eq 'CODE' or $self->throw_exception
 507     ('$coderef must be a CODE reference');
 508
 509   return $coderef->($self, $self->_dbh, @_) if $self->{_in_dbh_do}
 510       || $self->{transaction_depth};
 511
 512   local $self->{_in_dbh_do} = 1;
 513
 514   my @result;
 515   my $want_array = wantarray;
 516
 517   eval {
 518     $self->_verify_pid if $self->_dbh;
 519     $self->_populate_dbh if !$self->_dbh;
 520     if($want_array) {
 521         @result = $coderef->($self, $self->_dbh, @_);
 522     }
 523     elsif(defined $want_array) {
 524         $result[0] = $coderef->($self, $self->_dbh, @_);
 525     }
 526     else {
 527         $coderef->($self, $self->_dbh, @_);
 528     }
 529   };
 530
 531   my $exception = $@;
 532   if(!$exception) { return $want_array ? @result : $result[0] }
 533
 534   $self->throw_exception($exception) if $self->connected;
 535
 536   # We were not connected - reconnect and retry, but let any
 537   #  exception fall right through this time
 538   $self->_populate_dbh;
 539   $coderef->($self, $self->_dbh, @_);
 540 }
 541
 542 # This is basically a blend of dbh_do above and DBIx::Class::Storage::txn_do.
 543 # It also informs dbh_do to bypass itself while under the direction of txn_do,
 544 #  via $self->{_in_dbh_do} (this saves some redundant eval and errorcheck, etc)
 545 sub txn_do {
 546   my $self = shift;
 547   my $coderef = shift;
 548
 549   ref $coderef eq 'CODE' or $self->throw_exception
 550     ('$coderef must be a CODE reference');
 551
 552   local $self->{_in_dbh_do} = 1;
 553
 554   my @result;
 555   my $want_array = wantarray;
 556
 557   my $tried = 0;
 558   while(1) {
 559     eval {
 560       $self->_verify_pid if $self->_dbh;
 561       $self->_populate_dbh if !$self->_dbh;
 562
 563       $self->txn_begin;
 564       if($want_array) {
 565           @result = $coderef->(@_);
 566       }
 567       elsif(defined $want_array) {
 568           $result[0] = $coderef->(@_);
 569       }
 570       else {
 571           $coderef->(@_);
 572       }
 573       $self->txn_commit;
 574     };
 575
 576     my $exception = $@;
 577     if(!$exception) { return $want_array ? @result : $result[0] }
 578
 579     if($tried++ > 0 || $self->connected) {
 580       eval { $self->txn_rollback };
 581       my $rollback_exception = $@;
 582       if($rollback_exception) {
 583         my $exception_class = "DBIx::Class::Storage::NESTED_ROLLBACK_EXCEPTION";
 584         $self->throw_exception($exception)  # propagate nested rollback
 585           if $rollback_exception =~ /$exception_class/;
 586
 587         $self->throw_exception(
 588           "Transaction aborted: ${exception}. "
 589           . "Rollback failed: ${rollback_exception}"
 590         );
 591       }
 592       $self->throw_exception($exception)
 593     }
 594
 595     # We were not connected, and was first try - reconnect and retry
 596     # via the while loop
 597     $self->_populate_dbh;
 598   }
 599 }
 600
 601 =head2 disconnect
 602
 603 Our C<disconnect> method also performs a rollback first if the
 604 database is not in C<AutoCommit> mode.
 605
 606 =cut
 607
 608 sub disconnect {
 609   my ($self) = @_;
 610
 611   if( $self->connected ) {
 612     $self->_dbh->rollback unless $self->_dbh->{AutoCommit};
 613     $self->_dbh->disconnect;
 614     $self->_dbh(undef);
 615     $self->{_dbh_gen}++;
 616   }
 617 }
 618
 619 sub connected {
 620   my ($self) = @_;
 621
 622   if(my $dbh = $self->_dbh) {
 623       if(defined $self->_conn_tid && $self->_conn_tid != threads->tid) {
 624           $self->_dbh(undef);
 625           $self->{_dbh_gen}++;
 626           return;
 627       }
 628       else {
 629           $self->_verify_pid;
 630       }
 631       return ($dbh->FETCH('Active') && $dbh->ping);
 632   }
 633
 634   return 0;
 635 }
 636
 637 # handle pid changes correctly
 638 #  NOTE: assumes $self->_dbh is a valid $dbh
 639 sub _verify_pid {
 640   my ($self) = @_;
 641
 642   return if $self->_conn_pid == $$;
 643
 644   $self->_dbh->{InactiveDestroy} = 1;
 645   $self->_dbh(undef);
 646   $self->{_dbh_gen}++;
 647
 648   return;
 649 }
 650
 651 sub ensure_connected {
 652   my ($self) = @_;
 653
 654   unless ($self->connected) {
 655     $self->_populate_dbh;
 656   }
 657 }
 658
 659 =head2 dbh
 660
 661 Returns the dbh - a data base handle of class L<DBI>.
 662
 663 =cut
 664
 665 sub dbh {
 666   my ($self) = @_;
 667
 668   $self->ensure_connected;
 669   return $self->_dbh;
 670 }
 671
 672 sub _sql_maker_args {
 673     my ($self) = @_;
 674
 675     return ( bindtype=>'columns', limit_dialect => $self->dbh, %{$self->_sql_maker_opts} );
 676 }
 677
 678 sub sql_maker {
 679   my ($self) = @_;
 680   unless ($self->_sql_maker) {
 681     $self->_sql_maker(new DBIC::SQL::Abstract( $self->_sql_maker_args ));
 682   }
 683   return $self->_sql_maker;
 684 }
 685
 686 sub _populate_dbh {
 687   my ($self) = @_;
 688   my @info = @{$self->_connect_info || []};
 689   $self->_dbh($self->_connect(@info));
 690
 691   if(ref $self eq 'DBIx::Class::Storage::DBI') {
 692     my $driver = $self->_dbh->{Driver}->{Name};
 693     if ($self->load_optional_class("DBIx::Class::Storage::DBI::${driver}")) {
 694       bless $self, "DBIx::Class::Storage::DBI::${driver}";
 695       $self->_rebless() if $self->can('_rebless');
 696     }
 697   }
 698
 699   # if on-connect sql statements are given execute them
 700   foreach my $sql_statement (@{$self->on_connect_do || []}) {
 701     $self->debugobj->query_start($sql_statement) if $self->debug();
 702     $self->_dbh->do($sql_statement);
 703     $self->debugobj->query_end($sql_statement) if $self->debug();
 704   }
 705
 706   $self->_conn_pid($$);
 707   $self->_conn_tid(threads->tid) if $INC{'threads.pm'};
 708 }
 709
 710 sub _connect {
 711   my ($self, @info) = @_;
 712
 713   $self->throw_exception("You failed to provide any connection info")
 714       if !@info;
 715
 716   my ($old_connect_via, $dbh);
 717
 718   if ($INC{'Apache/DBI.pm'} && $ENV{MOD_PERL}) {
 719       $old_connect_via = $DBI::connect_via;
 720       $DBI::connect_via = 'connect';
 721   }
 722
 723   eval {
 724     if(ref $info[0] eq 'CODE') {
 725        $dbh = &{$info[0]}
 726     }
 727     else {
 728        $dbh = DBI->connect(@info);
 729        $dbh->{RaiseError} = 1;
 730        $dbh->{PrintError} = 0;
 731        $dbh->{PrintWarn} = 0;
 732     }
 733   };
 734
 735   $DBI::connect_via = $old_connect_via if $old_connect_via;
 736
 737   if (!$dbh || $@) {
 738     $self->throw_exception("DBI Connection failed: " . ($@ || $DBI::errstr));
 739   }
 740
 741   $dbh;
 742 }
 743
 744 sub _dbh_txn_begin {
 745   my ($self, $dbh) = @_;
 746   if ($dbh->{AutoCommit}) {
 747     $self->debugobj->txn_begin()
 748       if ($self->debug);
 749     $dbh->begin_work;
 750   }
 751 }
 752
 753 sub txn_begin {
 754   my $self = shift;
 755   $self->dbh_do($self->can('_dbh_txn_begin'))
 756     if $self->{transaction_depth}++ == 0;
 757 }
 758
 759 sub _dbh_txn_commit {
 760   my ($self, $dbh) = @_;
 761   if ($self->{transaction_depth} == 0) {
 762     unless ($dbh->{AutoCommit}) {
 763       $self->debugobj->txn_commit()
 764         if ($self->debug);
 765       $dbh->commit;
 766     }
 767   }
 768   else {
 769     if (--$self->{transaction_depth} == 0) {
 770       $self->debugobj->txn_commit()
 771         if ($self->debug);
 772       $dbh->commit;
 773     }
 774   }
 775 }
 776
 777 sub txn_commit {
 778   my $self = shift;
 779   $self->dbh_do($self->can('_dbh_txn_commit'));
 780 }
 781
 782 sub _dbh_txn_rollback {
 783   my ($self, $dbh) = @_;
 784   if ($self->{transaction_depth} == 0) {
 785     unless ($dbh->{AutoCommit}) {
 786       $self->debugobj->txn_rollback()
 787         if ($self->debug);
 788       $dbh->rollback;
 789     }
 790   }
 791   else {
 792     if (--$self->{transaction_depth} == 0) {
 793       $self->debugobj->txn_rollback()
 794         if ($self->debug);
 795       $dbh->rollback;
 796     }
 797     else {
 798       die DBIx::Class::Storage::NESTED_ROLLBACK_EXCEPTION->new;
 799     }
 800   }
 801 }
 802
 803 sub txn_rollback {
 804   my $self = shift;
 805
 806   eval { $self->dbh_do($self->can('_dbh_txn_rollback')) };
 807   if ($@) {
 808     my $error = $@;
 809     my $exception_class = "DBIx::Class::Storage::NESTED_ROLLBACK_EXCEPTION";
 810     $error =~ /$exception_class/ and $self->throw_exception($error);
 811     $self->{transaction_depth} = 0;          # ensure that a failed rollback
 812     $self->throw_exception($error);          # resets the transaction depth
 813   }
 814 }
 815
 816 # This used to be the top-half of _execute.  It was split out to make it
 817 #  easier to override in NoBindVars without duping the rest.  It takes up
 818 #  all of _execute's args, and emits $sql, @bind.
 819 sub _prep_for_execute {
 820   my ($self, $op, $extra_bind, $ident, @args) = @_;
 821
 822   my ($sql, @bind) = $self->sql_maker->$op($ident, @args);
 823   unshift(@bind,
 824     map { ref $_ eq 'ARRAY' ? $_ : [ '!!dummy', $_ ] } @$extra_bind)
 825       if $extra_bind;
 826   @bind = map { ref $_ ? ''.$_ : $_ } @bind; # stringify args
 827
 828   return ($sql, @bind);
 829 }
 830
 831 sub _execute {
 832   my ($self, $op, $extra_bind, $ident, $bind_attributes, @args) = @_;
 833
 834   if( blessed($ident) && $ident->isa("DBIx::Class::ResultSource") ) {
 835     $ident = $ident->from();
 836   }
 837
 838   my ($sql, @bind) = $self->sql_maker->$op($ident, @args);
 839   unshift(@bind,
 840     map { ref $_ eq 'ARRAY' ? $_ : [ '!!dummy', $_ ] } @$extra_bind)
 841       if $extra_bind;
 842   if ($self->debug) {
 843       my @debug_bind =
 844         map { defined ($_ && $_->[1]) ? qq{'$_->[1]'} : q{'NULL'} } @bind;
 845       $self->debugobj->query_start($sql, @debug_bind);
 846   }
 847
 848   my ($rv, $sth);
 849   RETRY: while (1) {
 850     $sth = eval { $self->sth($sql,$op) };
 851
 852     if (!$sth || $@) {
 853       $self->throw_exception(
 854         'no sth generated via sql (' . ($@ || $self->_dbh->errstr) . "): $sql"
 855       );
 856     }
 857
 858     if ($sth) {
 859       my $time = time();
 860       $rv = eval {
 861         my $placeholder_index = 1;
 862
 863         foreach my $bound (@bind) {
 864
 865           my $attributes = {};
 866           my($column_name, @data) = @$bound;
 867
 868           if( $bind_attributes ) {
 869             $attributes = $bind_attributes->{$column_name}
 870             if defined $bind_attributes->{$column_name};
 871           }
 872
 873           foreach my $data (@data)
 874           {
 875             $data = ref $data ? ''.$data : $data; # stringify args
 876
 877             $sth->bind_param($placeholder_index, $data, $attributes);
 878             $placeholder_index++;
 879           }
 880         }
 881         $sth->execute();
 882       };
 883
 884       if ($@ || !$rv) {
 885         $self->throw_exception("Error executing '$sql': ".($@ || $sth->errstr))
 886           if $self->connected;
 887         $self->_populate_dbh;
 888       } else {
 889         last RETRY;
 890       }
 891     } else {
 892       $self->throw_exception("'$sql' did not generate a statement.");
 893     }
 894   } # While(1) to retry if disconencted
 895
 896   if ($self->debug) {
 897      my @debug_bind =
 898        map { defined ($_ && $_->[1]) ? qq{'$_->[1]'} : q{'NULL'} } @bind;
 899      $self->debugobj->query_end($sql, @debug_bind);
 900   }
 901   return (wantarray ? ($rv, $sth, @bind) : $rv);
 902 }
 903
 904 sub insert {
 905   my ($self, $source, $to_insert) = @_;
 906
 907   my $ident = $source->from;
 908   my $bind_attributes = $self->source_bind_attributes($source);
 909
 910   $self->throw_exception(
 911     "Couldn't insert ".join(', ',
 912       map "$_ => $to_insert->{$_}", keys %$to_insert
 913     )." into ${ident}"
 914   ) unless ($self->_execute('insert' => [], $source, $bind_attributes, $to_insert));
 915   return $to_insert;
 916 }
 917
 918 ## Still not quite perfect, and EXPERIMENTAL
 919 ## Currently it is assumed that all values passed will be "normal", i.e. not
 920 ## scalar refs, or at least, all the same type as the first set, the statement is
 921 ## only prepped once.
 922 sub insert_bulk {
 923   my ($self, $source, $cols, $data) = @_;
 924   my %colvalues;
 925   my $table = $source->from;
 926   @colvalues{@$cols} = (0..$#$cols);
 927   my ($sql, @bind) = $self->sql_maker->insert($table, \%colvalues);
 928
 929   if ($self->debug) {
 930       my @debug_bind = map { defined $_->[1] ? qq{$_->[1]} : q{'NULL'} } @bind;
 931       $self->debugobj->query_start($sql, @debug_bind);
 932   }
 933   my $sth = $self->sth($sql);
 934
 935 #  @bind = map { ref $_ ? ''.$_ : $_ } @bind; # stringify args
 936
 937   my $rv;
 938
 939   ## This must be an arrayref, else nothing works!
 940
 941   my $tuple_status = [];
 942
 943   ##use Data::Dumper;
 944   ##print STDERR Dumper( $data, $sql, [@bind] );
 945
 946   if ($sth) {
 947
 948     my $time = time();
 949
 950     ## Get the bind_attributes, if any exist
 951     my $bind_attributes = $self->source_bind_attributes($source);
 952
 953     ## Bind the values and execute
 954     $rv = eval {
 955
 956      my $placeholder_index = 1;
 957
 958         foreach my $bound (@bind) {
 959
 960           my $attributes = {};
 961           my ($column_name, $data_index) = @$bound;
 962
 963           if( $bind_attributes ) {
 964             $attributes = $bind_attributes->{$column_name}
 965             if defined $bind_attributes->{$column_name};
 966           }
 967
 968           my @data = map { $_->[$data_index] } @$data;
 969
 970           $sth->bind_param_array( $placeholder_index, [@data], $attributes );
 971           $placeholder_index++;
 972       }
 973       $sth->execute_array( {ArrayTupleStatus => $tuple_status} );
 974
 975     };
 976
 977     if ($@ || !defined $rv) {
 978       my $errors = '';
 979       foreach my $tuple (@$tuple_status) {
 980           $errors .= "\n" . $tuple->[1] if(ref $tuple);
 981       }
 982       $self->throw_exception("Error executing '$sql': ".($@ || $errors));
 983     }
 984   } else {
 985     $self->throw_exception("'$sql' did not generate a statement.");
 986   }
 987   if ($self->debug) {
 988       my @debug_bind = map { defined $_ ? qq{`$_'} : q{`NULL'} } @bind;
 989       $self->debugobj->query_end($sql, @debug_bind);
 990   }
 991   return (wantarray ? ($rv, $sth, @bind) : $rv);
 992 }
 993
 994 sub update {
 995   my $self = shift @_;
 996   my $source = shift @_;
 997   my $bind_attributes = $self->source_bind_attributes($source);
 998
 999   return $self->_execute('update' => [], $source, $bind_attributes, @_);
1000 }
1001
1002
1003 sub delete {
1004   my $self = shift @_;
1005   my $source = shift @_;
1006
1007   my $bind_attrs = {}; ## If ever it's needed...
1008
1009   return $self->_execute('delete' => [], $source, $bind_attrs, @_);
1010 }
1011
1012 sub _select {
1013   my ($self, $ident, $select, $condition, $attrs) = @_;
1014   my $order = $attrs->{order_by};
1015   if (ref $condition eq 'SCALAR') {
1016     $order = $1 if $$condition =~ s/ORDER BY (.*)$//i;
1017   }
1018   if (exists $attrs->{group_by} || $attrs->{having}) {
1019     $order = {
1020       group_by => $attrs->{group_by},
1021       having => $attrs->{having},
1022       ($order ? (order_by => $order) : ())
1023     };
1024   }
1025   my $bind_attrs = {}; ## Future support
1026   my @args = ('select', $attrs->{bind}, $ident, $bind_attrs, $select, $condition, $order);
1027   if ($attrs->{software_limit} ||
1028       $self->sql_maker->_default_limit_syntax eq "GenericSubQ") {
1029         $attrs->{software_limit} = 1;
1030   } else {
1031     $self->throw_exception("rows attribute must be positive if present")
1032       if (defined($attrs->{rows}) && !($attrs->{rows} > 0));
1033     push @args, $attrs->{rows}, $attrs->{offset};
1034   }
1035   return $self->_execute(@args);
1036 }
1037
1038 sub source_bind_attributes {
1039   my ($self, $source) = @_;
1040
1041   my $bind_attributes;
1042   foreach my $column ($source->columns) {
1043
1044     my $data_type = $source->column_info($column)->{data_type} || '';
1045     $bind_attributes->{$column} = $self->bind_attribute_by_data_type($data_type)
1046      if $data_type;
1047   }
1048
1049   return $bind_attributes;
1050 }
1051
1052 =head2 select
1053
1054 =over 4
1055
1056 =item Arguments: $ident, $select, $condition, $attrs
1057
1058 =back
1059
1060 Handle a SQL select statement.
1061
1062 =cut
1063
1064 sub select {
1065   my $self = shift;
1066   my ($ident, $select, $condition, $attrs) = @_;
1067   return $self->cursor->new($self, \@_, $attrs);
1068 }
1069
1070 sub select_single {
1071   my $self = shift;
1072   my ($rv, $sth, @bind) = $self->_select(@_);
1073   my @row = $sth->fetchrow_array;
1074   # Need to call finish() to work round broken DBDs
1075   $sth->finish();
1076   return @row;
1077 }
1078
1079 =head2 sth
1080
1081 =over 4
1082
1083 =item Arguments: $sql
1084
1085 =back
1086
1087 Returns a L<DBI> sth (statement handle) for the supplied SQL.
1088
1089 =cut
1090
1091 sub _dbh_sth {
1092   my ($self, $dbh, $sql) = @_;
1093
1094   # 3 is the if_active parameter which avoids active sth re-use
1095   my $sth = $self->disable_sth_caching
1096     ? $dbh->prepare($sql)
1097     : $dbh->prepare_cached($sql, {}, 3);
1098
1099   $self->throw_exception(
1100     'no sth generated via sql (' . ($@ || $dbh->errstr) . "): $sql"
1101   ) if !$sth;
1102
1103   $sth;
1104 }
1105
1106 sub sth {
1107   my ($self, $sql) = @_;
1108   $self->dbh_do($self->can('_dbh_sth'), $sql);
1109 }
1110
1111 sub _dbh_columns_info_for {
1112   my ($self, $dbh, $table) = @_;
1113
1114   if ($dbh->can('column_info')) {
1115     my %result;
1116     eval {
1117       my ($schema,$tab) = $table =~ /^(.+?)\.(.+)$/ ? ($1,$2) : (undef,$table);
1118       my $sth = $dbh->column_info( undef,$schema, $tab, '%' );
1119       $sth->execute();
1120       while ( my $info = $sth->fetchrow_hashref() ){
1121         my %column_info;
1122         $column_info{data_type}   = $info->{TYPE_NAME};
1123         $column_info{size}      = $info->{COLUMN_SIZE};
1124         $column_info{is_nullable}   = $info->{NULLABLE} ? 1 : 0;
1125         $column_info{default_value} = $info->{COLUMN_DEF};
1126         my $col_name = $info->{COLUMN_NAME};
1127         $col_name =~ s/^\"(.*)\"$/$1/;
1128
1129         $result{$col_name} = \%column_info;
1130       }
1131     };
1132     return \%result if !$@ && scalar keys %result;
1133   }
1134
1135   my %result;
1136   my $sth = $dbh->prepare($self->sql_maker->select($table, undef, \'1 = 0'));
1137   $sth->execute;
1138   my @columns = @{$sth->{NAME_lc}};
1139   for my $i ( 0 .. $#columns ){
1140     my %column_info;
1141     $column_info{data_type} = $sth->{TYPE}->[$i];
1142     $column_info{size} = $sth->{PRECISION}->[$i];
1143     $column_info{is_nullable} = $sth->{NULLABLE}->[$i] ? 1 : 0;
1144
1145     if ($column_info{data_type} =~ m/^(.*?)\((.*?)\)$/) {
1146       $column_info{data_type} = $1;
1147       $column_info{size}    = $2;
1148     }
1149
1150     $result{$columns[$i]} = \%column_info;
1151   }
1152   $sth->finish;
1153
1154   foreach my $col (keys %result) {
1155     my $colinfo = $result{$col};
1156     my $type_num = $colinfo->{data_type};
1157     my $type_name;
1158     if(defined $type_num && $dbh->can('type_info')) {
1159       my $type_info = $dbh->type_info($type_num);
1160       $type_name = $type_info->{TYPE_NAME} if $type_info;
1161       $colinfo->{data_type} = $type_name if $type_name;
1162     }
1163   }
1164
1165   return \%result;
1166 }
1167
1168 sub columns_info_for {
1169   my ($self, $table) = @_;
1170   $self->dbh_do($self->can('_dbh_columns_info_for'), $table);
1171 }
1172
1173 =head2 last_insert_id
1174
1175 Return the row id of the last insert.
1176
1177 =cut
1178
1179 sub _dbh_last_insert_id {
1180     my ($self, $dbh, $source, $col) = @_;
1181     # XXX This is a SQLite-ism as a default... is there a DBI-generic way?
1182     $dbh->func('last_insert_rowid');
1183 }
1184
1185 sub last_insert_id {
1186   my $self = shift;
1187   $self->dbh_do($self->can('_dbh_last_insert_id'), @_);
1188 }
1189
1190 =head2 sqlt_type
1191
1192 Returns the database driver name.
1193
1194 =cut
1195
1196 sub sqlt_type { shift->dbh->{Driver}->{Name} }
1197
1198 =head2 bind_attribute_by_data_type
1199
1200 Given a datatype from column info, returns a database specific bind attribute for
1201 $dbh->bind_param($val,$attribute) or nothing if we will let the database planner
1202 just handle it.
1203
1204 Generally only needed for special case column types, like bytea in postgres.
1205
1206 =cut
1207
1208 sub bind_attribute_by_data_type {
1209     return;
1210 }
1211
1212 =head2 create_ddl_dir (EXPERIMENTAL)
1213
1214 =over 4
1215
1216 =item Arguments: $schema \@databases, $version, $directory, $preversion, $sqlt_args
1217
1218 =back
1219
1220 Creates a SQL file based on the Schema, for each of the specified
1221 database types, in the given directory.
1222
1223 Note that this feature is currently EXPERIMENTAL and may not work correctly
1224 across all databases, or fully handle complex relationships.
1225
1226 =cut
1227
1228 sub create_ddl_dir
1229 {
1230   my ($self, $schema, $databases, $version, $dir, $preversion, $sqltargs) = @_;
1231
1232   if(!$dir || !-d $dir)
1233   {
1234     warn "No directory given, using ./\n";
1235     $dir = "./";
1236   }
1237   $databases ||= ['MySQL', 'SQLite', 'PostgreSQL'];
1238   $databases = [ $databases ] if(ref($databases) ne 'ARRAY');
1239   $version ||= $schema->VERSION || '1.x';
1240   $sqltargs = { ( add_drop_table => 1 ), %{$sqltargs || {}} };
1241
1242   eval "use SQL::Translator";
1243   $self->throw_exception("Can't create a ddl file without SQL::Translator: $@") if $@;
1244
1245   my $sqlt = SQL::Translator->new({
1246 #      debug => 1,
1247       add_drop_table => 1,
1248   });
1249   foreach my $db (@$databases)
1250   {
1251     $sqlt->reset();
1252     $sqlt->parser('SQL::Translator::Parser::DBIx::Class');
1253 #    $sqlt->parser_args({'DBIx::Class' => $schema);
1254     $sqlt = $self->configure_sqlt($sqlt, $db);
1255     $sqlt->data($schema);
1256     $sqlt->producer($db);
1257
1258     my $file;
1259     my $filename = $schema->ddl_filename($db, $dir, $version);
1260     if(-e $filename)
1261     {
1262       warn("$filename already exists, skipping $db");
1263       next;
1264     }
1265
1266     my $output = $sqlt->translate;
1267     if(!$output)
1268     {
1269       warn("Failed to translate to $db, skipping. (" . $sqlt->error . ")");
1270       next;
1271     }
1272     if(!open($file, ">$filename"))
1273     {
1274         $self->throw_exception("Can't open $filename for writing ($!)");
1275         next;
1276     }
1277     print $file $output;
1278     close($file);
1279
1280     if($preversion)
1281     {
1282       eval "use SQL::Translator::Diff";
1283       if($@)
1284       {
1285         warn("Can't diff versions without SQL::Translator::Diff: $@");
1286         next;
1287       }
1288
1289       my $prefilename = $schema->ddl_filename($db, $dir, $preversion);
1290 #      print "Previous version $prefilename\n";
1291       if(!-e $prefilename)
1292       {
1293         warn("No previous schema file found ($prefilename)");
1294         next;
1295       }
1296       #### We need to reparse the SQLite file we just wrote, so that
1297       ##   Diff doesnt get all confoosed, and Diff is *very* confused.
1298       ##   FIXME: rip Diff to pieces!
1299 #      my $target_schema = $sqlt->schema;
1300 #      unless ( $target_schema->name ) {
1301 #        $target_schema->name( $filename );
1302 #      }
1303       my @input;
1304       push @input, {file => $prefilename, parser => $db};
1305       push @input, {file => $filename, parser => $db};
1306       my ( $source_schema, $source_db, $target_schema, $target_db ) = map {
1307         my $file   = $_->{'file'};
1308         my $parser = $_->{'parser'};
1309
1310         my $t = SQL::Translator->new;
1311         $t->debug( 0 );
1312         $t->trace( 0 );
1313         $t->parser( $parser )            or die $t->error;
1314         my $out = $t->translate( $file ) or die $t->error;
1315         my $schema = $t->schema;
1316         unless ( $schema->name ) {
1317           $schema->name( $file );
1318         }
1319         ($schema, $parser);
1320       } @input;
1321
1322       my $diff = SQL::Translator::Diff::schema_diff($source_schema, $db,
1323                                                     $target_schema, $db,
1324                                                     {}
1325                                                    );
1326       my $difffile = $schema->ddl_filename($db, $dir, $version, $preversion);
1327       print STDERR "Diff: $difffile: $db, $dir, $version, $preversion \n";
1328       if(-e $difffile)
1329       {
1330         warn("$difffile already exists, skipping");
1331         next;
1332       }
1333       if(!open $file, ">$difffile")
1334       {
1335         $self->throw_exception("Can't write to $difffile ($!)");
1336         next;
1337       }
1338       print $file $diff;
1339       close($file);
1340     }
1341   }
1342 }
1343
1344 sub configure_sqlt() {
1345   my $self = shift;
1346   my $tr = shift;
1347   my $db = shift || $self->sqlt_type;
1348   if ($db eq 'PostgreSQL') {
1349     $tr->quote_table_names(0);
1350     $tr->quote_field_names(0);
1351   }
1352   return $tr;
1353 }
1354
1355 =head2 deployment_statements
1356
1357 =over 4
1358
1359 =item Arguments: $schema, $type, $version, $directory, $sqlt_args
1360
1361 =back
1362
1363 Returns the statements used by L</deploy> and L<DBIx::Class::Schema/deploy>.
1364 The database driver name is given by C<$type>, though the value from
1365 L</sqlt_type> is used if it is not specified.
1366
1367 C<$directory> is used to return statements from files in a previously created
1368 L</create_ddl_dir> directory and is optional. The filenames are constructed
1369 from L<DBIx::Class::Schema/ddl_filename>, the schema name and the C<$version>.
1370
1371 If no C<$directory> is specified then the statements are constructed on the
1372 fly using L<SQL::Translator> and C<$version> is ignored.
1373
1374 See L<SQL::Translator/METHODS> for a list of values for C<$sqlt_args>.
1375
1376 =cut
1377
1378 sub deployment_statements {
1379   my ($self, $schema, $type, $version, $dir, $sqltargs) = @_;
1380   # Need to be connected to get the correct sqlt_type
1381   $self->ensure_connected() unless $type;
1382   $type ||= $self->sqlt_type;
1383   $version ||= $schema->VERSION || '1.x';
1384   $dir ||= './';
1385   my $filename = $schema->ddl_filename($type, $dir, $version);
1386   if(-f $filename)
1387   {
1388       my $file;
1389       open($file, "<$filename")
1390         or $self->throw_exception("Can't open $filename ($!)");
1391       my @rows = <$file>;
1392       close($file);
1393       return join('', @rows);
1394   }
1395
1396   eval "use SQL::Translator";
1397   if(!$@)
1398   {
1399     eval "use SQL::Translator::Parser::DBIx::Class;";
1400     $self->throw_exception($@) if $@;
1401     eval "use SQL::Translator::Producer::${type};";
1402     $self->throw_exception($@) if $@;
1403
1404     # sources needs to be a parser arg, but for simplicty allow at top level
1405     # coming in
1406     $sqltargs->{parser_args}{sources} = delete $sqltargs->{sources}
1407         if exists $sqltargs->{sources};
1408
1409     my $tr = SQL::Translator->new(%$sqltargs);
1410     SQL::Translator::Parser::DBIx::Class::parse( $tr, $schema );
1411     return "SQL::Translator::Producer::${type}"->can('produce')->($tr);
1412   }
1413
1414   $self->throw_exception("No SQL::Translator, and no Schema file found, aborting deploy");
1415   return;
1416
1417 }
1418
1419 sub deploy {
1420   my ($self, $schema, $type, $sqltargs, $dir) = @_;
1421   foreach my $statement ( $self->deployment_statements($schema, $type, undef, $dir, { no_comments => 1, %{ $sqltargs || {} } } ) ) {
1422     for ( split(";\n", $statement)) {
1423       next if($_ =~ /^--/);
1424       next if(!$_);
1425 #      next if($_ =~ /^DROP/m);
1426       next if($_ =~ /^BEGIN TRANSACTION/m);
1427       next if($_ =~ /^COMMIT/m);
1428       next if $_ =~ /^\s+$/; # skip whitespace only
1429       $self->debugobj->query_start($_) if $self->debug;
1430       $self->dbh->do($_) or warn "SQL was:\n $_"; # XXX exceptions?
1431       $self->debugobj->query_end($_) if $self->debug;
1432     }
1433   }
1434 }
1435
1436 =head2 datetime_parser
1437
1438 Returns the datetime parser class
1439
1440 =cut
1441
1442 sub datetime_parser {
1443   my $self = shift;
1444   return $self->{datetime_parser} ||= $self->build_datetime_parser(@_);
1445 }
1446
1447 =head2 datetime_parser_type
1448
1449 Defines (returns) the datetime parser class - currently hardwired to
1450 L<DateTime::Format::MySQL>
1451
1452 =cut
1453
1454 sub datetime_parser_type { "DateTime::Format::MySQL"; }
1455
1456 =head2 build_datetime_parser
1457
1458 See L</datetime_parser>
1459
1460 =cut
1461
1462 sub build_datetime_parser {
1463   my $self = shift;
1464   my $type = $self->datetime_parser_type(@_);
1465   eval "use ${type}";
1466   $self->throw_exception("Couldn't load ${type}: $@") if $@;
1467   return $type;
1468 }
1469
1470 sub DESTROY {
1471   my $self = shift;
1472   return if !$self->_dbh;
1473   $self->_verify_pid;
1474   $self->_dbh(undef);
1475 }
1476
1477 1;
1478
1479 =head1 SQL METHODS
1480
1481 The module defines a set of methods within the DBIC::SQL::Abstract
1482 namespace.  These build on L<SQL::Abstract::Limit> to provide the
1483 SQL query functions.
1484
1485 The following methods are extended:-
1486
1487 =over 4
1488
1489 =item delete
1490
1491 =item insert
1492
1493 =item select
1494
1495 =item update
1496
1497 =item limit_dialect
1498
1499 See L</connect_info> for details.
1500 For setting, this method is deprecated in favor of L</connect_info>.
1501
1502 =item quote_char
1503
1504 See L</connect_info> for details.
1505 For setting, this method is deprecated in favor of L</connect_info>.
1506
1507 =item name_sep
1508
1509 See L</connect_info> for details.
1510 For setting, this method is deprecated in favor of L</connect_info>.
1511
1512 =back
1513
1514 =head1 AUTHORS
1515
1516 Matt S. Trout <mst@shadowcatsystems.co.uk>
1517
1518 Andy Grundman <andy@hybridized.org>
1519
1520 =head1 LICENSE
1521
1522 You may distribute this code under the same terms as Perl itself.
1523
1524 =cut