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   my ($sql, @bind) = $self->sql_maker->$op($ident, @$args);
1177   unshift(@bind,
1178     map { ref $_ eq 'ARRAY' ? $_ : [ '!!dummy', $_ ] } @$extra_bind)
1179       if $extra_bind;
1180
1181   return ($sql, \@bind);
1182 }
1183
1184 sub _fix_bind_params {
1185     my ($self, @bind) = @_;
1186
1187     ### Turn @bind from something like this:
1188     ###   ( [ "artist", 1 ], [ "cdid", 1, 3 ] )
1189     ### to this:
1190     ###   ( "'1'", "'1'", "'3'" )
1191     return
1192         map {
1193             if ( defined( $_ && $_->[1] ) ) {
1194                 map { qq{'$_'}; } @{$_}[ 1 .. $#$_ ];
1195             }
1196             else { q{'NULL'}; }
1197         } @bind;
1198 }
1199
1200 sub _query_start {
1201     my ( $self, $sql, @bind ) = @_;
1202
1203     if ( $self->debug ) {
1204         @bind = $self->_fix_bind_params(@bind);
1205
1206         $self->debugobj->query_start( $sql, @bind );
1207     }
1208 }
1209
1210 sub _query_end {
1211     my ( $self, $sql, @bind ) = @_;
1212
1213     if ( $self->debug ) {
1214         @bind = $self->_fix_bind_params(@bind);
1215         $self->debugobj->query_end( $sql, @bind );
1216     }
1217 }
1218
1219 sub _dbh_execute {
1220   my ($self, $dbh, $op, $extra_bind, $ident, $bind_attributes, @args) = @_;
1221
1222   if( blessed($ident) && $ident->isa("DBIx::Class::ResultSource") ) {
1223     $ident = $ident->from();
1224   }
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, $ident, $select, $condition, $attrs) = @_;
1361   my $order = $attrs->{order_by};
1362
1363   if (ref $condition eq 'SCALAR') {
1364     my $unwrap = ${$condition};
1365     if ($unwrap =~ s/ORDER BY (.*)$//i) {
1366       $order = $1;
1367       $condition = \$unwrap;
1368     }
1369   }
1370
1371   my $for = delete $attrs->{for};
1372   my $sql_maker = $self->sql_maker;
1373   local $sql_maker->{for} = $for;
1374
1375   if (exists $attrs->{group_by} || $attrs->{having}) {
1376     $order = {
1377       group_by => $attrs->{group_by},
1378       having => $attrs->{having},
1379       ($order ? (order_by => $order) : ())
1380     };
1381   }
1382   my $bind_attrs = {}; ## Future support
1383   my @args = ('select', $attrs->{bind}, $ident, $bind_attrs, $select, $condition, $order);
1384   if ($attrs->{software_limit} ||
1385       $self->sql_maker->_default_limit_syntax eq "GenericSubQ") {
1386         $attrs->{software_limit} = 1;
1387   } else {
1388     $self->throw_exception("rows attribute must be positive if present")
1389       if (defined($attrs->{rows}) && !($attrs->{rows} > 0));
1390
1391     # MySQL actually recommends this approach.  I cringe.
1392     $attrs->{rows} = 2**48 if not defined $attrs->{rows} and defined $attrs->{offset};
1393     push @args, $attrs->{rows}, $attrs->{offset};
1394   }
1395
1396   return $self->_execute(@args);
1397 }
1398
1399 sub source_bind_attributes {
1400   my ($self, $source) = @_;
1401
1402   my $bind_attributes;
1403   foreach my $column ($source->columns) {
1404
1405     my $data_type = $source->column_info($column)->{data_type} || '';
1406     $bind_attributes->{$column} = $self->bind_attribute_by_data_type($data_type)
1407      if $data_type;
1408   }
1409
1410   return $bind_attributes;
1411 }
1412
1413 =head2 select
1414
1415 =over 4
1416
1417 =item Arguments: $ident, $select, $condition, $attrs
1418
1419 =back
1420
1421 Handle a SQL select statement.
1422
1423 =cut
1424
1425 sub select {
1426   my $self = shift;
1427   my ($ident, $select, $condition, $attrs) = @_;
1428   return $self->cursor_class->new($self, \@_, $attrs);
1429 }
1430
1431 sub select_single {
1432   my $self = shift;
1433   my ($rv, $sth, @bind) = $self->_select(@_);
1434   my @row = $sth->fetchrow_array;
1435   my @nextrow = $sth->fetchrow_array if @row;
1436   if(@row && @nextrow) {
1437     carp "Query returned more than one row.  SQL that returns multiple rows is DEPRECATED for ->find and ->single";
1438   }
1439   # Need to call finish() to work round broken DBDs
1440   $sth->finish();
1441   return @row;
1442 }
1443
1444 =head2 sth
1445
1446 =over 4
1447
1448 =item Arguments: $sql
1449
1450 =back
1451
1452 Returns a L<DBI> sth (statement handle) for the supplied SQL.
1453
1454 =cut
1455
1456 sub _dbh_sth {
1457   my ($self, $dbh, $sql) = @_;
1458
1459   # 3 is the if_active parameter which avoids active sth re-use
1460   my $sth = $self->disable_sth_caching
1461     ? $dbh->prepare($sql)
1462     : $dbh->prepare_cached($sql, {}, 3);
1463
1464   # XXX You would think RaiseError would make this impossible,
1465   #  but apparently that's not true :(
1466   $self->throw_exception($dbh->errstr) if !$sth;
1467
1468   $sth;
1469 }
1470
1471 sub sth {
1472   my ($self, $sql) = @_;
1473   $self->dbh_do('_dbh_sth', $sql);
1474 }
1475
1476 sub _dbh_columns_info_for {
1477   my ($self, $dbh, $table) = @_;
1478
1479   if ($dbh->can('column_info')) {
1480     my %result;
1481     eval {
1482       my ($schema,$tab) = $table =~ /^(.+?)\.(.+)$/ ? ($1,$2) : (undef,$table);
1483       my $sth = $dbh->column_info( undef,$schema, $tab, '%' );
1484       $sth->execute();
1485       while ( my $info = $sth->fetchrow_hashref() ){
1486         my %column_info;
1487         $column_info{data_type}   = $info->{TYPE_NAME};
1488         $column_info{size}      = $info->{COLUMN_SIZE};
1489         $column_info{is_nullable}   = $info->{NULLABLE} ? 1 : 0;
1490         $column_info{default_value} = $info->{COLUMN_DEF};
1491         my $col_name = $info->{COLUMN_NAME};
1492         $col_name =~ s/^\"(.*)\"$/$1/;
1493
1494         $result{$col_name} = \%column_info;
1495       }
1496     };
1497     return \%result if !$@ && scalar keys %result;
1498   }
1499
1500   my %result;
1501   my $sth = $dbh->prepare($self->sql_maker->select($table, undef, \'1 = 0'));
1502   $sth->execute;
1503   my @columns = @{$sth->{NAME_lc}};
1504   for my $i ( 0 .. $#columns ){
1505     my %column_info;
1506     $column_info{data_type} = $sth->{TYPE}->[$i];
1507     $column_info{size} = $sth->{PRECISION}->[$i];
1508     $column_info{is_nullable} = $sth->{NULLABLE}->[$i] ? 1 : 0;
1509
1510     if ($column_info{data_type} =~ m/^(.*?)\((.*?)\)$/) {
1511       $column_info{data_type} = $1;
1512       $column_info{size}    = $2;
1513     }
1514
1515     $result{$columns[$i]} = \%column_info;
1516   }
1517   $sth->finish;
1518
1519   foreach my $col (keys %result) {
1520     my $colinfo = $result{$col};
1521     my $type_num = $colinfo->{data_type};
1522     my $type_name;
1523     if(defined $type_num && $dbh->can('type_info')) {
1524       my $type_info = $dbh->type_info($type_num);
1525       $type_name = $type_info->{TYPE_NAME} if $type_info;
1526       $colinfo->{data_type} = $type_name if $type_name;
1527     }
1528   }
1529
1530   return \%result;
1531 }
1532
1533 sub columns_info_for {
1534   my ($self, $table) = @_;
1535   $self->dbh_do('_dbh_columns_info_for', $table);
1536 }
1537
1538 =head2 last_insert_id
1539
1540 Return the row id of the last insert.
1541
1542 =cut
1543
1544 sub _dbh_last_insert_id {
1545     my ($self, $dbh, $source, $col) = @_;
1546     # XXX This is a SQLite-ism as a default... is there a DBI-generic way?
1547     $dbh->func('last_insert_rowid');
1548 }
1549
1550 sub last_insert_id {
1551   my $self = shift;
1552   $self->dbh_do('_dbh_last_insert_id', @_);
1553 }
1554
1555 =head2 sqlt_type
1556
1557 Returns the database driver name.
1558
1559 =cut
1560
1561 sub sqlt_type { shift->dbh->{Driver}->{Name} }
1562
1563 =head2 bind_attribute_by_data_type
1564
1565 Given a datatype from column info, returns a database specific bind
1566 attribute for C<< $dbh->bind_param($val,$attribute) >> or nothing if we will
1567 let the database planner just handle it.
1568
1569 Generally only needed for special case column types, like bytea in postgres.
1570
1571 =cut
1572
1573 sub bind_attribute_by_data_type {
1574     return;
1575 }
1576
1577 =head2 create_ddl_dir
1578
1579 =over 4
1580
1581 =item Arguments: $schema \@databases, $version, $directory, $preversion, \%sqlt_args
1582
1583 =back
1584
1585 Creates a SQL file based on the Schema, for each of the specified
1586 database types, in the given directory.
1587
1588 By default, C<\%sqlt_args> will have
1589
1590  { add_drop_table => 1, ignore_constraint_names => 1, ignore_index_names => 1 }
1591
1592 merged with the hash passed in. To disable any of those features, pass in a
1593 hashref like the following
1594
1595  { ignore_constraint_names => 0, # ... other options }
1596
1597 =cut
1598
1599 sub create_ddl_dir {
1600   my ($self, $schema, $databases, $version, $dir, $preversion, $sqltargs) = @_;
1601
1602   if(!$dir || !-d $dir) {
1603     warn "No directory given, using ./\n";
1604     $dir = "./";
1605   }
1606   $databases ||= ['MySQL', 'SQLite', 'PostgreSQL'];
1607   $databases = [ $databases ] if(ref($databases) ne 'ARRAY');
1608
1609   my $schema_version = $schema->schema_version || '1.x';
1610   $version ||= $schema_version;
1611
1612   $sqltargs = {
1613     add_drop_table => 1,
1614     ignore_constraint_names => 1,
1615     ignore_index_names => 1,
1616     %{$sqltargs || {}}
1617   };
1618
1619   $self->throw_exception(q{Can't create a ddl file without SQL::Translator 0.09003: '}
1620       . $self->_check_sqlt_message . q{'})
1621           if !$self->_check_sqlt_version;
1622
1623   my $sqlt = SQL::Translator->new( $sqltargs );
1624
1625   $sqlt->parser('SQL::Translator::Parser::DBIx::Class');
1626   my $sqlt_schema = $sqlt->translate({ data => $schema }) or die $sqlt->error;
1627
1628   foreach my $db (@$databases) {
1629     $sqlt->reset();
1630     $sqlt = $self->configure_sqlt($sqlt, $db);
1631     $sqlt->{schema} = $sqlt_schema;
1632     $sqlt->producer($db);
1633
1634     my $file;
1635     my $filename = $schema->ddl_filename($db, $version, $dir);
1636     if (-e $filename && ($version eq $schema_version )) {
1637       # if we are dumping the current version, overwrite the DDL
1638       warn "Overwriting existing DDL file - $filename";
1639       unlink($filename);
1640     }
1641
1642     my $output = $sqlt->translate;
1643     if(!$output) {
1644       warn("Failed to translate to $db, skipping. (" . $sqlt->error . ")");
1645       next;
1646     }
1647     if(!open($file, ">$filename")) {
1648       $self->throw_exception("Can't open $filename for writing ($!)");
1649       next;
1650     }
1651     print $file $output;
1652     close($file);
1653
1654     next unless ($preversion);
1655
1656     require SQL::Translator::Diff;
1657
1658     my $prefilename = $schema->ddl_filename($db, $preversion, $dir);
1659     if(!-e $prefilename) {
1660       warn("No previous schema file found ($prefilename)");
1661       next;
1662     }
1663
1664     my $difffile = $schema->ddl_filename($db, $version, $dir, $preversion);
1665     if(-e $difffile) {
1666       warn("Overwriting existing diff file - $difffile");
1667       unlink($difffile);
1668     }
1669
1670     my $source_schema;
1671     {
1672       my $t = SQL::Translator->new($sqltargs);
1673       $t->debug( 0 );
1674       $t->trace( 0 );
1675       $t->parser( $db )                       or die $t->error;
1676       $t = $self->configure_sqlt($t, $db);
1677       my $out = $t->translate( $prefilename ) or die $t->error;
1678       $source_schema = $t->schema;
1679       unless ( $source_schema->name ) {
1680         $source_schema->name( $prefilename );
1681       }
1682     }
1683
1684     # The "new" style of producers have sane normalization and can support
1685     # diffing a SQL file against a DBIC->SQLT schema. Old style ones don't
1686     # And we have to diff parsed SQL against parsed SQL.
1687     my $dest_schema = $sqlt_schema;
1688
1689     unless ( "SQL::Translator::Producer::$db"->can('preprocess_schema') ) {
1690       my $t = SQL::Translator->new($sqltargs);
1691       $t->debug( 0 );
1692       $t->trace( 0 );
1693       $t->parser( $db )                    or die $t->error;
1694       $t = $self->configure_sqlt($t, $db);
1695       my $out = $t->translate( $filename ) or die $t->error;
1696       $dest_schema = $t->schema;
1697       $dest_schema->name( $filename )
1698         unless $dest_schema->name;
1699     }
1700
1701     my $diff = SQL::Translator::Diff::schema_diff($source_schema, $db,
1702                                                   $dest_schema,   $db,
1703                                                   $sqltargs
1704                                                  );
1705     if(!open $file, ">$difffile") {
1706       $self->throw_exception("Can't write to $difffile ($!)");
1707       next;
1708     }
1709     print $file $diff;
1710     close($file);
1711   }
1712 }
1713
1714 sub configure_sqlt() {
1715   my $self = shift;
1716   my $tr = shift;
1717   my $db = shift || $self->sqlt_type;
1718   if ($db eq 'PostgreSQL') {
1719     $tr->quote_table_names(0);
1720     $tr->quote_field_names(0);
1721   }
1722   return $tr;
1723 }
1724
1725 =head2 deployment_statements
1726
1727 =over 4
1728
1729 =item Arguments: $schema, $type, $version, $directory, $sqlt_args
1730
1731 =back
1732
1733 Returns the statements used by L</deploy> and L<DBIx::Class::Schema/deploy>.
1734 The database driver name is given by C<$type>, though the value from
1735 L</sqlt_type> is used if it is not specified.
1736
1737 C<$directory> is used to return statements from files in a previously created
1738 L</create_ddl_dir> directory and is optional. The filenames are constructed
1739 from L<DBIx::Class::Schema/ddl_filename>, the schema name and the C<$version>.
1740
1741 If no C<$directory> is specified then the statements are constructed on the
1742 fly using L<SQL::Translator> and C<$version> is ignored.
1743
1744 See L<SQL::Translator/METHODS> for a list of values for C<$sqlt_args>.
1745
1746 =cut
1747
1748 sub deployment_statements {
1749   my ($self, $schema, $type, $version, $dir, $sqltargs) = @_;
1750   # Need to be connected to get the correct sqlt_type
1751   $self->ensure_connected() unless $type;
1752   $type ||= $self->sqlt_type;
1753   $version ||= $schema->schema_version || '1.x';
1754   $dir ||= './';
1755   my $filename = $schema->ddl_filename($type, $dir, $version);
1756   if(-f $filename)
1757   {
1758       my $file;
1759       open($file, "<$filename")
1760         or $self->throw_exception("Can't open $filename ($!)");
1761       my @rows = <$file>;
1762       close($file);
1763       return join('', @rows);
1764   }
1765
1766   $self->throw_exception(q{Can't deploy without SQL::Translator 0.09003: '}
1767       . $self->_check_sqlt_message . q{'})
1768           if !$self->_check_sqlt_version;
1769
1770   require SQL::Translator::Parser::DBIx::Class;
1771   eval qq{use SQL::Translator::Producer::${type}};
1772   $self->throw_exception($@) if $@;
1773
1774   # sources needs to be a parser arg, but for simplicty allow at top level
1775   # coming in
1776   $sqltargs->{parser_args}{sources} = delete $sqltargs->{sources}
1777       if exists $sqltargs->{sources};
1778
1779   my $tr = SQL::Translator->new(%$sqltargs);
1780   SQL::Translator::Parser::DBIx::Class::parse( $tr, $schema );
1781   return "SQL::Translator::Producer::${type}"->can('produce')->($tr);
1782 }
1783
1784 sub deploy {
1785   my ($self, $schema, $type, $sqltargs, $dir) = @_;
1786   my $deploy = sub {
1787     my $line = shift;
1788     return if($line =~ /^--/);
1789     return if(!$line);
1790     # next if($line =~ /^DROP/m);
1791     return if($line =~ /^BEGIN TRANSACTION/m);
1792     return if($line =~ /^COMMIT/m);
1793     return if $line =~ /^\s+$/; # skip whitespace only
1794     $self->_query_start($line);
1795     eval {
1796       $self->dbh->do($line); # shouldn't be using ->dbh ?
1797     };
1798     if ($@) {
1799       warn qq{$@ (running "${line}")};
1800     }
1801     $self->_query_end($line);
1802   };
1803   my @statements = $self->deployment_statements($schema, $type, undef, $dir, { no_comments => 1, %{ $sqltargs || {} } } );
1804   if (@statements > 1) {
1805     foreach my $statement (@statements) {
1806       $deploy->( $statement );
1807     }
1808   }
1809   elsif (@statements == 1) {
1810     foreach my $line ( split(";\n", $statements[0])) {
1811       $deploy->( $line );
1812     }
1813   }
1814 }
1815
1816 =head2 datetime_parser
1817
1818 Returns the datetime parser class
1819
1820 =cut
1821
1822 sub datetime_parser {
1823   my $self = shift;
1824   return $self->{datetime_parser} ||= do {
1825     $self->ensure_connected;
1826     $self->build_datetime_parser(@_);
1827   };
1828 }
1829
1830 =head2 datetime_parser_type
1831
1832 Defines (returns) the datetime parser class - currently hardwired to
1833 L<DateTime::Format::MySQL>
1834
1835 =cut
1836
1837 sub datetime_parser_type { "DateTime::Format::MySQL"; }
1838
1839 =head2 build_datetime_parser
1840
1841 See L</datetime_parser>
1842
1843 =cut
1844
1845 sub build_datetime_parser {
1846   my $self = shift;
1847   my $type = $self->datetime_parser_type(@_);
1848   eval "use ${type}";
1849   $self->throw_exception("Couldn't load ${type}: $@") if $@;
1850   return $type;
1851 }
1852
1853 {
1854     my $_check_sqlt_version; # private
1855     my $_check_sqlt_message; # private
1856     sub _check_sqlt_version {
1857         return $_check_sqlt_version if defined $_check_sqlt_version;
1858         eval 'use SQL::Translator "0.09003"';
1859         $_check_sqlt_message = $@ || '';
1860         $_check_sqlt_version = !$@;
1861     }
1862
1863     sub _check_sqlt_message {
1864         _check_sqlt_version if !defined $_check_sqlt_message;
1865         $_check_sqlt_message;
1866     }
1867 }
1868
1869 =head2 is_replicating
1870
1871 A boolean that reports if a particular L<DBIx::Class::Storage::DBI> is set to
1872 replicate from a master database.  Default is undef, which is the result
1873 returned by databases that don't support replication.
1874
1875 =cut
1876
1877 sub is_replicating {
1878     return;
1879
1880 }
1881
1882 =head2 lag_behind_master
1883
1884 Returns a number that represents a certain amount of lag behind a master db
1885 when a given storage is replicating.  The number is database dependent, but
1886 starts at zero and increases with the amount of lag. Default in undef
1887
1888 =cut
1889
1890 sub lag_behind_master {
1891     return;
1892 }
1893
1894 sub DESTROY {
1895   my $self = shift;
1896   return if !$self->_dbh;
1897   $self->_verify_pid;
1898   $self->_dbh(undef);
1899 }
1900
1901 1;
1902
1903 =head1 USAGE NOTES
1904
1905 =head2 DBIx::Class and AutoCommit
1906
1907 DBIx::Class can do some wonderful magic with handling exceptions,
1908 disconnections, and transactions when you use C<< AutoCommit => 1 >>
1909 combined with C<txn_do> for transaction support.
1910
1911 If you set C<< AutoCommit => 0 >> in your connect info, then you are always
1912 in an assumed transaction between commits, and you're telling us you'd
1913 like to manage that manually.  A lot of the magic protections offered by
1914 this module will go away.  We can't protect you from exceptions due to database
1915 disconnects because we don't know anything about how to restart your
1916 transactions.  You're on your own for handling all sorts of exceptional
1917 cases if you choose the C<< AutoCommit => 0 >> path, just as you would
1918 be with raw DBI.
1919
1920
1921 =head1 SQL METHODS
1922
1923 The module defines a set of methods within the DBIC::SQL::Abstract
1924 namespace.  These build on L<SQL::Abstract::Limit> to provide the
1925 SQL query functions.
1926
1927 The following methods are extended:-
1928
1929 =over 4
1930
1931 =item delete
1932
1933 =item insert
1934
1935 =item select
1936
1937 =item update
1938
1939 =item limit_dialect
1940
1941 See L</connect_info> for details.
1942
1943 =item quote_char
1944
1945 See L</connect_info> for details.
1946
1947 =item name_sep
1948
1949 See L</connect_info> for details.
1950
1951 =back
1952
1953 =head1 AUTHORS
1954
1955 Matt S. Trout <mst@shadowcatsystems.co.uk>
1956
1957 Andy Grundman <andy@hybridized.org>
1958
1959 =head1 LICENSE
1960
1961 You may distribute this code under the same terms as Perl itself.
1962
1963 =cut