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