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