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