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