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 Carp::Clan qw/^DBIx::Class/;
   9 use DBI;
  10 use SQL::Abstract::Limit;
  11 use DBIx::Class::Storage::DBI::Cursor;
  12 use DBIx::Class::Storage::Statistics;
  13 use Scalar::Util qw/blessed weaken/;
  14
  15 __PACKAGE__->mk_group_accessors('simple' =>
  16     qw/_connect_info _dbi_connect_info _dbh _sql_maker _sql_maker_opts
  17        _conn_pid _conn_tid transaction_depth _dbh_autocommit savepoints/
  18 );
  19
  20 # the values for these accessors are picked out (and deleted) from
  21 # the attribute hashref passed to connect_info
  22 my @storage_options = qw/
  23   on_connect_do on_disconnect_do disable_sth_caching unsafe auto_savepoint
  24 /;
  25 __PACKAGE__->mk_group_accessors('simple' => @storage_options);
  26
  27
  28 # default cursor class, overridable in connect_info attributes
  29 __PACKAGE__->cursor_class('DBIx::Class::Storage::DBI::Cursor');
  30
  31 __PACKAGE__->mk_group_accessors('inherited' => qw/sql_maker_class/);
  32 __PACKAGE__->sql_maker_class('DBIC::SQL::Abstract');
  33
  34 BEGIN {
  35
  36 package # Hide from PAUSE
  37   DBIC::SQL::Abstract; # Would merge upstream, but nate doesn't reply :(
  38
  39 use base qw/SQL::Abstract::Limit/;
  40
  41 # This prevents the caching of $dbh in S::A::L, I believe
  42 sub new {
  43   my $self = shift->SUPER::new(@_);
  44
  45   # If limit_dialect is a ref (like a $dbh), go ahead and replace
  46   #   it with what it resolves to:
  47   $self->{limit_dialect} = $self->_find_syntax($self->{limit_dialect})
  48     if ref $self->{limit_dialect};
  49
  50   $self;
  51 }
  52
  53 # DB2 is the only remaining DB using this. Even though we are not sure if
  54 # RowNumberOver is still needed here (should be part of SQLA) leave the
  55 # code in place
  56 sub _RowNumberOver {
  57   my ($self, $sql, $order, $rows, $offset ) = @_;
  58
  59   $offset += 1;
  60   my $last = $rows + $offset;
  61   my ( $order_by ) = $self->_order_by( $order );
  62
  63   $sql = <<"SQL";
  64 SELECT * FROM
  65 (
  66    SELECT Q1.*, ROW_NUMBER() OVER( ) AS ROW_NUM FROM (
  67       $sql
  68       $order_by
  69    ) Q1
  70 ) Q2
  71 WHERE ROW_NUM BETWEEN $offset AND $last
  72
  73 SQL
  74
  75   return $sql;
  76 }
  77
  78
  79 # While we're at it, this should make LIMIT queries more efficient,
  80 #  without digging into things too deeply
  81 use Scalar::Util 'blessed';
  82 sub _find_syntax {
  83   my ($self, $syntax) = @_;
  84
  85   # DB2 is the only remaining DB using this. Even though we are not sure if
  86   # RowNumberOver is still needed here (should be part of SQLA) leave the
  87   # code in place
  88   my $dbhname = blessed($syntax) ? $syntax->{Driver}{Name} : $syntax;
  89   if(ref($self) && $dbhname && $dbhname eq 'DB2') {
  90     return 'RowNumberOver';
  91   }
  92
  93   $self->{_cached_syntax} ||= $self->SUPER::_find_syntax($syntax);
  94 }
  95
  96 sub select {
  97   my ($self, $table, $fields, $where, $order, @rest) = @_;
  98   if (ref $table eq 'SCALAR') {
  99     $table = $$table;
 100   }
 101   elsif (not ref $table) {
 102     $table = $self->_quote($table);
 103   }
 104   local $self->{rownum_hack_count} = 1
 105     if (defined $rest[0] && $self->{limit_dialect} eq 'RowNum');
 106   @rest = (-1) unless defined $rest[0];
 107   die "LIMIT 0 Does Not Compute" if $rest[0] == 0;
 108     # and anyway, SQL::Abstract::Limit will cause a barf if we don't first
 109   local $self->{having_bind} = [];
 110   my ($sql, @ret) = $self->SUPER::select(
 111     $table, $self->_recurse_fields($fields), $where, $order, @rest
 112   );
 113   $sql .=
 114     $self->{for} ?
 115     (
 116       $self->{for} eq 'update' ? ' FOR UPDATE' :
 117       $self->{for} eq 'shared' ? ' FOR SHARE'  :
 118       ''
 119     ) :
 120     ''
 121   ;
 122   return wantarray ? ($sql, @ret, @{$self->{having_bind}}) : $sql;
 123 }
 124
 125 sub insert {
 126   my $self = shift;
 127   my $table = shift;
 128   $table = $self->_quote($table) unless ref($table);
 129   $self->SUPER::insert($table, @_);
 130 }
 131
 132 sub update {
 133   my $self = shift;
 134   my $table = shift;
 135   $table = $self->_quote($table) unless ref($table);
 136   $self->SUPER::update($table, @_);
 137 }
 138
 139 sub delete {
 140   my $self = shift;
 141   my $table = shift;
 142   $table = $self->_quote($table) unless ref($table);
 143   $self->SUPER::delete($table, @_);
 144 }
 145
 146 sub _emulate_limit {
 147   my $self = shift;
 148   if ($_[3] == -1) {
 149     return $_[1].$self->_order_by($_[2]);
 150   } else {
 151     return $self->SUPER::_emulate_limit(@_);
 152   }
 153 }
 154
 155 sub _recurse_fields {
 156   my ($self, $fields, $params) = @_;
 157   my $ref = ref $fields;
 158   return $self->_quote($fields) unless $ref;
 159   return $$fields if $ref eq 'SCALAR';
 160
 161   if ($ref eq 'ARRAY') {
 162     return join(', ', map {
 163       $self->_recurse_fields($_)
 164         .(exists $self->{rownum_hack_count} && !($params && $params->{no_rownum_hack})
 165           ? ' AS col'.$self->{rownum_hack_count}++
 166           : '')
 167       } @$fields);
 168   } elsif ($ref eq 'HASH') {
 169     foreach my $func (keys %$fields) {
 170       return $self->_sqlcase($func)
 171         .'( '.$self->_recurse_fields($fields->{$func}).' )';
 172     }
 173   }
 174 }
 175
 176 sub _order_by {
 177   my $self = shift;
 178   my $ret = '';
 179   my @extra;
 180   if (ref $_[0] eq 'HASH') {
 181     if (defined $_[0]->{group_by}) {
 182       $ret = $self->_sqlcase(' group by ')
 183         .$self->_recurse_fields($_[0]->{group_by}, { no_rownum_hack => 1 });
 184     }
 185     if (defined $_[0]->{having}) {
 186       my $frag;
 187       ($frag, @extra) = $self->_recurse_where($_[0]->{having});
 188       push(@{$self->{having_bind}}, @extra);
 189       $ret .= $self->_sqlcase(' having ').$frag;
 190     }
 191     if (defined $_[0]->{order_by}) {
 192       $ret .= $self->_order_by($_[0]->{order_by});
 193     }
 194     if (grep { $_ =~ /^-(desc|asc)/i } keys %{$_[0]}) {
 195       return $self->SUPER::_order_by($_[0]);
 196     }
 197   } elsif (ref $_[0] eq 'SCALAR') {
 198     $ret = $self->_sqlcase(' order by ').${ $_[0] };
 199   } elsif (ref $_[0] eq 'ARRAY' && @{$_[0]}) {
 200     my @order = @{+shift};
 201     $ret = $self->_sqlcase(' order by ')
 202           .join(', ', map {
 203                         my $r = $self->_order_by($_, @_);
 204                         $r =~ s/^ ?ORDER BY //i;
 205                         $r;
 206                       } @order);
 207   } else {
 208     $ret = $self->SUPER::_order_by(@_);
 209   }
 210   return $ret;
 211 }
 212
 213 sub _order_directions {
 214   my ($self, $order) = @_;
 215   $order = $order->{order_by} if ref $order eq 'HASH';
 216   return $self->SUPER::_order_directions($order);
 217 }
 218
 219 sub _table {
 220   my ($self, $from) = @_;
 221   if (ref $from eq 'ARRAY') {
 222     return $self->_recurse_from(@$from);
 223   } elsif (ref $from eq 'HASH') {
 224     return $self->_make_as($from);
 225   } else {
 226     return $from; # would love to quote here but _table ends up getting called
 227                   # twice during an ->select without a limit clause due to
 228                   # the way S::A::Limit->select works. should maybe consider
 229                   # bypassing this and doing S::A::select($self, ...) in
 230                   # our select method above. meantime, quoting shims have
 231                   # been added to select/insert/update/delete here
 232   }
 233 }
 234
 235 sub _recurse_from {
 236   my ($self, $from, @join) = @_;
 237   my @sqlf;
 238   push(@sqlf, $self->_make_as($from));
 239   foreach my $j (@join) {
 240     my ($to, $on) = @$j;
 241
 242     # check whether a join type exists
 243     my $join_clause = '';
 244     my $to_jt = ref($to) eq 'ARRAY' ? $to->[0] : $to;
 245     if (ref($to_jt) eq 'HASH' and exists($to_jt->{-join_type})) {
 246       $join_clause = ' '.uc($to_jt->{-join_type}).' JOIN ';
 247     } else {
 248       $join_clause = ' JOIN ';
 249     }
 250     push(@sqlf, $join_clause);
 251
 252     if (ref $to eq 'ARRAY') {
 253       push(@sqlf, '(', $self->_recurse_from(@$to), ')');
 254     } else {
 255       push(@sqlf, $self->_make_as($to));
 256     }
 257     push(@sqlf, ' ON ', $self->_join_condition($on));
 258   }
 259   return join('', @sqlf);
 260 }
 261
 262 sub _make_as {
 263   my ($self, $from) = @_;
 264   return join(' ', map { (ref $_ eq 'SCALAR' ? $$_ : $self->_quote($_)) }
 265                      reverse each %{$self->_skip_options($from)});
 266 }
 267
 268 sub _skip_options {
 269   my ($self, $hash) = @_;
 270   my $clean_hash = {};
 271   $clean_hash->{$_} = $hash->{$_}
 272     for grep {!/^-/} keys %$hash;
 273   return $clean_hash;
 274 }
 275
 276 sub _join_condition {
 277   my ($self, $cond) = @_;
 278   if (ref $cond eq 'HASH') {
 279     my %j;
 280     for (keys %$cond) {
 281       my $v = $cond->{$_};
 282       if (ref $v) {
 283         # XXX no throw_exception() in this package and croak() fails with strange results
 284         Carp::croak(ref($v) . qq{ reference arguments are not supported in JOINS - try using \"..." instead'})
 285             if ref($v) ne 'SCALAR';
 286         $j{$_} = $v;
 287       }
 288       else {
 289         my $x = '= '.$self->_quote($v); $j{$_} = \$x;
 290       }
 291     };
 292     return scalar($self->_recurse_where(\%j));
 293   } elsif (ref $cond eq 'ARRAY') {
 294     return join(' OR ', map { $self->_join_condition($_) } @$cond);
 295   } else {
 296     die "Can't handle this yet!";
 297   }
 298 }
 299
 300 sub _quote {
 301   my ($self, $label) = @_;
 302   return '' unless defined $label;
 303   return "*" if $label eq '*';
 304   return $label unless $self->{quote_char};
 305   if(ref $self->{quote_char} eq "ARRAY"){
 306     return $self->{quote_char}->[0] . $label . $self->{quote_char}->[1]
 307       if !defined $self->{name_sep};
 308     my $sep = $self->{name_sep};
 309     return join($self->{name_sep},
 310         map { $self->{quote_char}->[0] . $_ . $self->{quote_char}->[1]  }
 311        split(/\Q$sep\E/,$label));
 312   }
 313   return $self->SUPER::_quote($label);
 314 }
 315
 316 sub limit_dialect {
 317     my $self = shift;
 318     $self->{limit_dialect} = shift if @_;
 319     return $self->{limit_dialect};
 320 }
 321
 322 sub quote_char {
 323     my $self = shift;
 324     $self->{quote_char} = shift if @_;
 325     return $self->{quote_char};
 326 }
 327
 328 sub name_sep {
 329     my $self = shift;
 330     $self->{name_sep} = shift if @_;
 331     return $self->{name_sep};
 332 }
 333
 334 } # End of BEGIN block
 335
 336 =head1 NAME
 337
 338 DBIx::Class::Storage::DBI - DBI storage handler
 339
 340 =head1 SYNOPSIS
 341
 342   my $schema = MySchema->connect('dbi:SQLite:my.db');
 343
 344   $schema->storage->debug(1);
 345   $schema->dbh_do("DROP TABLE authors");
 346
 347   $schema->resultset('Book')->search({
 348      written_on => $schema->storage->datetime_parser(DateTime->now)
 349   });
 350
 351 =head1 DESCRIPTION
 352
 353 This class represents the connection to an RDBMS via L<DBI>.  See
 354 L<DBIx::Class::Storage> for general information.  This pod only
 355 documents DBI-specific methods and behaviors.
 356
 357 =head1 METHODS
 358
 359 =cut
 360
 361 sub new {
 362   my $new = shift->next::method(@_);
 363
 364   $new->transaction_depth(0);
 365   $new->_sql_maker_opts({});
 366   $new->{savepoints} = [];
 367   $new->{_in_dbh_do} = 0;
 368   $new->{_dbh_gen} = 0;
 369
 370   $new;
 371 }
 372
 373 =head2 connect_info
 374
 375 This method is normally called by L<DBIx::Class::Schema/connection>, which
 376 encapsulates its argument list in an arrayref before passing them here.
 377
 378 The argument list may contain:
 379
 380 =over
 381
 382 =item *
 383
 384 The same 4-element argument set one would normally pass to
 385 L<DBI/connect>, optionally followed by
 386 L<extra attributes|/DBIx::Class specific connection attributes>
 387 recognized by DBIx::Class:
 388
 389   $connect_info_args = [ $dsn, $user, $password, \%dbi_attributes?, \%extra_attributes? ];
 390
 391 =item *
 392
 393 A single code reference which returns a connected
 394 L<DBI database handle|DBI/connect> optionally followed by
 395 L<extra attributes|/DBIx::Class specific connection attributes> recognized
 396 by DBIx::Class:
 397
 398   $connect_info_args = [ sub { DBI->connect (...) }, \%extra_attributes? ];
 399
 400 =item *
 401
 402 A single hashref with all the attributes and the dsn/user/password
 403 mixed together:
 404
 405   $connect_info_args = [{
 406     dsn => $dsn,
 407     user => $user,
 408     password => $pass,
 409     %dbi_attributes,
 410     %extra_attributes,
 411   }];
 412
 413 This is particularly useful for L<Catalyst> based applications, allowing the
 414 following config (L<Config::General> style):
 415
 416   <Model::DB>
 417     schema_class   App::DB
 418     <connect_info>
 419       dsn          dbi:mysql:database=test
 420       user         testuser
 421       password     TestPass
 422       AutoCommit   1
 423     </connect_info>
 424   </Model::DB>
 425
 426 =back
 427
 428 Please note that the L<DBI> docs recommend that you always explicitly
 429 set C<AutoCommit> to either I<0> or I<1>.  L<DBIx::Class> further
 430 recommends that it be set to I<1>, and that you perform transactions
 431 via our L<DBIx::Class::Schema/txn_do> method.  L<DBIx::Class> will set it
 432 to I<1> if you do not do explicitly set it to zero.  This is the default
 433 for most DBDs. See L</DBIx::Class and AutoCommit> for details.
 434
 435 =head3 DBIx::Class specific connection attributes
 436
 437 In addition to the standard L<DBI|DBI/ATTRIBUTES_COMMON_TO_ALL_HANDLES>
 438 L<connection|DBI/Database_Handle_Attributes> attributes, DBIx::Class recognizes
 439 the following connection options. These options can be mixed in with your other
 440 L<DBI> connection attributes, or placed in a seperate hashref
 441 (C<\%extra_attributes>) as shown above.
 442
 443 Every time C<connect_info> is invoked, any previous settings for
 444 these options will be cleared before setting the new ones, regardless of
 445 whether any options are specified in the new C<connect_info>.
 446
 447
 448 =over
 449
 450 =item on_connect_do
 451
 452 Specifies things to do immediately after connecting or re-connecting to
 453 the database.  Its value may contain:
 454
 455 =over
 456
 457 =item an array reference
 458
 459 This contains SQL statements to execute in order.  Each element contains
 460 a string or a code reference that returns a string.
 461
 462 =item a code reference
 463
 464 This contains some code to execute.  Unlike code references within an
 465 array reference, its return value is ignored.
 466
 467 =back
 468
 469 =item on_disconnect_do
 470
 471 Takes arguments in the same form as L</on_connect_do> and executes them
 472 immediately before disconnecting from the database.
 473
 474 Note, this only runs if you explicitly call L</disconnect> on the
 475 storage object.
 476
 477 =item disable_sth_caching
 478
 479 If set to a true value, this option will disable the caching of
 480 statement handles via L<DBI/prepare_cached>.
 481
 482 =item limit_dialect
 483
 484 Sets the limit dialect. This is useful for JDBC-bridge among others
 485 where the remote SQL-dialect cannot be determined by the name of the
 486 driver alone. See also L<SQL::Abstract::Limit>.
 487
 488 =item quote_char
 489
 490 Specifies what characters to use to quote table and column names. If
 491 you use this you will want to specify L</name_sep> as well.
 492
 493 C<quote_char> expects either a single character, in which case is it
 494 is placed on either side of the table/column name, or an arrayref of length
 495 2 in which case the table/column name is placed between the elements.
 496
 497 For example under MySQL you should use C<< quote_char => '`' >>, and for
 498 SQL Server you should use C<< quote_char => [qw/[ ]/] >>.
 499
 500 =item name_sep
 501
 502 This only needs to be used in conjunction with C<quote_char>, and is used to
 503 specify the charecter that seperates elements (schemas, tables, columns) from
 504 each other. In most cases this is simply a C<.>.
 505
 506 The consequences of not supplying this value is that L<SQL::Abstract>
 507 will assume DBIx::Class' uses of aliases to be complete column
 508 names. The output will look like I<"me.name"> when it should actually
 509 be I<"me"."name">.
 510
 511 =item unsafe
 512
 513 This Storage driver normally installs its own C<HandleError>, sets
 514 C<RaiseError> and C<ShowErrorStatement> on, and sets C<PrintError> off on
 515 all database handles, including those supplied by a coderef.  It does this
 516 so that it can have consistent and useful error behavior.
 517
 518 If you set this option to a true value, Storage will not do its usual
 519 modifications to the database handle's attributes, and instead relies on
 520 the settings in your connect_info DBI options (or the values you set in
 521 your connection coderef, in the case that you are connecting via coderef).
 522
 523 Note that your custom settings can cause Storage to malfunction,
 524 especially if you set a C<HandleError> handler that suppresses exceptions
 525 and/or disable C<RaiseError>.
 526
 527 =item auto_savepoint
 528
 529 If this option is true, L<DBIx::Class> will use savepoints when nesting
 530 transactions, making it possible to recover from failure in the inner
 531 transaction without having to abort all outer transactions.
 532
 533 =item cursor_class
 534
 535 Use this argument to supply a cursor class other than the default
 536 L<DBIx::Class::Storage::DBI::Cursor>.
 537
 538 =back
 539
 540 Some real-life examples of arguments to L</connect_info> and
 541 L<DBIx::Class::Schema/connect>
 542
 543   # Simple SQLite connection
 544   ->connect_info([ 'dbi:SQLite:./foo.db' ]);
 545
 546   # Connect via subref
 547   ->connect_info([ sub { DBI->connect(...) } ]);
 548
 549   # A bit more complicated
 550   ->connect_info(
 551     [
 552       'dbi:Pg:dbname=foo',
 553       'postgres',
 554       'my_pg_password',
 555       { AutoCommit => 1 },
 556       { quote_char => q{"}, name_sep => q{.} },
 557     ]
 558   );
 559
 560   # Equivalent to the previous example
 561   ->connect_info(
 562     [
 563       'dbi:Pg:dbname=foo',
 564       'postgres',
 565       'my_pg_password',
 566       { AutoCommit => 1, quote_char => q{"}, name_sep => q{.} },
 567     ]
 568   );
 569
 570   # Same, but with hashref as argument
 571   # See parse_connect_info for explanation
 572   ->connect_info(
 573     [{
 574       dsn         => 'dbi:Pg:dbname=foo',
 575       user        => 'postgres',
 576       password    => 'my_pg_password',
 577       AutoCommit  => 1,
 578       quote_char  => q{"},
 579       name_sep    => q{.},
 580     }]
 581   );
 582
 583   # Subref + DBIx::Class-specific connection options
 584   ->connect_info(
 585     [
 586       sub { DBI->connect(...) },
 587       {
 588           quote_char => q{`},
 589           name_sep => q{@},
 590           on_connect_do => ['SET search_path TO myschema,otherschema,public'],
 591           disable_sth_caching => 1,
 592       },
 593     ]
 594   );
 595
 596
 597
 598 =cut
 599
 600 sub connect_info {
 601   my ($self, $info_arg) = @_;
 602
 603   return $self->_connect_info if !$info_arg;
 604
 605   my @args = @$info_arg;  # take a shallow copy for further mutilation
 606   $self->_connect_info([@args]); # copy for _connect_info
 607
 608
 609   # combine/pre-parse arguments depending on invocation style
 610
 611   my %attrs;
 612   if (ref $args[0] eq 'CODE') {     # coderef with optional \%extra_attributes
 613     %attrs = %{ $args[1] || {} };
 614     @args = $args[0];
 615   }
 616   elsif (ref $args[0] eq 'HASH') { # single hashref (i.e. Catalyst config)
 617     %attrs = %{$args[0]};
 618     @args = ();
 619     for (qw/password user dsn/) {
 620       unshift @args, delete $attrs{$_};
 621     }
 622   }
 623   else {                # otherwise assume dsn/user/password + \%attrs + \%extra_attrs
 624     %attrs = (
 625       % { $args[3] || {} },
 626       % { $args[4] || {} },
 627     );
 628     @args = @args[0,1,2];
 629   }
 630
 631   # Kill sql_maker/_sql_maker_opts, so we get a fresh one with only
 632   #  the new set of options
 633   $self->_sql_maker(undef);
 634   $self->_sql_maker_opts({});
 635
 636   if(keys %attrs) {
 637     for my $storage_opt (@storage_options, 'cursor_class') {    # @storage_options is declared at the top of the module
 638       if(my $value = delete $attrs{$storage_opt}) {
 639         $self->$storage_opt($value);
 640       }
 641     }
 642     for my $sql_maker_opt (qw/limit_dialect quote_char name_sep/) {
 643       if(my $opt_val = delete $attrs{$sql_maker_opt}) {
 644         $self->_sql_maker_opts->{$sql_maker_opt} = $opt_val;
 645       }
 646     }
 647   }
 648
 649   %attrs = () if (ref $args[0] eq 'CODE');  # _connect() never looks past $args[0] in this case
 650
 651   $self->_dbi_connect_info([@args, keys %attrs ? \%attrs : ()]);
 652   $self->_connect_info;
 653 }
 654
 655 =head2 on_connect_do
 656
 657 This method is deprecated in favour of setting via L</connect_info>.
 658
 659
 660 =head2 dbh_do
 661
 662 Arguments: ($subref | $method_name), @extra_coderef_args?
 663
 664 Execute the given $subref or $method_name using the new exception-based
 665 connection management.
 666
 667 The first two arguments will be the storage object that C<dbh_do> was called
 668 on and a database handle to use.  Any additional arguments will be passed
 669 verbatim to the called subref as arguments 2 and onwards.
 670
 671 Using this (instead of $self->_dbh or $self->dbh) ensures correct
 672 exception handling and reconnection (or failover in future subclasses).
 673
 674 Your subref should have no side-effects outside of the database, as
 675 there is the potential for your subref to be partially double-executed
 676 if the database connection was stale/dysfunctional.
 677
 678 Example:
 679
 680   my @stuff = $schema->storage->dbh_do(
 681     sub {
 682       my ($storage, $dbh, @cols) = @_;
 683       my $cols = join(q{, }, @cols);
 684       $dbh->selectrow_array("SELECT $cols FROM foo");
 685     },
 686     @column_list
 687   );
 688
 689 =cut
 690
 691 sub dbh_do {
 692   my $self = shift;
 693   my $code = shift;
 694
 695   my $dbh = $self->_dbh;
 696
 697   return $self->$code($dbh, @_) if $self->{_in_dbh_do}
 698       || $self->{transaction_depth};
 699
 700   local $self->{_in_dbh_do} = 1;
 701
 702   my @result;
 703   my $want_array = wantarray;
 704
 705   eval {
 706     $self->_verify_pid if $dbh;
 707     if(!$self->_dbh) {
 708         $self->_populate_dbh;
 709         $dbh = $self->_dbh;
 710     }
 711
 712     if($want_array) {
 713         @result = $self->$code($dbh, @_);
 714     }
 715     elsif(defined $want_array) {
 716         $result[0] = $self->$code($dbh, @_);
 717     }
 718     else {
 719         $self->$code($dbh, @_);
 720     }
 721   };
 722
 723   my $exception = $@;
 724   if(!$exception) { return $want_array ? @result : $result[0] }
 725
 726   $self->throw_exception($exception) if $self->connected;
 727
 728   # We were not connected - reconnect and retry, but let any
 729   #  exception fall right through this time
 730   $self->_populate_dbh;
 731   $self->$code($self->_dbh, @_);
 732 }
 733
 734 # This is basically a blend of dbh_do above and DBIx::Class::Storage::txn_do.
 735 # It also informs dbh_do to bypass itself while under the direction of txn_do,
 736 #  via $self->{_in_dbh_do} (this saves some redundant eval and errorcheck, etc)
 737 sub txn_do {
 738   my $self = shift;
 739   my $coderef = shift;
 740
 741   ref $coderef eq 'CODE' or $self->throw_exception
 742     ('$coderef must be a CODE reference');
 743
 744   return $coderef->(@_) if $self->{transaction_depth} && ! $self->auto_savepoint;
 745
 746   local $self->{_in_dbh_do} = 1;
 747
 748   my @result;
 749   my $want_array = wantarray;
 750
 751   my $tried = 0;
 752   while(1) {
 753     eval {
 754       $self->_verify_pid if $self->_dbh;
 755       $self->_populate_dbh if !$self->_dbh;
 756
 757       $self->txn_begin;
 758       if($want_array) {
 759           @result = $coderef->(@_);
 760       }
 761       elsif(defined $want_array) {
 762           $result[0] = $coderef->(@_);
 763       }
 764       else {
 765           $coderef->(@_);
 766       }
 767       $self->txn_commit;
 768     };
 769
 770     my $exception = $@;
 771     if(!$exception) { return $want_array ? @result : $result[0] }
 772
 773     if($tried++ > 0 || $self->connected) {
 774       eval { $self->txn_rollback };
 775       my $rollback_exception = $@;
 776       if($rollback_exception) {
 777         my $exception_class = "DBIx::Class::Storage::NESTED_ROLLBACK_EXCEPTION";
 778         $self->throw_exception($exception)  # propagate nested rollback
 779           if $rollback_exception =~ /$exception_class/;
 780
 781         $self->throw_exception(
 782           "Transaction aborted: ${exception}. "
 783           . "Rollback failed: ${rollback_exception}"
 784         );
 785       }
 786       $self->throw_exception($exception)
 787     }
 788
 789     # We were not connected, and was first try - reconnect and retry
 790     # via the while loop
 791     $self->_populate_dbh;
 792   }
 793 }
 794
 795 =head2 disconnect
 796
 797 Our C<disconnect> method also performs a rollback first if the
 798 database is not in C<AutoCommit> mode.
 799
 800 =cut
 801
 802 sub disconnect {
 803   my ($self) = @_;
 804
 805   if( $self->connected ) {
 806     my $connection_do = $self->on_disconnect_do;
 807     $self->_do_connection_actions($connection_do) if ref($connection_do);
 808
 809     $self->_dbh->rollback unless $self->_dbh_autocommit;
 810     $self->_dbh->disconnect;
 811     $self->_dbh(undef);
 812     $self->{_dbh_gen}++;
 813   }
 814 }
 815
 816 =head2 with_deferred_fk_checks
 817
 818 =over 4
 819
 820 =item Arguments: C<$coderef>
 821
 822 =item Return Value: The return value of $coderef
 823
 824 =back
 825
 826 Storage specific method to run the code ref with FK checks deferred or
 827 in MySQL's case disabled entirely.
 828
 829 =cut
 830
 831 # Storage subclasses should override this
 832 sub with_deferred_fk_checks {
 833   my ($self, $sub) = @_;
 834
 835   $sub->();
 836 }
 837
 838 sub connected {
 839   my ($self) = @_;
 840
 841   if(my $dbh = $self->_dbh) {
 842       if(defined $self->_conn_tid && $self->_conn_tid != threads->tid) {
 843           $self->_dbh(undef);
 844           $self->{_dbh_gen}++;
 845           return;
 846       }
 847       else {
 848           $self->_verify_pid;
 849           return 0 if !$self->_dbh;
 850       }
 851       return ($dbh->FETCH('Active') && $dbh->ping);
 852   }
 853
 854   return 0;
 855 }
 856
 857 # handle pid changes correctly
 858 #  NOTE: assumes $self->_dbh is a valid $dbh
 859 sub _verify_pid {
 860   my ($self) = @_;
 861
 862   return if defined $self->_conn_pid && $self->_conn_pid == $$;
 863
 864   $self->_dbh->{InactiveDestroy} = 1;
 865   $self->_dbh(undef);
 866   $self->{_dbh_gen}++;
 867
 868   return;
 869 }
 870
 871 sub ensure_connected {
 872   my ($self) = @_;
 873
 874   unless ($self->connected) {
 875     $self->_populate_dbh;
 876   }
 877 }
 878
 879 =head2 dbh
 880
 881 Returns the dbh - a data base handle of class L<DBI>.
 882
 883 =cut
 884
 885 sub dbh {
 886   my ($self) = @_;
 887
 888   $self->ensure_connected;
 889   return $self->_dbh;
 890 }
 891
 892 sub _sql_maker_args {
 893     my ($self) = @_;
 894
 895     return ( bindtype=>'columns', array_datatypes => 1, limit_dialect => $self->dbh, %{$self->_sql_maker_opts} );
 896 }
 897
 898 sub sql_maker {
 899   my ($self) = @_;
 900   unless ($self->_sql_maker) {
 901     my $sql_maker_class = $self->sql_maker_class;
 902     $self->_sql_maker($sql_maker_class->new( $self->_sql_maker_args ));
 903   }
 904   return $self->_sql_maker;
 905 }
 906
 907 sub _rebless {}
 908
 909 sub _populate_dbh {
 910   my ($self) = @_;
 911   my @info = @{$self->_dbi_connect_info || []};
 912   $self->_dbh($self->_connect(@info));
 913
 914   # Always set the transaction depth on connect, since
 915   #  there is no transaction in progress by definition
 916   $self->{transaction_depth} = $self->_dbh_autocommit ? 0 : 1;
 917
 918   if(ref $self eq 'DBIx::Class::Storage::DBI') {
 919     my $driver = $self->_dbh->{Driver}->{Name};
 920     if ($self->load_optional_class("DBIx::Class::Storage::DBI::${driver}")) {
 921       bless $self, "DBIx::Class::Storage::DBI::${driver}";
 922       $self->_rebless();
 923     }
 924   }
 925
 926   $self->_conn_pid($$);
 927   $self->_conn_tid(threads->tid) if $INC{'threads.pm'};
 928
 929   my $connection_do = $self->on_connect_do;
 930   $self->_do_connection_actions($connection_do) if ref($connection_do);
 931 }
 932
 933 sub _do_connection_actions {
 934   my $self = shift;
 935   my $connection_do = shift;
 936
 937   if (ref $connection_do eq 'ARRAY') {
 938     $self->_do_query($_) foreach @$connection_do;
 939   }
 940   elsif (ref $connection_do eq 'CODE') {
 941     $connection_do->($self);
 942   }
 943
 944   return $self;
 945 }
 946
 947 sub _do_query {
 948   my ($self, $action) = @_;
 949
 950   if (ref $action eq 'CODE') {
 951     $action = $action->($self);
 952     $self->_do_query($_) foreach @$action;
 953   }
 954   else {
 955     # Most debuggers expect ($sql, @bind), so we need to exclude
 956     # the attribute hash which is the second argument to $dbh->do
 957     # furthermore the bind values are usually to be presented
 958     # as named arrayref pairs, so wrap those here too
 959     my @do_args = (ref $action eq 'ARRAY') ? (@$action) : ($action);
 960     my $sql = shift @do_args;
 961     my $attrs = shift @do_args;
 962     my @bind = map { [ undef, $_ ] } @do_args;
 963
 964     $self->_query_start($sql, @bind);
 965     $self->_dbh->do($sql, $attrs, @do_args);
 966     $self->_query_end($sql, @bind);
 967   }
 968
 969   return $self;
 970 }
 971
 972 sub _connect {
 973   my ($self, @info) = @_;
 974
 975   $self->throw_exception("You failed to provide any connection info")
 976     if !@info;
 977
 978   my ($old_connect_via, $dbh);
 979
 980   if ($INC{'Apache/DBI.pm'} && $ENV{MOD_PERL}) {
 981     $old_connect_via = $DBI::connect_via;
 982     $DBI::connect_via = 'connect';
 983   }
 984
 985   eval {
 986     if(ref $info[0] eq 'CODE') {
 987        $dbh = &{$info[0]}
 988     }
 989     else {
 990        $dbh = DBI->connect(@info);
 991     }
 992
 993     if($dbh && !$self->unsafe) {
 994       my $weak_self = $self;
 995       weaken($weak_self);
 996       $dbh->{HandleError} = sub {
 997           if ($weak_self) {
 998             $weak_self->throw_exception("DBI Exception: $_[0]");
 999           }
1000           else {
1001             croak ("DBI Exception: $_[0]");
1002           }
1003       };
1004       $dbh->{ShowErrorStatement} = 1;
1005       $dbh->{RaiseError} = 1;
1006       $dbh->{PrintError} = 0;
1007     }
1008   };
1009
1010   $DBI::connect_via = $old_connect_via if $old_connect_via;
1011
1012   $self->throw_exception("DBI Connection failed: " . ($@||$DBI::errstr))
1013     if !$dbh || $@;
1014
1015   $self->_dbh_autocommit($dbh->{AutoCommit});
1016
1017   $dbh;
1018 }
1019
1020 sub svp_begin {
1021   my ($self, $name) = @_;
1022
1023   $name = $self->_svp_generate_name
1024     unless defined $name;
1025
1026   $self->throw_exception ("You can't use savepoints outside a transaction")
1027     if $self->{transaction_depth} == 0;
1028
1029   $self->throw_exception ("Your Storage implementation doesn't support savepoints")
1030     unless $self->can('_svp_begin');
1031
1032   push @{ $self->{savepoints} }, $name;
1033
1034   $self->debugobj->svp_begin($name) if $self->debug;
1035
1036   return $self->_svp_begin($name);
1037 }
1038
1039 sub svp_release {
1040   my ($self, $name) = @_;
1041
1042   $self->throw_exception ("You can't use savepoints outside a transaction")
1043     if $self->{transaction_depth} == 0;
1044
1045   $self->throw_exception ("Your Storage implementation doesn't support savepoints")
1046     unless $self->can('_svp_release');
1047
1048   if (defined $name) {
1049     $self->throw_exception ("Savepoint '$name' does not exist")
1050       unless grep { $_ eq $name } @{ $self->{savepoints} };
1051
1052     # Dig through the stack until we find the one we are releasing.  This keeps
1053     # the stack up to date.
1054     my $svp;
1055
1056     do { $svp = pop @{ $self->{savepoints} } } while $svp ne $name;
1057   } else {
1058     $name = pop @{ $self->{savepoints} };
1059   }
1060
1061   $self->debugobj->svp_release($name) if $self->debug;
1062
1063   return $self->_svp_release($name);
1064 }
1065
1066 sub svp_rollback {
1067   my ($self, $name) = @_;
1068
1069   $self->throw_exception ("You can't use savepoints outside a transaction")
1070     if $self->{transaction_depth} == 0;
1071
1072   $self->throw_exception ("Your Storage implementation doesn't support savepoints")
1073     unless $self->can('_svp_rollback');
1074
1075   if (defined $name) {
1076       # If they passed us a name, verify that it exists in the stack
1077       unless(grep({ $_ eq $name } @{ $self->{savepoints} })) {
1078           $self->throw_exception("Savepoint '$name' does not exist!");
1079       }
1080
1081       # Dig through the stack until we find the one we are releasing.  This keeps
1082       # the stack up to date.
1083       while(my $s = pop(@{ $self->{savepoints} })) {
1084           last if($s eq $name);
1085       }
1086       # Add the savepoint back to the stack, as a rollback doesn't remove the
1087       # named savepoint, only everything after it.
1088       push(@{ $self->{savepoints} }, $name);
1089   } else {
1090       # We'll assume they want to rollback to the last savepoint
1091       $name = $self->{savepoints}->[-1];
1092   }
1093
1094   $self->debugobj->svp_rollback($name) if $self->debug;
1095
1096   return $self->_svp_rollback($name);
1097 }
1098
1099 sub _svp_generate_name {
1100     my ($self) = @_;
1101
1102     return 'savepoint_'.scalar(@{ $self->{'savepoints'} });
1103 }
1104
1105 sub txn_begin {
1106   my $self = shift;
1107   $self->ensure_connected();
1108   if($self->{transaction_depth} == 0) {
1109     $self->debugobj->txn_begin()
1110       if $self->debug;
1111     # this isn't ->_dbh-> because
1112     #  we should reconnect on begin_work
1113     #  for AutoCommit users
1114     $self->dbh->begin_work;
1115   } elsif ($self->auto_savepoint) {
1116     $self->svp_begin;
1117   }
1118   $self->{transaction_depth}++;
1119 }
1120
1121 sub txn_commit {
1122   my $self = shift;
1123   if ($self->{transaction_depth} == 1) {
1124     my $dbh = $self->_dbh;
1125     $self->debugobj->txn_commit()
1126       if ($self->debug);
1127     $dbh->commit;
1128     $self->{transaction_depth} = 0
1129       if $self->_dbh_autocommit;
1130   }
1131   elsif($self->{transaction_depth} > 1) {
1132     $self->{transaction_depth}--;
1133     $self->svp_release
1134       if $self->auto_savepoint;
1135   }
1136 }
1137
1138 sub txn_rollback {
1139   my $self = shift;
1140   my $dbh = $self->_dbh;
1141   eval {
1142     if ($self->{transaction_depth} == 1) {
1143       $self->debugobj->txn_rollback()
1144         if ($self->debug);
1145       $self->{transaction_depth} = 0
1146         if $self->_dbh_autocommit;
1147       $dbh->rollback;
1148     }
1149     elsif($self->{transaction_depth} > 1) {
1150       $self->{transaction_depth}--;
1151       if ($self->auto_savepoint) {
1152         $self->svp_rollback;
1153         $self->svp_release;
1154       }
1155     }
1156     else {
1157       die DBIx::Class::Storage::NESTED_ROLLBACK_EXCEPTION->new;
1158     }
1159   };
1160   if ($@) {
1161     my $error = $@;
1162     my $exception_class = "DBIx::Class::Storage::NESTED_ROLLBACK_EXCEPTION";
1163     $error =~ /$exception_class/ and $self->throw_exception($error);
1164     # ensure that a failed rollback resets the transaction depth
1165     $self->{transaction_depth} = $self->_dbh_autocommit ? 0 : 1;
1166     $self->throw_exception($error);
1167   }
1168 }
1169
1170 # This used to be the top-half of _execute.  It was split out to make it
1171 #  easier to override in NoBindVars without duping the rest.  It takes up
1172 #  all of _execute's args, and emits $sql, @bind.
1173 sub _prep_for_execute {
1174   my ($self, $op, $extra_bind, $ident, $args) = @_;
1175
1176   if( blessed($ident) && $ident->isa("DBIx::Class::ResultSource") ) {
1177     $ident = $ident->from();
1178   }
1179
1180   my ($sql, @bind) = $self->sql_maker->$op($ident, @$args);
1181
1182   unshift(@bind,
1183     map { ref $_ eq 'ARRAY' ? $_ : [ '!!dummy', $_ ] } @$extra_bind)
1184       if $extra_bind;
1185   return ($sql, \@bind);
1186 }
1187
1188 sub _fix_bind_params {
1189     my ($self, @bind) = @_;
1190
1191     ### Turn @bind from something like this:
1192     ###   ( [ "artist", 1 ], [ "cdid", 1, 3 ] )
1193     ### to this:
1194     ###   ( "'1'", "'1'", "'3'" )
1195     return
1196         map {
1197             if ( defined( $_ && $_->[1] ) ) {
1198                 map { qq{'$_'}; } @{$_}[ 1 .. $#$_ ];
1199             }
1200             else { q{'NULL'}; }
1201         } @bind;
1202 }
1203
1204 sub _query_start {
1205     my ( $self, $sql, @bind ) = @_;
1206
1207     if ( $self->debug ) {
1208         @bind = $self->_fix_bind_params(@bind);
1209
1210         $self->debugobj->query_start( $sql, @bind );
1211     }
1212 }
1213
1214 sub _query_end {
1215     my ( $self, $sql, @bind ) = @_;
1216
1217     if ( $self->debug ) {
1218         @bind = $self->_fix_bind_params(@bind);
1219         $self->debugobj->query_end( $sql, @bind );
1220     }
1221 }
1222
1223 sub _dbh_execute {
1224   my ($self, $dbh, $op, $extra_bind, $ident, $bind_attributes, @args) = @_;
1225
1226   my ($sql, $bind) = $self->_prep_for_execute($op, $extra_bind, $ident, \@args);
1227
1228   $self->_query_start( $sql, @$bind );
1229
1230   my $sth = $self->sth($sql,$op);
1231
1232   my $placeholder_index = 1;
1233
1234   foreach my $bound (@$bind) {
1235     my $attributes = {};
1236     my($column_name, @data) = @$bound;
1237
1238     if ($bind_attributes) {
1239       $attributes = $bind_attributes->{$column_name}
1240       if defined $bind_attributes->{$column_name};
1241     }
1242
1243     foreach my $data (@data) {
1244       my $ref = ref $data;
1245       $data = $ref && $ref ne 'ARRAY' ? ''.$data : $data; # stringify args (except arrayrefs)
1246
1247       $sth->bind_param($placeholder_index, $data, $attributes);
1248       $placeholder_index++;
1249     }
1250   }
1251
1252   # Can this fail without throwing an exception anyways???
1253   my $rv = $sth->execute();
1254   $self->throw_exception($sth->errstr) if !$rv;
1255
1256   $self->_query_end( $sql, @$bind );
1257
1258   return (wantarray ? ($rv, $sth, @$bind) : $rv);
1259 }
1260
1261 sub _execute {
1262     my $self = shift;
1263     $self->dbh_do('_dbh_execute', @_)
1264 }
1265
1266 sub insert {
1267   my ($self, $source, $to_insert) = @_;
1268
1269   my $ident = $source->from;
1270   my $bind_attributes = $self->source_bind_attributes($source);
1271
1272   $self->ensure_connected;
1273   foreach my $col ( $source->columns ) {
1274     if ( !defined $to_insert->{$col} ) {
1275       my $col_info = $source->column_info($col);
1276
1277       if ( $col_info->{auto_nextval} ) {
1278         $to_insert->{$col} = $self->_sequence_fetch( 'nextval', $col_info->{sequence} || $self->_dbh_get_autoinc_seq($self->dbh, $source) );
1279       }
1280     }
1281   }
1282
1283   $self->_execute('insert' => [], $source, $bind_attributes, $to_insert);
1284
1285   return $to_insert;
1286 }
1287
1288 ## Still not quite perfect, and EXPERIMENTAL
1289 ## Currently it is assumed that all values passed will be "normal", i.e. not
1290 ## scalar refs, or at least, all the same type as the first set, the statement is
1291 ## only prepped once.
1292 sub insert_bulk {
1293   my ($self, $source, $cols, $data) = @_;
1294   my %colvalues;
1295   my $table = $source->from;
1296   @colvalues{@$cols} = (0..$#$cols);
1297   my ($sql, @bind) = $self->sql_maker->insert($table, \%colvalues);
1298
1299   $self->_query_start( $sql, @bind );
1300   my $sth = $self->sth($sql);
1301
1302 #  @bind = map { ref $_ ? ''.$_ : $_ } @bind; # stringify args
1303
1304   ## This must be an arrayref, else nothing works!
1305
1306   my $tuple_status = [];
1307
1308   ##use Data::Dumper;
1309   ##print STDERR Dumper( $data, $sql, [@bind] );
1310
1311   my $time = time();
1312
1313   ## Get the bind_attributes, if any exist
1314   my $bind_attributes = $self->source_bind_attributes($source);
1315
1316   ## Bind the values and execute
1317   my $placeholder_index = 1;
1318
1319   foreach my $bound (@bind) {
1320
1321     my $attributes = {};
1322     my ($column_name, $data_index) = @$bound;
1323
1324     if( $bind_attributes ) {
1325       $attributes = $bind_attributes->{$column_name}
1326       if defined $bind_attributes->{$column_name};
1327     }
1328
1329     my @data = map { $_->[$data_index] } @$data;
1330
1331     $sth->bind_param_array( $placeholder_index, [@data], $attributes );
1332     $placeholder_index++;
1333   }
1334   my $rv = $sth->execute_array({ArrayTupleStatus => $tuple_status});
1335   $self->throw_exception($sth->errstr) if !$rv;
1336
1337   $self->_query_end( $sql, @bind );
1338   return (wantarray ? ($rv, $sth, @bind) : $rv);
1339 }
1340
1341 sub update {
1342   my $self = shift @_;
1343   my $source = shift @_;
1344   my $bind_attributes = $self->source_bind_attributes($source);
1345
1346   return $self->_execute('update' => [], $source, $bind_attributes, @_);
1347 }
1348
1349
1350 sub delete {
1351   my $self = shift @_;
1352   my $source = shift @_;
1353
1354   my $bind_attrs = {}; ## If ever it's needed...
1355
1356   return $self->_execute('delete' => [], $source, $bind_attrs, @_);
1357 }
1358
1359 sub _select {
1360   my $self = shift;
1361   my $sql_maker = $self->sql_maker;
1362   local $sql_maker->{for};
1363   return $self->_execute($self->_select_args(@_));
1364 }
1365
1366 sub _select_args {
1367   my ($self, $ident, $select, $condition, $attrs) = @_;
1368   my $order = $attrs->{order_by};
1369
1370   if (ref $condition eq 'SCALAR') {
1371     my $unwrap = ${$condition};
1372     if ($unwrap =~ s/ORDER BY (.*)$//i) {
1373       $order = $1;
1374       $condition = \$unwrap;
1375     }
1376   }
1377
1378   my $for = delete $attrs->{for};
1379   my $sql_maker = $self->sql_maker;
1380   local $sql_maker->{for} = $for;
1381
1382   if (exists $attrs->{group_by} || $attrs->{having}) {
1383     $order = {
1384       group_by => $attrs->{group_by},
1385       having => $attrs->{having},
1386       ($order ? (order_by => $order) : ())
1387     };
1388   }
1389   my $bind_attrs = {}; ## Future support
1390   my @args = ('select', $attrs->{bind}, $ident, $bind_attrs, $select, $condition, $order);
1391   if ($attrs->{software_limit} ||
1392       $self->sql_maker->_default_limit_syntax eq "GenericSubQ") {
1393         $attrs->{software_limit} = 1;
1394   } else {
1395     $self->throw_exception("rows attribute must be positive if present")
1396       if (defined($attrs->{rows}) && !($attrs->{rows} > 0));
1397
1398     # MySQL actually recommends this approach.  I cringe.
1399     $attrs->{rows} = 2**48 if not defined $attrs->{rows} and defined $attrs->{offset};
1400     push @args, $attrs->{rows}, $attrs->{offset};
1401   }
1402
1403   return @args;
1404 }
1405
1406 sub source_bind_attributes {
1407   my ($self, $source) = @_;
1408
1409   my $bind_attributes;
1410   foreach my $column ($source->columns) {
1411
1412     my $data_type = $source->column_info($column)->{data_type} || '';
1413     $bind_attributes->{$column} = $self->bind_attribute_by_data_type($data_type)
1414      if $data_type;
1415   }
1416
1417   return $bind_attributes;
1418 }
1419
1420 =head2 select
1421
1422 =over 4
1423
1424 =item Arguments: $ident, $select, $condition, $attrs
1425
1426 =back
1427
1428 Handle a SQL select statement.
1429
1430 =cut
1431
1432 sub select {
1433   my $self = shift;
1434   my ($ident, $select, $condition, $attrs) = @_;
1435   return $self->cursor_class->new($self, \@_, $attrs);
1436 }
1437
1438 sub select_single {
1439   my $self = shift;
1440   my ($rv, $sth, @bind) = $self->_select(@_);
1441   my @row = $sth->fetchrow_array;
1442   my @nextrow = $sth->fetchrow_array if @row;
1443   if(@row && @nextrow) {
1444     carp "Query returned more than one row.  SQL that returns multiple rows is DEPRECATED for ->find and ->single";
1445   }
1446   # Need to call finish() to work round broken DBDs
1447   $sth->finish();
1448   return @row;
1449 }
1450
1451 =head2 sth
1452
1453 =over 4
1454
1455 =item Arguments: $sql
1456
1457 =back
1458
1459 Returns a L<DBI> sth (statement handle) for the supplied SQL.
1460
1461 =cut
1462
1463 sub _dbh_sth {
1464   my ($self, $dbh, $sql) = @_;
1465
1466   # 3 is the if_active parameter which avoids active sth re-use
1467   my $sth = $self->disable_sth_caching
1468     ? $dbh->prepare($sql)
1469     : $dbh->prepare_cached($sql, {}, 3);
1470
1471   # XXX You would think RaiseError would make this impossible,
1472   #  but apparently that's not true :(
1473   $self->throw_exception($dbh->errstr) if !$sth;
1474
1475   $sth;
1476 }
1477
1478 sub sth {
1479   my ($self, $sql) = @_;
1480   $self->dbh_do('_dbh_sth', $sql);
1481 }
1482
1483 sub _dbh_columns_info_for {
1484   my ($self, $dbh, $table) = @_;
1485
1486   if ($dbh->can('column_info')) {
1487     my %result;
1488     eval {
1489       my ($schema,$tab) = $table =~ /^(.+?)\.(.+)$/ ? ($1,$2) : (undef,$table);
1490       my $sth = $dbh->column_info( undef,$schema, $tab, '%' );
1491       $sth->execute();
1492       while ( my $info = $sth->fetchrow_hashref() ){
1493         my %column_info;
1494         $column_info{data_type}   = $info->{TYPE_NAME};
1495         $column_info{size}      = $info->{COLUMN_SIZE};
1496         $column_info{is_nullable}   = $info->{NULLABLE} ? 1 : 0;
1497         $column_info{default_value} = $info->{COLUMN_DEF};
1498         my $col_name = $info->{COLUMN_NAME};
1499         $col_name =~ s/^\"(.*)\"$/$1/;
1500
1501         $result{$col_name} = \%column_info;
1502       }
1503     };
1504     return \%result if !$@ && scalar keys %result;
1505   }
1506
1507   my %result;
1508   my $sth = $dbh->prepare($self->sql_maker->select($table, undef, \'1 = 0'));
1509   $sth->execute;
1510   my @columns = @{$sth->{NAME_lc}};
1511   for my $i ( 0 .. $#columns ){
1512     my %column_info;
1513     $column_info{data_type} = $sth->{TYPE}->[$i];
1514     $column_info{size} = $sth->{PRECISION}->[$i];
1515     $column_info{is_nullable} = $sth->{NULLABLE}->[$i] ? 1 : 0;
1516
1517     if ($column_info{data_type} =~ m/^(.*?)\((.*?)\)$/) {
1518       $column_info{data_type} = $1;
1519       $column_info{size}    = $2;
1520     }
1521
1522     $result{$columns[$i]} = \%column_info;
1523   }
1524   $sth->finish;
1525
1526   foreach my $col (keys %result) {
1527     my $colinfo = $result{$col};
1528     my $type_num = $colinfo->{data_type};
1529     my $type_name;
1530     if(defined $type_num && $dbh->can('type_info')) {
1531       my $type_info = $dbh->type_info($type_num);
1532       $type_name = $type_info->{TYPE_NAME} if $type_info;
1533       $colinfo->{data_type} = $type_name if $type_name;
1534     }
1535   }
1536
1537   return \%result;
1538 }
1539
1540 sub columns_info_for {
1541   my ($self, $table) = @_;
1542   $self->dbh_do('_dbh_columns_info_for', $table);
1543 }
1544
1545 =head2 last_insert_id
1546
1547 Return the row id of the last insert.
1548
1549 =cut
1550
1551 sub _dbh_last_insert_id {
1552     my ($self, $dbh, $source, $col) = @_;
1553     # XXX This is a SQLite-ism as a default... is there a DBI-generic way?
1554     $dbh->func('last_insert_rowid');
1555 }
1556
1557 sub last_insert_id {
1558   my $self = shift;
1559   $self->dbh_do('_dbh_last_insert_id', @_);
1560 }
1561
1562 =head2 sqlt_type
1563
1564 Returns the database driver name.
1565
1566 =cut
1567
1568 sub sqlt_type { shift->dbh->{Driver}->{Name} }
1569
1570 =head2 bind_attribute_by_data_type
1571
1572 Given a datatype from column info, returns a database specific bind
1573 attribute for C<< $dbh->bind_param($val,$attribute) >> or nothing if we will
1574 let the database planner just handle it.
1575
1576 Generally only needed for special case column types, like bytea in postgres.
1577
1578 =cut
1579
1580 sub bind_attribute_by_data_type {
1581     return;
1582 }
1583
1584 =head2 create_ddl_dir
1585
1586 =over 4
1587
1588 =item Arguments: $schema \@databases, $version, $directory, $preversion, \%sqlt_args
1589
1590 =back
1591
1592 Creates a SQL file based on the Schema, for each of the specified
1593 database types, in the given directory.
1594
1595 By default, C<\%sqlt_args> will have
1596
1597  { add_drop_table => 1, ignore_constraint_names => 1, ignore_index_names => 1 }
1598
1599 merged with the hash passed in. To disable any of those features, pass in a
1600 hashref like the following
1601
1602  { ignore_constraint_names => 0, # ... other options }
1603
1604 =cut
1605
1606 sub create_ddl_dir {
1607   my ($self, $schema, $databases, $version, $dir, $preversion, $sqltargs) = @_;
1608
1609   if(!$dir || !-d $dir) {
1610     warn "No directory given, using ./\n";
1611     $dir = "./";
1612   }
1613   $databases ||= ['MySQL', 'SQLite', 'PostgreSQL'];
1614   $databases = [ $databases ] if(ref($databases) ne 'ARRAY');
1615
1616   my $schema_version = $schema->schema_version || '1.x';
1617   $version ||= $schema_version;
1618
1619   $sqltargs = {
1620     add_drop_table => 1,
1621     ignore_constraint_names => 1,
1622     ignore_index_names => 1,
1623     %{$sqltargs || {}}
1624   };
1625
1626   $self->throw_exception(q{Can't create a ddl file without SQL::Translator 0.09003: '}
1627       . $self->_check_sqlt_message . q{'})
1628           if !$self->_check_sqlt_version;
1629
1630   my $sqlt = SQL::Translator->new( $sqltargs );
1631
1632   $sqlt->parser('SQL::Translator::Parser::DBIx::Class');
1633   my $sqlt_schema = $sqlt->translate({ data => $schema }) or die $sqlt->error;
1634
1635   foreach my $db (@$databases) {
1636     $sqlt->reset();
1637     $sqlt = $self->configure_sqlt($sqlt, $db);
1638     $sqlt->{schema} = $sqlt_schema;
1639     $sqlt->producer($db);
1640
1641     my $file;
1642     my $filename = $schema->ddl_filename($db, $version, $dir);
1643     if (-e $filename && ($version eq $schema_version )) {
1644       # if we are dumping the current version, overwrite the DDL
1645       warn "Overwriting existing DDL file - $filename";
1646       unlink($filename);
1647     }
1648
1649     my $output = $sqlt->translate;
1650     if(!$output) {
1651       warn("Failed to translate to $db, skipping. (" . $sqlt->error . ")");
1652       next;
1653     }
1654     if(!open($file, ">$filename")) {
1655       $self->throw_exception("Can't open $filename for writing ($!)");
1656       next;
1657     }
1658     print $file $output;
1659     close($file);
1660
1661     next unless ($preversion);
1662
1663     require SQL::Translator::Diff;
1664
1665     my $prefilename = $schema->ddl_filename($db, $preversion, $dir);
1666     if(!-e $prefilename) {
1667       warn("No previous schema file found ($prefilename)");
1668       next;
1669     }
1670
1671     my $difffile = $schema->ddl_filename($db, $version, $dir, $preversion);
1672     if(-e $difffile) {
1673       warn("Overwriting existing diff file - $difffile");
1674       unlink($difffile);
1675     }
1676
1677     my $source_schema;
1678     {
1679       my $t = SQL::Translator->new($sqltargs);
1680       $t->debug( 0 );
1681       $t->trace( 0 );
1682       $t->parser( $db )                       or die $t->error;
1683       $t = $self->configure_sqlt($t, $db);
1684       my $out = $t->translate( $prefilename ) or die $t->error;
1685       $source_schema = $t->schema;
1686       unless ( $source_schema->name ) {
1687         $source_schema->name( $prefilename );
1688       }
1689     }
1690
1691     # The "new" style of producers have sane normalization and can support
1692     # diffing a SQL file against a DBIC->SQLT schema. Old style ones don't
1693     # And we have to diff parsed SQL against parsed SQL.
1694     my $dest_schema = $sqlt_schema;
1695
1696     unless ( "SQL::Translator::Producer::$db"->can('preprocess_schema') ) {
1697       my $t = SQL::Translator->new($sqltargs);
1698       $t->debug( 0 );
1699       $t->trace( 0 );
1700       $t->parser( $db )                    or die $t->error;
1701       $t = $self->configure_sqlt($t, $db);
1702       my $out = $t->translate( $filename ) or die $t->error;
1703       $dest_schema = $t->schema;
1704       $dest_schema->name( $filename )
1705         unless $dest_schema->name;
1706     }
1707
1708     my $diff = SQL::Translator::Diff::schema_diff($source_schema, $db,
1709                                                   $dest_schema,   $db,
1710                                                   $sqltargs
1711                                                  );
1712     if(!open $file, ">$difffile") {
1713       $self->throw_exception("Can't write to $difffile ($!)");
1714       next;
1715     }
1716     print $file $diff;
1717     close($file);
1718   }
1719 }
1720
1721 sub configure_sqlt() {
1722   my $self = shift;
1723   my $tr = shift;
1724   my $db = shift || $self->sqlt_type;
1725   if ($db eq 'PostgreSQL') {
1726     $tr->quote_table_names(0);
1727     $tr->quote_field_names(0);
1728   }
1729   return $tr;
1730 }
1731
1732 =head2 deployment_statements
1733
1734 =over 4
1735
1736 =item Arguments: $schema, $type, $version, $directory, $sqlt_args
1737
1738 =back
1739
1740 Returns the statements used by L</deploy> and L<DBIx::Class::Schema/deploy>.
1741 The database driver name is given by C<$type>, though the value from
1742 L</sqlt_type> is used if it is not specified.
1743
1744 C<$directory> is used to return statements from files in a previously created
1745 L</create_ddl_dir> directory and is optional. The filenames are constructed
1746 from L<DBIx::Class::Schema/ddl_filename>, the schema name and the C<$version>.
1747
1748 If no C<$directory> is specified then the statements are constructed on the
1749 fly using L<SQL::Translator> and C<$version> is ignored.
1750
1751 See L<SQL::Translator/METHODS> for a list of values for C<$sqlt_args>.
1752
1753 =cut
1754
1755 sub deployment_statements {
1756   my ($self, $schema, $type, $version, $dir, $sqltargs) = @_;
1757   # Need to be connected to get the correct sqlt_type
1758   $self->ensure_connected() unless $type;
1759   $type ||= $self->sqlt_type;
1760   $version ||= $schema->schema_version || '1.x';
1761   $dir ||= './';
1762   my $filename = $schema->ddl_filename($type, $dir, $version);
1763   if(-f $filename)
1764   {
1765       my $file;
1766       open($file, "<$filename")
1767         or $self->throw_exception("Can't open $filename ($!)");
1768       my @rows = <$file>;
1769       close($file);
1770       return join('', @rows);
1771   }
1772
1773   $self->throw_exception(q{Can't deploy without SQL::Translator 0.09003: '}
1774       . $self->_check_sqlt_message . q{'})
1775           if !$self->_check_sqlt_version;
1776
1777   require SQL::Translator::Parser::DBIx::Class;
1778   eval qq{use SQL::Translator::Producer::${type}};
1779   $self->throw_exception($@) if $@;
1780
1781   # sources needs to be a parser arg, but for simplicty allow at top level
1782   # coming in
1783   $sqltargs->{parser_args}{sources} = delete $sqltargs->{sources}
1784       if exists $sqltargs->{sources};
1785
1786   my $tr = SQL::Translator->new(%$sqltargs);
1787   SQL::Translator::Parser::DBIx::Class::parse( $tr, $schema );
1788   return "SQL::Translator::Producer::${type}"->can('produce')->($tr);
1789 }
1790
1791 sub deploy {
1792   my ($self, $schema, $type, $sqltargs, $dir) = @_;
1793   my $deploy = sub {
1794     my $line = shift;
1795     return if($line =~ /^--/);
1796     return if(!$line);
1797     # next if($line =~ /^DROP/m);
1798     return if($line =~ /^BEGIN TRANSACTION/m);
1799     return if($line =~ /^COMMIT/m);
1800     return if $line =~ /^\s+$/; # skip whitespace only
1801     $self->_query_start($line);
1802     eval {
1803       $self->dbh->do($line); # shouldn't be using ->dbh ?
1804     };
1805     if ($@) {
1806       warn qq{$@ (running "${line}")};
1807     }
1808     $self->_query_end($line);
1809   };
1810   my @statements = $self->deployment_statements($schema, $type, undef, $dir, { no_comments => 1, %{ $sqltargs || {} } } );
1811   if (@statements > 1) {
1812     foreach my $statement (@statements) {
1813       $deploy->( $statement );
1814     }
1815   }
1816   elsif (@statements == 1) {
1817     foreach my $line ( split(";\n", $statements[0])) {
1818       $deploy->( $line );
1819     }
1820   }
1821 }
1822
1823 =head2 datetime_parser
1824
1825 Returns the datetime parser class
1826
1827 =cut
1828
1829 sub datetime_parser {
1830   my $self = shift;
1831   return $self->{datetime_parser} ||= do {
1832     $self->ensure_connected;
1833     $self->build_datetime_parser(@_);
1834   };
1835 }
1836
1837 =head2 datetime_parser_type
1838
1839 Defines (returns) the datetime parser class - currently hardwired to
1840 L<DateTime::Format::MySQL>
1841
1842 =cut
1843
1844 sub datetime_parser_type { "DateTime::Format::MySQL"; }
1845
1846 =head2 build_datetime_parser
1847
1848 See L</datetime_parser>
1849
1850 =cut
1851
1852 sub build_datetime_parser {
1853   my $self = shift;
1854   my $type = $self->datetime_parser_type(@_);
1855   eval "use ${type}";
1856   $self->throw_exception("Couldn't load ${type}: $@") if $@;
1857   return $type;
1858 }
1859
1860 {
1861     my $_check_sqlt_version; # private
1862     my $_check_sqlt_message; # private
1863     sub _check_sqlt_version {
1864         return $_check_sqlt_version if defined $_check_sqlt_version;
1865         eval 'use SQL::Translator "0.09003"';
1866         $_check_sqlt_message = $@ || '';
1867         $_check_sqlt_version = !$@;
1868     }
1869
1870     sub _check_sqlt_message {
1871         _check_sqlt_version if !defined $_check_sqlt_message;
1872         $_check_sqlt_message;
1873     }
1874 }
1875
1876 =head2 is_replicating
1877
1878 A boolean that reports if a particular L<DBIx::Class::Storage::DBI> is set to
1879 replicate from a master database.  Default is undef, which is the result
1880 returned by databases that don't support replication.
1881
1882 =cut
1883
1884 sub is_replicating {
1885     return;
1886
1887 }
1888
1889 =head2 lag_behind_master
1890
1891 Returns a number that represents a certain amount of lag behind a master db
1892 when a given storage is replicating.  The number is database dependent, but
1893 starts at zero and increases with the amount of lag. Default in undef
1894
1895 =cut
1896
1897 sub lag_behind_master {
1898     return;
1899 }
1900
1901 sub DESTROY {
1902   my $self = shift;
1903   return if !$self->_dbh;
1904   $self->_verify_pid;
1905   $self->_dbh(undef);
1906 }
1907
1908 1;
1909
1910 =head1 USAGE NOTES
1911
1912 =head2 DBIx::Class and AutoCommit
1913
1914 DBIx::Class can do some wonderful magic with handling exceptions,
1915 disconnections, and transactions when you use C<< AutoCommit => 1 >>
1916 combined with C<txn_do> for transaction support.
1917
1918 If you set C<< AutoCommit => 0 >> in your connect info, then you are always
1919 in an assumed transaction between commits, and you're telling us you'd
1920 like to manage that manually.  A lot of the magic protections offered by
1921 this module will go away.  We can't protect you from exceptions due to database
1922 disconnects because we don't know anything about how to restart your
1923 transactions.  You're on your own for handling all sorts of exceptional
1924 cases if you choose the C<< AutoCommit => 0 >> path, just as you would
1925 be with raw DBI.
1926
1927
1928 =head1 SQL METHODS
1929
1930 The module defines a set of methods within the DBIC::SQL::Abstract
1931 namespace.  These build on L<SQL::Abstract::Limit> to provide the
1932 SQL query functions.
1933
1934 The following methods are extended:-
1935
1936 =over 4
1937
1938 =item delete
1939
1940 =item insert
1941
1942 =item select
1943
1944 =item update
1945
1946 =item limit_dialect
1947
1948 See L</connect_info> for details.
1949
1950 =item quote_char
1951
1952 See L</connect_info> for details.
1953
1954 =item name_sep
1955
1956 See L</connect_info> for details.
1957
1958 =back
1959
1960 =head1 AUTHORS
1961
1962 Matt S. Trout <mst@shadowcatsystems.co.uk>
1963
1964 Andy Grundman <andy@hybridized.org>
1965
1966 =head1 LICENSE
1967
1968 You may distribute this code under the same terms as Perl itself.
1969
1970 =cut