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     my @to_run = (ref $action eq 'ARRAY') ? (@$action) : ($action);
 956     $self->_query_start(@to_run);
 957     $self->_dbh->do(@to_run);
 958     $self->_query_end(@to_run);
 959   }
 960
 961   return $self;
 962 }
 963
 964 sub _connect {
 965   my ($self, @info) = @_;
 966
 967   $self->throw_exception("You failed to provide any connection info")
 968     if !@info;
 969
 970   my ($old_connect_via, $dbh);
 971
 972   if ($INC{'Apache/DBI.pm'} && $ENV{MOD_PERL}) {
 973     $old_connect_via = $DBI::connect_via;
 974     $DBI::connect_via = 'connect';
 975   }
 976
 977   eval {
 978     if(ref $info[0] eq 'CODE') {
 979        $dbh = &{$info[0]}
 980     }
 981     else {
 982        $dbh = DBI->connect(@info);
 983     }
 984
 985     if($dbh && !$self->unsafe) {
 986       my $weak_self = $self;
 987       weaken($weak_self);
 988       $dbh->{HandleError} = sub {
 989           if ($weak_self) {
 990             $weak_self->throw_exception("DBI Exception: $_[0]");
 991           }
 992           else {
 993             croak ("DBI Exception: $_[0]");
 994           }
 995       };
 996       $dbh->{ShowErrorStatement} = 1;
 997       $dbh->{RaiseError} = 1;
 998       $dbh->{PrintError} = 0;
 999     }
1000   };
1001
1002   $DBI::connect_via = $old_connect_via if $old_connect_via;
1003
1004   $self->throw_exception("DBI Connection failed: " . ($@||$DBI::errstr))
1005     if !$dbh || $@;
1006
1007   $self->_dbh_autocommit($dbh->{AutoCommit});
1008
1009   $dbh;
1010 }
1011
1012 sub svp_begin {
1013   my ($self, $name) = @_;
1014
1015   $name = $self->_svp_generate_name
1016     unless defined $name;
1017
1018   $self->throw_exception ("You can't use savepoints outside a transaction")
1019     if $self->{transaction_depth} == 0;
1020
1021   $self->throw_exception ("Your Storage implementation doesn't support savepoints")
1022     unless $self->can('_svp_begin');
1023
1024   push @{ $self->{savepoints} }, $name;
1025
1026   $self->debugobj->svp_begin($name) if $self->debug;
1027
1028   return $self->_svp_begin($name);
1029 }
1030
1031 sub svp_release {
1032   my ($self, $name) = @_;
1033
1034   $self->throw_exception ("You can't use savepoints outside a transaction")
1035     if $self->{transaction_depth} == 0;
1036
1037   $self->throw_exception ("Your Storage implementation doesn't support savepoints")
1038     unless $self->can('_svp_release');
1039
1040   if (defined $name) {
1041     $self->throw_exception ("Savepoint '$name' does not exist")
1042       unless grep { $_ eq $name } @{ $self->{savepoints} };
1043
1044     # Dig through the stack until we find the one we are releasing.  This keeps
1045     # the stack up to date.
1046     my $svp;
1047
1048     do { $svp = pop @{ $self->{savepoints} } } while $svp ne $name;
1049   } else {
1050     $name = pop @{ $self->{savepoints} };
1051   }
1052
1053   $self->debugobj->svp_release($name) if $self->debug;
1054
1055   return $self->_svp_release($name);
1056 }
1057
1058 sub svp_rollback {
1059   my ($self, $name) = @_;
1060
1061   $self->throw_exception ("You can't use savepoints outside a transaction")
1062     if $self->{transaction_depth} == 0;
1063
1064   $self->throw_exception ("Your Storage implementation doesn't support savepoints")
1065     unless $self->can('_svp_rollback');
1066
1067   if (defined $name) {
1068       # If they passed us a name, verify that it exists in the stack
1069       unless(grep({ $_ eq $name } @{ $self->{savepoints} })) {
1070           $self->throw_exception("Savepoint '$name' does not exist!");
1071       }
1072
1073       # Dig through the stack until we find the one we are releasing.  This keeps
1074       # the stack up to date.
1075       while(my $s = pop(@{ $self->{savepoints} })) {
1076           last if($s eq $name);
1077       }
1078       # Add the savepoint back to the stack, as a rollback doesn't remove the
1079       # named savepoint, only everything after it.
1080       push(@{ $self->{savepoints} }, $name);
1081   } else {
1082       # We'll assume they want to rollback to the last savepoint
1083       $name = $self->{savepoints}->[-1];
1084   }
1085
1086   $self->debugobj->svp_rollback($name) if $self->debug;
1087
1088   return $self->_svp_rollback($name);
1089 }
1090
1091 sub _svp_generate_name {
1092     my ($self) = @_;
1093
1094     return 'savepoint_'.scalar(@{ $self->{'savepoints'} });
1095 }
1096
1097 sub txn_begin {
1098   my $self = shift;
1099   $self->ensure_connected();
1100   if($self->{transaction_depth} == 0) {
1101     $self->debugobj->txn_begin()
1102       if $self->debug;
1103     # this isn't ->_dbh-> because
1104     #  we should reconnect on begin_work
1105     #  for AutoCommit users
1106     $self->dbh->begin_work;
1107   } elsif ($self->auto_savepoint) {
1108     $self->svp_begin;
1109   }
1110   $self->{transaction_depth}++;
1111 }
1112
1113 sub txn_commit {
1114   my $self = shift;
1115   if ($self->{transaction_depth} == 1) {
1116     my $dbh = $self->_dbh;
1117     $self->debugobj->txn_commit()
1118       if ($self->debug);
1119     $dbh->commit;
1120     $self->{transaction_depth} = 0
1121       if $self->_dbh_autocommit;
1122   }
1123   elsif($self->{transaction_depth} > 1) {
1124     $self->{transaction_depth}--;
1125     $self->svp_release
1126       if $self->auto_savepoint;
1127   }
1128 }
1129
1130 sub txn_rollback {
1131   my $self = shift;
1132   my $dbh = $self->_dbh;
1133   eval {
1134     if ($self->{transaction_depth} == 1) {
1135       $self->debugobj->txn_rollback()
1136         if ($self->debug);
1137       $self->{transaction_depth} = 0
1138         if $self->_dbh_autocommit;
1139       $dbh->rollback;
1140     }
1141     elsif($self->{transaction_depth} > 1) {
1142       $self->{transaction_depth}--;
1143       if ($self->auto_savepoint) {
1144         $self->svp_rollback;
1145         $self->svp_release;
1146       }
1147     }
1148     else {
1149       die DBIx::Class::Storage::NESTED_ROLLBACK_EXCEPTION->new;
1150     }
1151   };
1152   if ($@) {
1153     my $error = $@;
1154     my $exception_class = "DBIx::Class::Storage::NESTED_ROLLBACK_EXCEPTION";
1155     $error =~ /$exception_class/ and $self->throw_exception($error);
1156     # ensure that a failed rollback resets the transaction depth
1157     $self->{transaction_depth} = $self->_dbh_autocommit ? 0 : 1;
1158     $self->throw_exception($error);
1159   }
1160 }
1161
1162 # This used to be the top-half of _execute.  It was split out to make it
1163 #  easier to override in NoBindVars without duping the rest.  It takes up
1164 #  all of _execute's args, and emits $sql, @bind.
1165 sub _prep_for_execute {
1166   my ($self, $op, $extra_bind, $ident, $args) = @_;
1167
1168   my ($sql, @bind) = $self->sql_maker->$op($ident, @$args);
1169   unshift(@bind,
1170     map { ref $_ eq 'ARRAY' ? $_ : [ '!!dummy', $_ ] } @$extra_bind)
1171       if $extra_bind;
1172
1173   return ($sql, \@bind);
1174 }
1175
1176 sub _fix_bind_params {
1177     my ($self, @bind) = @_;
1178
1179     ### Turn @bind from something like this:
1180     ###   ( [ "artist", 1 ], [ "cdid", 1, 3 ] )
1181     ### to this:
1182     ###   ( "'1'", "'1'", "'3'" )
1183     return
1184         map {
1185             if ( defined( $_ && $_->[1] ) ) {
1186                 map { qq{'$_'}; } @{$_}[ 1 .. $#$_ ];
1187             }
1188             else { q{'NULL'}; }
1189         } @bind;
1190 }
1191
1192 sub _query_start {
1193     my ( $self, $sql, @bind ) = @_;
1194
1195     if ( $self->debug ) {
1196         @bind = $self->_fix_bind_params(@bind);
1197
1198         $self->debugobj->query_start( $sql, @bind );
1199     }
1200 }
1201
1202 sub _query_end {
1203     my ( $self, $sql, @bind ) = @_;
1204
1205     if ( $self->debug ) {
1206         @bind = $self->_fix_bind_params(@bind);
1207         $self->debugobj->query_end( $sql, @bind );
1208     }
1209 }
1210
1211 sub _dbh_execute {
1212   my ($self, $dbh, $op, $extra_bind, $ident, $bind_attributes, @args) = @_;
1213
1214   if( blessed($ident) && $ident->isa("DBIx::Class::ResultSource") ) {
1215     $ident = $ident->from();
1216   }
1217
1218   my ($sql, $bind) = $self->_prep_for_execute($op, $extra_bind, $ident, \@args);
1219
1220   $self->_query_start( $sql, @$bind );
1221
1222   my $sth = $self->sth($sql,$op);
1223
1224   my $placeholder_index = 1;
1225
1226   foreach my $bound (@$bind) {
1227     my $attributes = {};
1228     my($column_name, @data) = @$bound;
1229
1230     if ($bind_attributes) {
1231       $attributes = $bind_attributes->{$column_name}
1232       if defined $bind_attributes->{$column_name};
1233     }
1234
1235     foreach my $data (@data) {
1236       my $ref = ref $data;
1237       $data = $ref && $ref ne 'ARRAY' ? ''.$data : $data; # stringify args (except arrayrefs)
1238
1239       $sth->bind_param($placeholder_index, $data, $attributes);
1240       $placeholder_index++;
1241     }
1242   }
1243
1244   # Can this fail without throwing an exception anyways???
1245   my $rv = $sth->execute();
1246   $self->throw_exception($sth->errstr) if !$rv;
1247
1248   $self->_query_end( $sql, @$bind );
1249
1250   return (wantarray ? ($rv, $sth, @$bind) : $rv);
1251 }
1252
1253 sub _execute {
1254     my $self = shift;
1255     $self->dbh_do('_dbh_execute', @_)
1256 }
1257
1258 sub insert {
1259   my ($self, $source, $to_insert) = @_;
1260
1261   my $ident = $source->from;
1262   my $bind_attributes = $self->source_bind_attributes($source);
1263
1264   $self->ensure_connected;
1265   foreach my $col ( $source->columns ) {
1266     if ( !defined $to_insert->{$col} ) {
1267       my $col_info = $source->column_info($col);
1268
1269       if ( $col_info->{auto_nextval} ) {
1270         $to_insert->{$col} = $self->_sequence_fetch( 'nextval', $col_info->{sequence} || $self->_dbh_get_autoinc_seq($self->dbh, $source) );
1271       }
1272     }
1273   }
1274
1275   $self->_execute('insert' => [], $source, $bind_attributes, $to_insert);
1276
1277   return $to_insert;
1278 }
1279
1280 ## Still not quite perfect, and EXPERIMENTAL
1281 ## Currently it is assumed that all values passed will be "normal", i.e. not
1282 ## scalar refs, or at least, all the same type as the first set, the statement is
1283 ## only prepped once.
1284 sub insert_bulk {
1285   my ($self, $source, $cols, $data) = @_;
1286   my %colvalues;
1287   my $table = $source->from;
1288   @colvalues{@$cols} = (0..$#$cols);
1289   my ($sql, @bind) = $self->sql_maker->insert($table, \%colvalues);
1290
1291   $self->_query_start( $sql, @bind );
1292   my $sth = $self->sth($sql);
1293
1294 #  @bind = map { ref $_ ? ''.$_ : $_ } @bind; # stringify args
1295
1296   ## This must be an arrayref, else nothing works!
1297
1298   my $tuple_status = [];
1299
1300   ##use Data::Dumper;
1301   ##print STDERR Dumper( $data, $sql, [@bind] );
1302
1303   my $time = time();
1304
1305   ## Get the bind_attributes, if any exist
1306   my $bind_attributes = $self->source_bind_attributes($source);
1307
1308   ## Bind the values and execute
1309   my $placeholder_index = 1;
1310
1311   foreach my $bound (@bind) {
1312
1313     my $attributes = {};
1314     my ($column_name, $data_index) = @$bound;
1315
1316     if( $bind_attributes ) {
1317       $attributes = $bind_attributes->{$column_name}
1318       if defined $bind_attributes->{$column_name};
1319     }
1320
1321     my @data = map { $_->[$data_index] } @$data;
1322
1323     $sth->bind_param_array( $placeholder_index, [@data], $attributes );
1324     $placeholder_index++;
1325   }
1326   my $rv = $sth->execute_array({ArrayTupleStatus => $tuple_status});
1327   $self->throw_exception($sth->errstr) if !$rv;
1328
1329   $self->_query_end( $sql, @bind );
1330   return (wantarray ? ($rv, $sth, @bind) : $rv);
1331 }
1332
1333 sub update {
1334   my $self = shift @_;
1335   my $source = shift @_;
1336   my $bind_attributes = $self->source_bind_attributes($source);
1337
1338   return $self->_execute('update' => [], $source, $bind_attributes, @_);
1339 }
1340
1341
1342 sub delete {
1343   my $self = shift @_;
1344   my $source = shift @_;
1345
1346   my $bind_attrs = {}; ## If ever it's needed...
1347
1348   return $self->_execute('delete' => [], $source, $bind_attrs, @_);
1349 }
1350
1351 sub _select {
1352   my ($self, $ident, $select, $condition, $attrs) = @_;
1353   my $order = $attrs->{order_by};
1354
1355   if (ref $condition eq 'SCALAR') {
1356     my $unwrap = ${$condition};
1357     if ($unwrap =~ s/ORDER BY (.*)$//i) {
1358       $order = $1;
1359       $condition = \$unwrap;
1360     }
1361   }
1362
1363   my $for = delete $attrs->{for};
1364   my $sql_maker = $self->sql_maker;
1365   local $sql_maker->{for} = $for;
1366
1367   if (exists $attrs->{group_by} || $attrs->{having}) {
1368     $order = {
1369       group_by => $attrs->{group_by},
1370       having => $attrs->{having},
1371       ($order ? (order_by => $order) : ())
1372     };
1373   }
1374   my $bind_attrs = {}; ## Future support
1375   my @args = ('select', $attrs->{bind}, $ident, $bind_attrs, $select, $condition, $order);
1376   if ($attrs->{software_limit} ||
1377       $self->sql_maker->_default_limit_syntax eq "GenericSubQ") {
1378         $attrs->{software_limit} = 1;
1379   } else {
1380     $self->throw_exception("rows attribute must be positive if present")
1381       if (defined($attrs->{rows}) && !($attrs->{rows} > 0));
1382
1383     # MySQL actually recommends this approach.  I cringe.
1384     $attrs->{rows} = 2**48 if not defined $attrs->{rows} and defined $attrs->{offset};
1385     push @args, $attrs->{rows}, $attrs->{offset};
1386   }
1387
1388   return $self->_execute(@args);
1389 }
1390
1391 sub source_bind_attributes {
1392   my ($self, $source) = @_;
1393
1394   my $bind_attributes;
1395   foreach my $column ($source->columns) {
1396
1397     my $data_type = $source->column_info($column)->{data_type} || '';
1398     $bind_attributes->{$column} = $self->bind_attribute_by_data_type($data_type)
1399      if $data_type;
1400   }
1401
1402   return $bind_attributes;
1403 }
1404
1405 =head2 select
1406
1407 =over 4
1408
1409 =item Arguments: $ident, $select, $condition, $attrs
1410
1411 =back
1412
1413 Handle a SQL select statement.
1414
1415 =cut
1416
1417 sub select {
1418   my $self = shift;
1419   my ($ident, $select, $condition, $attrs) = @_;
1420   return $self->cursor_class->new($self, \@_, $attrs);
1421 }
1422
1423 sub select_single {
1424   my $self = shift;
1425   my ($rv, $sth, @bind) = $self->_select(@_);
1426   my @row = $sth->fetchrow_array;
1427   my @nextrow = $sth->fetchrow_array if @row;
1428   if(@row && @nextrow) {
1429     carp "Query returned more than one row.  SQL that returns multiple rows is DEPRECATED for ->find and ->single";
1430   }
1431   # Need to call finish() to work round broken DBDs
1432   $sth->finish();
1433   return @row;
1434 }
1435
1436 =head2 sth
1437
1438 =over 4
1439
1440 =item Arguments: $sql
1441
1442 =back
1443
1444 Returns a L<DBI> sth (statement handle) for the supplied SQL.
1445
1446 =cut
1447
1448 sub _dbh_sth {
1449   my ($self, $dbh, $sql) = @_;
1450
1451   # 3 is the if_active parameter which avoids active sth re-use
1452   my $sth = $self->disable_sth_caching
1453     ? $dbh->prepare($sql)
1454     : $dbh->prepare_cached($sql, {}, 3);
1455
1456   # XXX You would think RaiseError would make this impossible,
1457   #  but apparently that's not true :(
1458   $self->throw_exception($dbh->errstr) if !$sth;
1459
1460   $sth;
1461 }
1462
1463 sub sth {
1464   my ($self, $sql) = @_;
1465   $self->dbh_do('_dbh_sth', $sql);
1466 }
1467
1468 sub _dbh_columns_info_for {
1469   my ($self, $dbh, $table) = @_;
1470
1471   if ($dbh->can('column_info')) {
1472     my %result;
1473     eval {
1474       my ($schema,$tab) = $table =~ /^(.+?)\.(.+)$/ ? ($1,$2) : (undef,$table);
1475       my $sth = $dbh->column_info( undef,$schema, $tab, '%' );
1476       $sth->execute();
1477       while ( my $info = $sth->fetchrow_hashref() ){
1478         my %column_info;
1479         $column_info{data_type}   = $info->{TYPE_NAME};
1480         $column_info{size}      = $info->{COLUMN_SIZE};
1481         $column_info{is_nullable}   = $info->{NULLABLE} ? 1 : 0;
1482         $column_info{default_value} = $info->{COLUMN_DEF};
1483         my $col_name = $info->{COLUMN_NAME};
1484         $col_name =~ s/^\"(.*)\"$/$1/;
1485
1486         $result{$col_name} = \%column_info;
1487       }
1488     };
1489     return \%result if !$@ && scalar keys %result;
1490   }
1491
1492   my %result;
1493   my $sth = $dbh->prepare($self->sql_maker->select($table, undef, \'1 = 0'));
1494   $sth->execute;
1495   my @columns = @{$sth->{NAME_lc}};
1496   for my $i ( 0 .. $#columns ){
1497     my %column_info;
1498     $column_info{data_type} = $sth->{TYPE}->[$i];
1499     $column_info{size} = $sth->{PRECISION}->[$i];
1500     $column_info{is_nullable} = $sth->{NULLABLE}->[$i] ? 1 : 0;
1501
1502     if ($column_info{data_type} =~ m/^(.*?)\((.*?)\)$/) {
1503       $column_info{data_type} = $1;
1504       $column_info{size}    = $2;
1505     }
1506
1507     $result{$columns[$i]} = \%column_info;
1508   }
1509   $sth->finish;
1510
1511   foreach my $col (keys %result) {
1512     my $colinfo = $result{$col};
1513     my $type_num = $colinfo->{data_type};
1514     my $type_name;
1515     if(defined $type_num && $dbh->can('type_info')) {
1516       my $type_info = $dbh->type_info($type_num);
1517       $type_name = $type_info->{TYPE_NAME} if $type_info;
1518       $colinfo->{data_type} = $type_name if $type_name;
1519     }
1520   }
1521
1522   return \%result;
1523 }
1524
1525 sub columns_info_for {
1526   my ($self, $table) = @_;
1527   $self->dbh_do('_dbh_columns_info_for', $table);
1528 }
1529
1530 =head2 last_insert_id
1531
1532 Return the row id of the last insert.
1533
1534 =cut
1535
1536 sub _dbh_last_insert_id {
1537     my ($self, $dbh, $source, $col) = @_;
1538     # XXX This is a SQLite-ism as a default... is there a DBI-generic way?
1539     $dbh->func('last_insert_rowid');
1540 }
1541
1542 sub last_insert_id {
1543   my $self = shift;
1544   $self->dbh_do('_dbh_last_insert_id', @_);
1545 }
1546
1547 =head2 sqlt_type
1548
1549 Returns the database driver name.
1550
1551 =cut
1552
1553 sub sqlt_type { shift->dbh->{Driver}->{Name} }
1554
1555 =head2 bind_attribute_by_data_type
1556
1557 Given a datatype from column info, returns a database specific bind
1558 attribute for C<< $dbh->bind_param($val,$attribute) >> or nothing if we will
1559 let the database planner just handle it.
1560
1561 Generally only needed for special case column types, like bytea in postgres.
1562
1563 =cut
1564
1565 sub bind_attribute_by_data_type {
1566     return;
1567 }
1568
1569 =head2 create_ddl_dir
1570
1571 =over 4
1572
1573 =item Arguments: $schema \@databases, $version, $directory, $preversion, \%sqlt_args
1574
1575 =back
1576
1577 Creates a SQL file based on the Schema, for each of the specified
1578 database types, in the given directory.
1579
1580 By default, C<\%sqlt_args> will have
1581
1582  { add_drop_table => 1, ignore_constraint_names => 1, ignore_index_names => 1 }
1583
1584 merged with the hash passed in. To disable any of those features, pass in a
1585 hashref like the following
1586
1587  { ignore_constraint_names => 0, # ... other options }
1588
1589 =cut
1590
1591 sub create_ddl_dir {
1592   my ($self, $schema, $databases, $version, $dir, $preversion, $sqltargs) = @_;
1593
1594   if(!$dir || !-d $dir) {
1595     warn "No directory given, using ./\n";
1596     $dir = "./";
1597   }
1598   $databases ||= ['MySQL', 'SQLite', 'PostgreSQL'];
1599   $databases = [ $databases ] if(ref($databases) ne 'ARRAY');
1600
1601   my $schema_version = $schema->schema_version || '1.x';
1602   $version ||= $schema_version;
1603
1604   $sqltargs = {
1605     add_drop_table => 1,
1606     ignore_constraint_names => 1,
1607     ignore_index_names => 1,
1608     %{$sqltargs || {}}
1609   };
1610
1611   $self->throw_exception(q{Can't create a ddl file without SQL::Translator 0.09: '}
1612       . $self->_check_sqlt_message . q{'})
1613           if !$self->_check_sqlt_version;
1614
1615   my $sqlt = SQL::Translator->new( $sqltargs );
1616
1617   $sqlt->parser('SQL::Translator::Parser::DBIx::Class');
1618   my $sqlt_schema = $sqlt->translate({ data => $schema }) or die $sqlt->error;
1619
1620   foreach my $db (@$databases) {
1621     $sqlt->reset();
1622     $sqlt = $self->configure_sqlt($sqlt, $db);
1623     $sqlt->{schema} = $sqlt_schema;
1624     $sqlt->producer($db);
1625
1626     my $file;
1627     my $filename = $schema->ddl_filename($db, $version, $dir);
1628     if (-e $filename && ($version eq $schema_version )) {
1629       # if we are dumping the current version, overwrite the DDL
1630       warn "Overwriting existing DDL file - $filename";
1631       unlink($filename);
1632     }
1633
1634     my $output = $sqlt->translate;
1635     if(!$output) {
1636       warn("Failed to translate to $db, skipping. (" . $sqlt->error . ")");
1637       next;
1638     }
1639     if(!open($file, ">$filename")) {
1640       $self->throw_exception("Can't open $filename for writing ($!)");
1641       next;
1642     }
1643     print $file $output;
1644     close($file);
1645
1646     next unless ($preversion);
1647
1648     require SQL::Translator::Diff;
1649
1650     my $prefilename = $schema->ddl_filename($db, $preversion, $dir);
1651     if(!-e $prefilename) {
1652       warn("No previous schema file found ($prefilename)");
1653       next;
1654     }
1655
1656     my $difffile = $schema->ddl_filename($db, $version, $dir, $preversion);
1657     if(-e $difffile) {
1658       warn("Overwriting existing diff file - $difffile");
1659       unlink($difffile);
1660     }
1661
1662     my $source_schema;
1663     {
1664       my $t = SQL::Translator->new($sqltargs);
1665       $t->debug( 0 );
1666       $t->trace( 0 );
1667       $t->parser( $db )                       or die $t->error;
1668       $t = $self->configure_sqlt($t, $db);
1669       my $out = $t->translate( $prefilename ) or die $t->error;
1670       $source_schema = $t->schema;
1671       unless ( $source_schema->name ) {
1672         $source_schema->name( $prefilename );
1673       }
1674     }
1675
1676     # The "new" style of producers have sane normalization and can support
1677     # diffing a SQL file against a DBIC->SQLT schema. Old style ones don't
1678     # And we have to diff parsed SQL against parsed SQL.
1679     my $dest_schema = $sqlt_schema;
1680
1681     unless ( "SQL::Translator::Producer::$db"->can('preprocess_schema') ) {
1682       my $t = SQL::Translator->new($sqltargs);
1683       $t->debug( 0 );
1684       $t->trace( 0 );
1685       $t->parser( $db )                    or die $t->error;
1686       $t = $self->configure_sqlt($t, $db);
1687       my $out = $t->translate( $filename ) or die $t->error;
1688       $dest_schema = $t->schema;
1689       $dest_schema->name( $filename )
1690         unless $dest_schema->name;
1691     }
1692
1693     my $diff = SQL::Translator::Diff::schema_diff($source_schema, $db,
1694                                                   $dest_schema,   $db,
1695                                                   $sqltargs
1696                                                  );
1697     if(!open $file, ">$difffile") {
1698       $self->throw_exception("Can't write to $difffile ($!)");
1699       next;
1700     }
1701     print $file $diff;
1702     close($file);
1703   }
1704 }
1705
1706 sub configure_sqlt() {
1707   my $self = shift;
1708   my $tr = shift;
1709   my $db = shift || $self->sqlt_type;
1710   if ($db eq 'PostgreSQL') {
1711     $tr->quote_table_names(0);
1712     $tr->quote_field_names(0);
1713   }
1714   return $tr;
1715 }
1716
1717 =head2 deployment_statements
1718
1719 =over 4
1720
1721 =item Arguments: $schema, $type, $version, $directory, $sqlt_args
1722
1723 =back
1724
1725 Returns the statements used by L</deploy> and L<DBIx::Class::Schema/deploy>.
1726 The database driver name is given by C<$type>, though the value from
1727 L</sqlt_type> is used if it is not specified.
1728
1729 C<$directory> is used to return statements from files in a previously created
1730 L</create_ddl_dir> directory and is optional. The filenames are constructed
1731 from L<DBIx::Class::Schema/ddl_filename>, the schema name and the C<$version>.
1732
1733 If no C<$directory> is specified then the statements are constructed on the
1734 fly using L<SQL::Translator> and C<$version> is ignored.
1735
1736 See L<SQL::Translator/METHODS> for a list of values for C<$sqlt_args>.
1737
1738 =cut
1739
1740 sub deployment_statements {
1741   my ($self, $schema, $type, $version, $dir, $sqltargs) = @_;
1742   # Need to be connected to get the correct sqlt_type
1743   $self->ensure_connected() unless $type;
1744   $type ||= $self->sqlt_type;
1745   $version ||= $schema->schema_version || '1.x';
1746   $dir ||= './';
1747   my $filename = $schema->ddl_filename($type, $dir, $version);
1748   if(-f $filename)
1749   {
1750       my $file;
1751       open($file, "<$filename")
1752         or $self->throw_exception("Can't open $filename ($!)");
1753       my @rows = <$file>;
1754       close($file);
1755       return join('', @rows);
1756   }
1757
1758   $self->throw_exception(q{Can't deploy without SQL::Translator 0.09: '}
1759       . $self->_check_sqlt_message . q{'})
1760           if !$self->_check_sqlt_version;
1761
1762   require SQL::Translator::Parser::DBIx::Class;
1763   eval qq{use SQL::Translator::Producer::${type}};
1764   $self->throw_exception($@) if $@;
1765
1766   # sources needs to be a parser arg, but for simplicty allow at top level
1767   # coming in
1768   $sqltargs->{parser_args}{sources} = delete $sqltargs->{sources}
1769       if exists $sqltargs->{sources};
1770
1771   my $tr = SQL::Translator->new(%$sqltargs);
1772   SQL::Translator::Parser::DBIx::Class::parse( $tr, $schema );
1773   return "SQL::Translator::Producer::${type}"->can('produce')->($tr);
1774 }
1775
1776 sub deploy {
1777   my ($self, $schema, $type, $sqltargs, $dir) = @_;
1778   foreach my $statement ( $self->deployment_statements($schema, $type, undef, $dir, { no_comments => 1, %{ $sqltargs || {} } } ) ) {
1779     foreach my $line ( split(";\n", $statement)) {
1780       next if($line =~ /^--/);
1781       next if(!$line);
1782 #      next if($line =~ /^DROP/m);
1783       next if($line =~ /^BEGIN TRANSACTION/m);
1784       next if($line =~ /^COMMIT/m);
1785       next if $line =~ /^\s+$/; # skip whitespace only
1786       $self->_query_start($line);
1787       eval {
1788         $self->dbh->do($line); # shouldn't be using ->dbh ?
1789       };
1790       if ($@) {
1791         warn qq{$@ (running "${line}")};
1792       }
1793       $self->_query_end($line);
1794     }
1795   }
1796 }
1797
1798 =head2 datetime_parser
1799
1800 Returns the datetime parser class
1801
1802 =cut
1803
1804 sub datetime_parser {
1805   my $self = shift;
1806   return $self->{datetime_parser} ||= do {
1807     $self->ensure_connected;
1808     $self->build_datetime_parser(@_);
1809   };
1810 }
1811
1812 =head2 datetime_parser_type
1813
1814 Defines (returns) the datetime parser class - currently hardwired to
1815 L<DateTime::Format::MySQL>
1816
1817 =cut
1818
1819 sub datetime_parser_type { "DateTime::Format::MySQL"; }
1820
1821 =head2 build_datetime_parser
1822
1823 See L</datetime_parser>
1824
1825 =cut
1826
1827 sub build_datetime_parser {
1828   my $self = shift;
1829   my $type = $self->datetime_parser_type(@_);
1830   eval "use ${type}";
1831   $self->throw_exception("Couldn't load ${type}: $@") if $@;
1832   return $type;
1833 }
1834
1835 {
1836     my $_check_sqlt_version; # private
1837     my $_check_sqlt_message; # private
1838     sub _check_sqlt_version {
1839         return $_check_sqlt_version if defined $_check_sqlt_version;
1840         eval 'use SQL::Translator "0.09"';
1841         $_check_sqlt_message = $@ || '';
1842         $_check_sqlt_version = !$@;
1843     }
1844
1845     sub _check_sqlt_message {
1846         _check_sqlt_version if !defined $_check_sqlt_message;
1847         $_check_sqlt_message;
1848     }
1849 }
1850
1851 =head2 is_replicating
1852
1853 A boolean that reports if a particular L<DBIx::Class::Storage::DBI> is set to
1854 replicate from a master database.  Default is undef, which is the result
1855 returned by databases that don't support replication.
1856
1857 =cut
1858
1859 sub is_replicating {
1860     return;
1861
1862 }
1863
1864 =head2 lag_behind_master
1865
1866 Returns a number that represents a certain amount of lag behind a master db
1867 when a given storage is replicating.  The number is database dependent, but
1868 starts at zero and increases with the amount of lag. Default in undef
1869
1870 =cut
1871
1872 sub lag_behind_master {
1873     return;
1874 }
1875
1876 sub DESTROY {
1877   my $self = shift;
1878   return if !$self->_dbh;
1879   $self->_verify_pid;
1880   $self->_dbh(undef);
1881 }
1882
1883 1;
1884
1885 =head1 USAGE NOTES
1886
1887 =head2 DBIx::Class and AutoCommit
1888
1889 DBIx::Class can do some wonderful magic with handling exceptions,
1890 disconnections, and transactions when you use C<< AutoCommit => 1 >>
1891 combined with C<txn_do> for transaction support.
1892
1893 If you set C<< AutoCommit => 0 >> in your connect info, then you are always
1894 in an assumed transaction between commits, and you're telling us you'd
1895 like to manage that manually.  A lot of the magic protections offered by
1896 this module will go away.  We can't protect you from exceptions due to database
1897 disconnects because we don't know anything about how to restart your
1898 transactions.  You're on your own for handling all sorts of exceptional
1899 cases if you choose the C<< AutoCommit => 0 >> path, just as you would
1900 be with raw DBI.
1901
1902
1903 =head1 SQL METHODS
1904
1905 The module defines a set of methods within the DBIC::SQL::Abstract
1906 namespace.  These build on L<SQL::Abstract::Limit> to provide the
1907 SQL query functions.
1908
1909 The following methods are extended:-
1910
1911 =over 4
1912
1913 =item delete
1914
1915 =item insert
1916
1917 =item select
1918
1919 =item update
1920
1921 =item limit_dialect
1922
1923 See L</connect_info> for details.
1924
1925 =item quote_char
1926
1927 See L</connect_info> for details.
1928
1929 =item name_sep
1930
1931 See L</connect_info> for details.
1932
1933 =back
1934
1935 =head1 AUTHORS
1936
1937 Matt S. Trout <mst@shadowcatsystems.co.uk>
1938
1939 Andy Grundman <andy@hybridized.org>
1940
1941 =head1 LICENSE
1942
1943 You may distribute this code under the same terms as Perl itself.
1944
1945 =cut