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   $self->_determine_driver;
 618
 619   # Always set the transaction depth on connect, since
 620   #  there is no transaction in progress by definition
 621   $self->{transaction_depth} = $self->_dbh_autocommit ? 0 : 1;
 622
 623   $self->_conn_pid($$);
 624   $self->_conn_tid(threads->tid) if $INC{'threads.pm'};
 625
 626   my $connection_do = $self->on_connect_do;
 627   $self->_do_connection_actions($connection_do) if $connection_do;
 628 }
 629
 630 sub _determine_driver {
 631   my ($self) = @_;
 632
 633   if (ref $self eq 'DBIx::Class::Storage::DBI') {
 634     my $driver;
 635
 636     if ($self->_dbh) { # we are connected
 637       $driver = $self->_dbh->{Driver}{Name};
 638     } else {
 639       # try to use dsn to not require being connected, the driver may still
 640       # force a connection in _rebless to determine version
 641       ($driver) = $self->_dbi_connect_info->[0] =~ /dbi:([^:]+):/i;
 642     }
 643
 644     if ($self->load_optional_class("DBIx::Class::Storage::DBI::${driver}")) {
 645       bless $self, "DBIx::Class::Storage::DBI::${driver}";
 646       $self->_rebless();
 647     }
 648   }
 649 }
 650
 651 sub _do_connection_actions {
 652   my $self = shift;
 653   my $connection_do = shift;
 654
 655   if (!ref $connection_do) {
 656     $self->_do_query($connection_do);
 657   }
 658   elsif (ref $connection_do eq 'ARRAY') {
 659     $self->_do_query($_) foreach @$connection_do;
 660   }
 661   elsif (ref $connection_do eq 'CODE') {
 662     $connection_do->($self);
 663   }
 664   else {
 665     $self->throw_exception (sprintf ("Don't know how to process conection actions of type '%s'", ref $connection_do) );
 666   }
 667
 668   return $self;
 669 }
 670
 671 sub _do_query {
 672   my ($self, $action) = @_;
 673
 674   if (ref $action eq 'CODE') {
 675     $action = $action->($self);
 676     $self->_do_query($_) foreach @$action;
 677   }
 678   else {
 679     # Most debuggers expect ($sql, @bind), so we need to exclude
 680     # the attribute hash which is the second argument to $dbh->do
 681     # furthermore the bind values are usually to be presented
 682     # as named arrayref pairs, so wrap those here too
 683     my @do_args = (ref $action eq 'ARRAY') ? (@$action) : ($action);
 684     my $sql = shift @do_args;
 685     my $attrs = shift @do_args;
 686     my @bind = map { [ undef, $_ ] } @do_args;
 687
 688     $self->_query_start($sql, @bind);
 689     $self->_dbh->do($sql, $attrs, @do_args);
 690     $self->_query_end($sql, @bind);
 691   }
 692
 693   return $self;
 694 }
 695
 696 sub _connect {
 697   my ($self, @info) = @_;
 698
 699   $self->throw_exception("You failed to provide any connection info")
 700     if !@info;
 701
 702   my ($old_connect_via, $dbh);
 703
 704   if ($INC{'Apache/DBI.pm'} && $ENV{MOD_PERL}) {
 705     $old_connect_via = $DBI::connect_via;
 706     $DBI::connect_via = 'connect';
 707   }
 708
 709   eval {
 710     if(ref $info[0] eq 'CODE') {
 711        $dbh = &{$info[0]}
 712     }
 713     else {
 714        $dbh = DBI->connect(@info);
 715     }
 716
 717     if($dbh && !$self->unsafe) {
 718       my $weak_self = $self;
 719       weaken($weak_self);
 720       $dbh->{HandleError} = sub {
 721           if ($weak_self) {
 722             $weak_self->throw_exception("DBI Exception: $_[0]");
 723           }
 724           else {
 725             croak ("DBI Exception: $_[0]");
 726           }
 727       };
 728       $dbh->{ShowErrorStatement} = 1;
 729       $dbh->{RaiseError} = 1;
 730       $dbh->{PrintError} = 0;
 731     }
 732   };
 733
 734   $DBI::connect_via = $old_connect_via if $old_connect_via;
 735
 736   $self->throw_exception("DBI Connection failed: " . ($@||$DBI::errstr))
 737     if !$dbh || $@;
 738
 739   $self->_dbh_autocommit($dbh->{AutoCommit});
 740
 741   $dbh;
 742 }
 743
 744 sub svp_begin {
 745   my ($self, $name) = @_;
 746
 747   $name = $self->_svp_generate_name
 748     unless defined $name;
 749
 750   $self->throw_exception ("You can't use savepoints outside a transaction")
 751     if $self->{transaction_depth} == 0;
 752
 753   $self->throw_exception ("Your Storage implementation doesn't support savepoints")
 754     unless $self->can('_svp_begin');
 755
 756   push @{ $self->{savepoints} }, $name;
 757
 758   $self->debugobj->svp_begin($name) if $self->debug;
 759
 760   return $self->_svp_begin($name);
 761 }
 762
 763 sub svp_release {
 764   my ($self, $name) = @_;
 765
 766   $self->throw_exception ("You can't use savepoints outside a transaction")
 767     if $self->{transaction_depth} == 0;
 768
 769   $self->throw_exception ("Your Storage implementation doesn't support savepoints")
 770     unless $self->can('_svp_release');
 771
 772   if (defined $name) {
 773     $self->throw_exception ("Savepoint '$name' does not exist")
 774       unless grep { $_ eq $name } @{ $self->{savepoints} };
 775
 776     # Dig through the stack until we find the one we are releasing.  This keeps
 777     # the stack up to date.
 778     my $svp;
 779
 780     do { $svp = pop @{ $self->{savepoints} } } while $svp ne $name;
 781   } else {
 782     $name = pop @{ $self->{savepoints} };
 783   }
 784
 785   $self->debugobj->svp_release($name) if $self->debug;
 786
 787   return $self->_svp_release($name);
 788 }
 789
 790 sub svp_rollback {
 791   my ($self, $name) = @_;
 792
 793   $self->throw_exception ("You can't use savepoints outside a transaction")
 794     if $self->{transaction_depth} == 0;
 795
 796   $self->throw_exception ("Your Storage implementation doesn't support savepoints")
 797     unless $self->can('_svp_rollback');
 798
 799   if (defined $name) {
 800       # If they passed us a name, verify that it exists in the stack
 801       unless(grep({ $_ eq $name } @{ $self->{savepoints} })) {
 802           $self->throw_exception("Savepoint '$name' does not exist!");
 803       }
 804
 805       # Dig through the stack until we find the one we are releasing.  This keeps
 806       # the stack up to date.
 807       while(my $s = pop(@{ $self->{savepoints} })) {
 808           last if($s eq $name);
 809       }
 810       # Add the savepoint back to the stack, as a rollback doesn't remove the
 811       # named savepoint, only everything after it.
 812       push(@{ $self->{savepoints} }, $name);
 813   } else {
 814       # We'll assume they want to rollback to the last savepoint
 815       $name = $self->{savepoints}->[-1];
 816   }
 817
 818   $self->debugobj->svp_rollback($name) if $self->debug;
 819
 820   return $self->_svp_rollback($name);
 821 }
 822
 823 sub _svp_generate_name {
 824     my ($self) = @_;
 825
 826     return 'savepoint_'.scalar(@{ $self->{'savepoints'} });
 827 }
 828
 829 sub txn_begin {
 830   my $self = shift;
 831   $self->ensure_connected();
 832   if($self->{transaction_depth} == 0) {
 833     $self->debugobj->txn_begin()
 834       if $self->debug;
 835     # this isn't ->_dbh-> because
 836     #  we should reconnect on begin_work
 837     #  for AutoCommit users
 838     $self->dbh->begin_work;
 839   } elsif ($self->auto_savepoint) {
 840     $self->svp_begin;
 841   }
 842   $self->{transaction_depth}++;
 843 }
 844
 845 sub txn_commit {
 846   my $self = shift;
 847   if ($self->{transaction_depth} == 1) {
 848     my $dbh = $self->_dbh;
 849     $self->debugobj->txn_commit()
 850       if ($self->debug);
 851     $dbh->commit;
 852     $self->{transaction_depth} = 0
 853       if $self->_dbh_autocommit;
 854   }
 855   elsif($self->{transaction_depth} > 1) {
 856     $self->{transaction_depth}--;
 857     $self->svp_release
 858       if $self->auto_savepoint;
 859   }
 860 }
 861
 862 sub txn_rollback {
 863   my $self = shift;
 864   my $dbh = $self->_dbh;
 865   eval {
 866     if ($self->{transaction_depth} == 1) {
 867       $self->debugobj->txn_rollback()
 868         if ($self->debug);
 869       $self->{transaction_depth} = 0
 870         if $self->_dbh_autocommit;
 871       $dbh->rollback;
 872     }
 873     elsif($self->{transaction_depth} > 1) {
 874       $self->{transaction_depth}--;
 875       if ($self->auto_savepoint) {
 876         $self->svp_rollback;
 877         $self->svp_release;
 878       }
 879     }
 880     else {
 881       die DBIx::Class::Storage::NESTED_ROLLBACK_EXCEPTION->new;
 882     }
 883   };
 884   if ($@) {
 885     my $error = $@;
 886     my $exception_class = "DBIx::Class::Storage::NESTED_ROLLBACK_EXCEPTION";
 887     $error =~ /$exception_class/ and $self->throw_exception($error);
 888     # ensure that a failed rollback resets the transaction depth
 889     $self->{transaction_depth} = $self->_dbh_autocommit ? 0 : 1;
 890     $self->throw_exception($error);
 891   }
 892 }
 893
 894 # This used to be the top-half of _execute.  It was split out to make it
 895 #  easier to override in NoBindVars without duping the rest.  It takes up
 896 #  all of _execute's args, and emits $sql, @bind.
 897 sub _prep_for_execute {
 898   my ($self, $op, $extra_bind, $ident, $args) = @_;
 899
 900   if( blessed($ident) && $ident->isa("DBIx::Class::ResultSource") ) {
 901     $ident = $ident->from();
 902   }
 903
 904   my ($sql, @bind) = $self->sql_maker->$op($ident, @$args);
 905
 906   unshift(@bind,
 907     map { ref $_ eq 'ARRAY' ? $_ : [ '!!dummy', $_ ] } @$extra_bind)
 908       if $extra_bind;
 909   return ($sql, \@bind);
 910 }
 911
 912 sub _fix_bind_params {
 913     my ($self, @bind) = @_;
 914
 915     ### Turn @bind from something like this:
 916     ###   ( [ "artist", 1 ], [ "cdid", 1, 3 ] )
 917     ### to this:
 918     ###   ( "'1'", "'1'", "'3'" )
 919     return
 920         map {
 921             if ( defined( $_ && $_->[1] ) ) {
 922                 map { qq{'$_'}; } @{$_}[ 1 .. $#$_ ];
 923             }
 924             else { q{'NULL'}; }
 925         } @bind;
 926 }
 927
 928 sub _query_start {
 929     my ( $self, $sql, @bind ) = @_;
 930
 931     if ( $self->debug ) {
 932         @bind = $self->_fix_bind_params(@bind);
 933
 934         $self->debugobj->query_start( $sql, @bind );
 935     }
 936 }
 937
 938 sub _query_end {
 939     my ( $self, $sql, @bind ) = @_;
 940
 941     if ( $self->debug ) {
 942         @bind = $self->_fix_bind_params(@bind);
 943         $self->debugobj->query_end( $sql, @bind );
 944     }
 945 }
 946
 947 sub _dbh_execute {
 948   my ($self, $dbh, $op, $extra_bind, $ident, $bind_attributes, @args) = @_;
 949
 950   my ($sql, $bind) = $self->_prep_for_execute($op, $extra_bind, $ident, \@args);
 951
 952   $self->_query_start( $sql, @$bind );
 953
 954   my $sth = $self->sth($sql,$op);
 955
 956   my $placeholder_index = 1;
 957
 958   foreach my $bound (@$bind) {
 959     my $attributes = {};
 960     my($column_name, @data) = @$bound;
 961
 962     if ($bind_attributes) {
 963       $attributes = $bind_attributes->{$column_name}
 964       if defined $bind_attributes->{$column_name};
 965     }
 966
 967     foreach my $data (@data) {
 968       my $ref = ref $data;
 969       $data = $ref && $ref ne 'ARRAY' ? ''.$data : $data; # stringify args (except arrayrefs)
 970
 971       $sth->bind_param($placeholder_index, $data, $attributes);
 972       $placeholder_index++;
 973     }
 974   }
 975
 976   # Can this fail without throwing an exception anyways???
 977   my $rv = $sth->execute();
 978   $self->throw_exception($sth->errstr) if !$rv;
 979
 980   $self->_query_end( $sql, @$bind );
 981
 982   return (wantarray ? ($rv, $sth, @$bind) : $rv);
 983 }
 984
 985 sub _execute {
 986     my $self = shift;
 987     $self->dbh_do('_dbh_execute', @_)
 988 }
 989
 990 sub insert {
 991   my ($self, $source, $to_insert) = @_;
 992
 993   my $ident = $source->from;
 994   my $bind_attributes = $self->source_bind_attributes($source);
 995
 996   my $updated_cols = {};
 997
 998   $self->ensure_connected;
 999   foreach my $col ( $source->columns ) {
1000     if ( !defined $to_insert->{$col} ) {
1001       my $col_info = $source->column_info($col);
1002
1003       if ( $col_info->{auto_nextval} ) {
1004         $updated_cols->{$col} = $to_insert->{$col} = $self->_sequence_fetch( 'nextval', $col_info->{sequence} || $self->_dbh_get_autoinc_seq($self->dbh, $source) );
1005       }
1006     }
1007   }
1008
1009   $self->_execute('insert' => [], $source, $bind_attributes, $to_insert);
1010
1011   return $updated_cols;
1012 }
1013
1014 ## Still not quite perfect, and EXPERIMENTAL
1015 ## Currently it is assumed that all values passed will be "normal", i.e. not
1016 ## scalar refs, or at least, all the same type as the first set, the statement is
1017 ## only prepped once.
1018 sub insert_bulk {
1019   my ($self, $source, $cols, $data) = @_;
1020   my %colvalues;
1021   my $table = $source->from;
1022   @colvalues{@$cols} = (0..$#$cols);
1023   my ($sql, @bind) = $self->sql_maker->insert($table, \%colvalues);
1024
1025   $self->_query_start( $sql, @bind );
1026   my $sth = $self->sth($sql);
1027
1028 #  @bind = map { ref $_ ? ''.$_ : $_ } @bind; # stringify args
1029
1030   ## This must be an arrayref, else nothing works!
1031   my $tuple_status = [];
1032
1033   ## Get the bind_attributes, if any exist
1034   my $bind_attributes = $self->source_bind_attributes($source);
1035
1036   ## Bind the values and execute
1037   my $placeholder_index = 1;
1038
1039   foreach my $bound (@bind) {
1040
1041     my $attributes = {};
1042     my ($column_name, $data_index) = @$bound;
1043
1044     if( $bind_attributes ) {
1045       $attributes = $bind_attributes->{$column_name}
1046       if defined $bind_attributes->{$column_name};
1047     }
1048
1049     my @data = map { $_->[$data_index] } @$data;
1050
1051     $sth->bind_param_array( $placeholder_index, [@data], $attributes );
1052     $placeholder_index++;
1053   }
1054   my $rv = $sth->execute_array({ArrayTupleStatus => $tuple_status});
1055   $self->throw_exception($sth->errstr) if !$rv;
1056
1057   $self->_query_end( $sql, @bind );
1058   return (wantarray ? ($rv, $sth, @bind) : $rv);
1059 }
1060
1061 sub update {
1062   my $self = shift @_;
1063   my $source = shift @_;
1064   my $bind_attributes = $self->source_bind_attributes($source);
1065
1066   return $self->_execute('update' => [], $source, $bind_attributes, @_);
1067 }
1068
1069
1070 sub delete {
1071   my $self = shift @_;
1072   my $source = shift @_;
1073
1074   my $bind_attrs = {}; ## If ever it's needed...
1075
1076   return $self->_execute('delete' => [], $source, $bind_attrs, @_);
1077 }
1078
1079 # We were sent here because the $rs contains a complex search
1080 # which will require a subquery to select the correct rows
1081 # (i.e. joined or limited resultsets)
1082 #
1083 # Genarating a single PK column subquery is trivial and supported
1084 # by all RDBMS. However if we have a multicolumn PK, things get ugly.
1085 # Look at multipk_update_delete()
1086 sub subq_update_delete {
1087   my $self = shift;
1088   my ($rs, $op, $values) = @_;
1089
1090   if ($rs->result_source->primary_columns == 1) {
1091     return $self->_onepk_update_delete (@_);
1092   }
1093   else {
1094     return $self->_multipk_update_delete (@_);
1095   }
1096 }
1097
1098 # Generally a single PK resultset operation is trivially expressed
1099 # with PK IN (subquery). However some databases (mysql) do not support
1100 # modification of a table mentioned in the subselect. This method
1101 # should be overriden in the appropriate storage class to be smarter
1102 # in such situations
1103 sub _onepk_update_delete {
1104
1105   my $self = shift;
1106   my ($rs, $op, $values) = @_;
1107
1108   my $rsrc = $rs->result_source;
1109   my $attrs = $rs->_resolved_attrs;
1110   my @pcols = $rsrc->primary_columns;
1111
1112   $self->throw_exception ('_onepk_update_delete can not be called on resultsets selecting multiple columns')
1113     if (ref $attrs->{select} eq 'ARRAY' and @{$attrs->{select}} > 1);
1114
1115   return $self->$op (
1116     $rsrc,
1117     $op eq 'update' ? $values : (),
1118     { $pcols[0] => { -in => $rs->as_query } },
1119   );
1120 }
1121
1122 # ANSI SQL does not provide a reliable way to perform a multicol-PK
1123 # resultset update/delete involving subqueries. So resort to simple
1124 # (and inefficient) delete_all style per-row opearations, while allowing
1125 # specific storages to override this with a faster implementation.
1126 #
1127 # We do not use $row->$op style queries, because resultset update/delete
1128 # is not expected to cascade (this is what delete_all/update_all is for).
1129 #
1130 # There should be no race conditions as the entire operation is rolled
1131 # in a transaction.
1132 sub _multipk_update_delete {
1133   my $self = shift;
1134   my ($rs, $op, $values) = @_;
1135
1136   my $rsrc = $rs->result_source;
1137   my @pcols = $rsrc->primary_columns;
1138   my $attrs = $rs->_resolved_attrs;
1139
1140   $self->throw_exception ('Number of columns selected by supplied resultset does not match number of primary keys')
1141     if ( ref $attrs->{select} ne 'ARRAY' or @{$attrs->{select}} != @pcols );
1142
1143   my $guard = $self->txn_scope_guard;
1144
1145   my $subrs_cur = $rs->cursor;
1146   while (my @pks = $subrs_cur->next) {
1147
1148     my $cond;
1149     for my $i (0.. $#pcols) {
1150       $cond->{$pcols[$i]} = $pks[$i];
1151     }
1152
1153     $self->$op (
1154       $rsrc,
1155       $op eq 'update' ? $values : (),
1156       $cond,
1157     );
1158   }
1159
1160   $guard->commit;
1161
1162   return 1;
1163 }
1164
1165
1166 sub _select {
1167   my $self = shift;
1168   my $sql_maker = $self->sql_maker;
1169   local $sql_maker->{for};
1170   return $self->_execute($self->_select_args(@_));
1171 }
1172
1173 sub _select_args {
1174   my ($self, $ident, $select, $condition, $attrs) = @_;
1175   my $order = $attrs->{order_by};
1176
1177   my $for = delete $attrs->{for};
1178   my $sql_maker = $self->sql_maker;
1179   $sql_maker->{for} = $for;
1180
1181   if (exists $attrs->{group_by} || $attrs->{having}) {
1182     $order = {
1183       group_by => $attrs->{group_by},
1184       having => $attrs->{having},
1185       ($order ? (order_by => $order) : ())
1186     };
1187   }
1188   my $bind_attrs = {}; ## Future support
1189   my @args = ('select', $attrs->{bind}, $ident, $bind_attrs, $select, $condition, $order);
1190   if ($attrs->{software_limit} ||
1191       $self->sql_maker->_default_limit_syntax eq "GenericSubQ") {
1192         $attrs->{software_limit} = 1;
1193   } else {
1194     $self->throw_exception("rows attribute must be positive if present")
1195       if (defined($attrs->{rows}) && !($attrs->{rows} > 0));
1196
1197     # MySQL actually recommends this approach.  I cringe.
1198     $attrs->{rows} = 2**48 if not defined $attrs->{rows} and defined $attrs->{offset};
1199     push @args, $attrs->{rows}, $attrs->{offset};
1200   }
1201   return @args;
1202 }
1203
1204 sub source_bind_attributes {
1205   my ($self, $source) = @_;
1206
1207   my $bind_attributes;
1208   foreach my $column ($source->columns) {
1209
1210     my $data_type = $source->column_info($column)->{data_type} || '';
1211     $bind_attributes->{$column} = $self->bind_attribute_by_data_type($data_type)
1212      if $data_type;
1213   }
1214
1215   return $bind_attributes;
1216 }
1217
1218 =head2 select
1219
1220 =over 4
1221
1222 =item Arguments: $ident, $select, $condition, $attrs
1223
1224 =back
1225
1226 Handle a SQL select statement.
1227
1228 =cut
1229
1230 sub select {
1231   my $self = shift;
1232   my ($ident, $select, $condition, $attrs) = @_;
1233   return $self->cursor_class->new($self, \@_, $attrs);
1234 }
1235
1236 sub select_single {
1237   my $self = shift;
1238   my ($rv, $sth, @bind) = $self->_select(@_);
1239   my @row = $sth->fetchrow_array;
1240   my @nextrow = $sth->fetchrow_array if @row;
1241   if(@row && @nextrow) {
1242     carp "Query returned more than one row.  SQL that returns multiple rows is DEPRECATED for ->find and ->single";
1243   }
1244   # Need to call finish() to work round broken DBDs
1245   $sth->finish();
1246   return @row;
1247 }
1248
1249 =head2 sth
1250
1251 =over 4
1252
1253 =item Arguments: $sql
1254
1255 =back
1256
1257 Returns a L<DBI> sth (statement handle) for the supplied SQL.
1258
1259 =cut
1260
1261 sub _dbh_sth {
1262   my ($self, $dbh, $sql) = @_;
1263
1264   # 3 is the if_active parameter which avoids active sth re-use
1265   my $sth = $self->disable_sth_caching
1266     ? $dbh->prepare($sql)
1267     : $dbh->prepare_cached($sql, {}, 3);
1268
1269   # XXX You would think RaiseError would make this impossible,
1270   #  but apparently that's not true :(
1271   $self->throw_exception($dbh->errstr) if !$sth;
1272
1273   $sth;
1274 }
1275
1276 sub sth {
1277   my ($self, $sql) = @_;
1278   $self->dbh_do('_dbh_sth', $sql);
1279 }
1280
1281 sub _dbh_columns_info_for {
1282   my ($self, $dbh, $table) = @_;
1283
1284   if ($dbh->can('column_info')) {
1285     my %result;
1286     eval {
1287       my ($schema,$tab) = $table =~ /^(.+?)\.(.+)$/ ? ($1,$2) : (undef,$table);
1288       my $sth = $dbh->column_info( undef,$schema, $tab, '%' );
1289       $sth->execute();
1290       while ( my $info = $sth->fetchrow_hashref() ){
1291         my %column_info;
1292         $column_info{data_type}   = $info->{TYPE_NAME};
1293         $column_info{size}      = $info->{COLUMN_SIZE};
1294         $column_info{is_nullable}   = $info->{NULLABLE} ? 1 : 0;
1295         $column_info{default_value} = $info->{COLUMN_DEF};
1296         my $col_name = $info->{COLUMN_NAME};
1297         $col_name =~ s/^\"(.*)\"$/$1/;
1298
1299         $result{$col_name} = \%column_info;
1300       }
1301     };
1302     return \%result if !$@ && scalar keys %result;
1303   }
1304
1305   my %result;
1306   my $sth = $dbh->prepare($self->sql_maker->select($table, undef, \'1 = 0'));
1307   $sth->execute;
1308   my @columns = @{$sth->{NAME_lc}};
1309   for my $i ( 0 .. $#columns ){
1310     my %column_info;
1311     $column_info{data_type} = $sth->{TYPE}->[$i];
1312     $column_info{size} = $sth->{PRECISION}->[$i];
1313     $column_info{is_nullable} = $sth->{NULLABLE}->[$i] ? 1 : 0;
1314
1315     if ($column_info{data_type} =~ m/^(.*?)\((.*?)\)$/) {
1316       $column_info{data_type} = $1;
1317       $column_info{size}    = $2;
1318     }
1319
1320     $result{$columns[$i]} = \%column_info;
1321   }
1322   $sth->finish;
1323
1324   foreach my $col (keys %result) {
1325     my $colinfo = $result{$col};
1326     my $type_num = $colinfo->{data_type};
1327     my $type_name;
1328     if(defined $type_num && $dbh->can('type_info')) {
1329       my $type_info = $dbh->type_info($type_num);
1330       $type_name = $type_info->{TYPE_NAME} if $type_info;
1331       $colinfo->{data_type} = $type_name if $type_name;
1332     }
1333   }
1334
1335   return \%result;
1336 }
1337
1338 sub columns_info_for {
1339   my ($self, $table) = @_;
1340   $self->dbh_do('_dbh_columns_info_for', $table);
1341 }
1342
1343 =head2 last_insert_id
1344
1345 Return the row id of the last insert.
1346
1347 =cut
1348
1349 sub _dbh_last_insert_id {
1350     # All Storage's need to register their own _dbh_last_insert_id
1351     # the old SQLite-based method was highly inappropriate
1352
1353     my $self = shift;
1354     my $class = ref $self;
1355     $self->throw_exception (<<EOE);
1356
1357 No _dbh_last_insert_id() method found in $class.
1358 Since the method of obtaining the autoincrement id of the last insert
1359 operation varies greatly between different databases, this method must be
1360 individually implemented for every storage class.
1361 EOE
1362 }
1363
1364 sub last_insert_id {
1365   my $self = shift;
1366   $self->dbh_do('_dbh_last_insert_id', @_);
1367 }
1368
1369 =head2 sqlt_type
1370
1371 Returns the database driver name.
1372
1373 =cut
1374
1375 sub sqlt_type { shift->dbh->{Driver}->{Name} }
1376
1377 =head2 bind_attribute_by_data_type
1378
1379 Given a datatype from column info, returns a database specific bind
1380 attribute for C<< $dbh->bind_param($val,$attribute) >> or nothing if we will
1381 let the database planner just handle it.
1382
1383 Generally only needed for special case column types, like bytea in postgres.
1384
1385 =cut
1386
1387 sub bind_attribute_by_data_type {
1388     return;
1389 }
1390
1391 =head2 create_ddl_dir
1392
1393 =over 4
1394
1395 =item Arguments: $schema \@databases, $version, $directory, $preversion, \%sqlt_args
1396
1397 =back
1398
1399 Creates a SQL file based on the Schema, for each of the specified
1400 database types, in the given directory.
1401
1402 By default, C<\%sqlt_args> will have
1403
1404  { add_drop_table => 1, ignore_constraint_names => 1, ignore_index_names => 1 }
1405
1406 merged with the hash passed in. To disable any of those features, pass in a
1407 hashref like the following
1408
1409  { ignore_constraint_names => 0, # ... other options }
1410
1411 =cut
1412
1413 sub create_ddl_dir {
1414   my ($self, $schema, $databases, $version, $dir, $preversion, $sqltargs) = @_;
1415
1416   if(!$dir || !-d $dir) {
1417     carp "No directory given, using ./\n";
1418     $dir = "./";
1419   }
1420   $databases ||= ['MySQL', 'SQLite', 'PostgreSQL'];
1421   $databases = [ $databases ] if(ref($databases) ne 'ARRAY');
1422
1423   my $schema_version = $schema->schema_version || '1.x';
1424   $version ||= $schema_version;
1425
1426   $sqltargs = {
1427     add_drop_table => 1,
1428     ignore_constraint_names => 1,
1429     ignore_index_names => 1,
1430     %{$sqltargs || {}}
1431   };
1432
1433   $self->throw_exception(q{Can't create a ddl file without SQL::Translator 0.09003: '}
1434       . $self->_check_sqlt_message . q{'})
1435           if !$self->_check_sqlt_version;
1436
1437   my $sqlt = SQL::Translator->new( $sqltargs );
1438
1439   $sqlt->parser('SQL::Translator::Parser::DBIx::Class');
1440   my $sqlt_schema = $sqlt->translate({ data => $schema })
1441     or $self->throw_exception ($sqlt->error);
1442
1443   foreach my $db (@$databases) {
1444     $sqlt->reset();
1445     $sqlt->{schema} = $sqlt_schema;
1446     $sqlt->producer($db);
1447
1448     my $file;
1449     my $filename = $schema->ddl_filename($db, $version, $dir);
1450     if (-e $filename && ($version eq $schema_version )) {
1451       # if we are dumping the current version, overwrite the DDL
1452       carp "Overwriting existing DDL file - $filename";
1453       unlink($filename);
1454     }
1455
1456     my $output = $sqlt->translate;
1457     if(!$output) {
1458       carp("Failed to translate to $db, skipping. (" . $sqlt->error . ")");
1459       next;
1460     }
1461     if(!open($file, ">$filename")) {
1462       $self->throw_exception("Can't open $filename for writing ($!)");
1463       next;
1464     }
1465     print $file $output;
1466     close($file);
1467
1468     next unless ($preversion);
1469
1470     require SQL::Translator::Diff;
1471
1472     my $prefilename = $schema->ddl_filename($db, $preversion, $dir);
1473     if(!-e $prefilename) {
1474       carp("No previous schema file found ($prefilename)");
1475       next;
1476     }
1477
1478     my $difffile = $schema->ddl_filename($db, $version, $dir, $preversion);
1479     if(-e $difffile) {
1480       carp("Overwriting existing diff file - $difffile");
1481       unlink($difffile);
1482     }
1483
1484     my $source_schema;
1485     {
1486       my $t = SQL::Translator->new($sqltargs);
1487       $t->debug( 0 );
1488       $t->trace( 0 );
1489
1490       $t->parser( $db )
1491         or $self->throw_exception ($t->error);
1492
1493       my $out = $t->translate( $prefilename )
1494         or $self->throw_exception ($t->error);
1495
1496       $source_schema = $t->schema;
1497
1498       $source_schema->name( $prefilename )
1499         unless ( $source_schema->name );
1500     }
1501
1502     # The "new" style of producers have sane normalization and can support
1503     # diffing a SQL file against a DBIC->SQLT schema. Old style ones don't
1504     # And we have to diff parsed SQL against parsed SQL.
1505     my $dest_schema = $sqlt_schema;
1506
1507     unless ( "SQL::Translator::Producer::$db"->can('preprocess_schema') ) {
1508       my $t = SQL::Translator->new($sqltargs);
1509       $t->debug( 0 );
1510       $t->trace( 0 );
1511
1512       $t->parser( $db )
1513         or $self->throw_exception ($t->error);
1514
1515       my $out = $t->translate( $filename )
1516         or $self->throw_exception ($t->error);
1517
1518       $dest_schema = $t->schema;
1519
1520       $dest_schema->name( $filename )
1521         unless $dest_schema->name;
1522     }
1523
1524     my $diff = SQL::Translator::Diff::schema_diff($source_schema, $db,
1525                                                   $dest_schema,   $db,
1526                                                   $sqltargs
1527                                                  );
1528     if(!open $file, ">$difffile") {
1529       $self->throw_exception("Can't write to $difffile ($!)");
1530       next;
1531     }
1532     print $file $diff;
1533     close($file);
1534   }
1535 }
1536
1537 =head2 deployment_statements
1538
1539 =over 4
1540
1541 =item Arguments: $schema, $type, $version, $directory, $sqlt_args
1542
1543 =back
1544
1545 Returns the statements used by L</deploy> and L<DBIx::Class::Schema/deploy>.
1546 The database driver name is given by C<$type>, though the value from
1547 L</sqlt_type> is used if it is not specified.
1548
1549 C<$directory> is used to return statements from files in a previously created
1550 L</create_ddl_dir> directory and is optional. The filenames are constructed
1551 from L<DBIx::Class::Schema/ddl_filename>, the schema name and the C<$version>.
1552
1553 If no C<$directory> is specified then the statements are constructed on the
1554 fly using L<SQL::Translator> and C<$version> is ignored.
1555
1556 See L<SQL::Translator/METHODS> for a list of values for C<$sqlt_args>.
1557
1558 =cut
1559
1560 sub deployment_statements {
1561   my ($self, $schema, $type, $version, $dir, $sqltargs) = @_;
1562   # Need to be connected to get the correct sqlt_type
1563   $self->ensure_connected() unless $type;
1564   $type ||= $self->sqlt_type;
1565   $version ||= $schema->schema_version || '1.x';
1566   $dir ||= './';
1567   my $filename = $schema->ddl_filename($type, $version, $dir);
1568   if(-f $filename)
1569   {
1570       my $file;
1571       open($file, "<$filename")
1572         or $self->throw_exception("Can't open $filename ($!)");
1573       my @rows = <$file>;
1574       close($file);
1575       return join('', @rows);
1576   }
1577
1578   $self->throw_exception(q{Can't deploy without SQL::Translator 0.09003: '}
1579       . $self->_check_sqlt_message . q{'})
1580           if !$self->_check_sqlt_version;
1581
1582   require SQL::Translator::Parser::DBIx::Class;
1583   eval qq{use SQL::Translator::Producer::${type}};
1584   $self->throw_exception($@) if $@;
1585
1586   # sources needs to be a parser arg, but for simplicty allow at top level
1587   # coming in
1588   $sqltargs->{parser_args}{sources} = delete $sqltargs->{sources}
1589       if exists $sqltargs->{sources};
1590
1591   my $tr = SQL::Translator->new(%$sqltargs);
1592   SQL::Translator::Parser::DBIx::Class::parse( $tr, $schema );
1593   return "SQL::Translator::Producer::${type}"->can('produce')->($tr);
1594 }
1595
1596 sub deploy {
1597   my ($self, $schema, $type, $sqltargs, $dir) = @_;
1598   my $deploy = sub {
1599     my $line = shift;
1600     return if($line =~ /^--/);
1601     return if(!$line);
1602     # next if($line =~ /^DROP/m);
1603     return if($line =~ /^BEGIN TRANSACTION/m);
1604     return if($line =~ /^COMMIT/m);
1605     return if $line =~ /^\s+$/; # skip whitespace only
1606     $self->_query_start($line);
1607     eval {
1608       $self->dbh->do($line); # shouldn't be using ->dbh ?
1609     };
1610     if ($@) {
1611       carp qq{$@ (running "${line}")};
1612     }
1613     $self->_query_end($line);
1614   };
1615   my @statements = $self->deployment_statements($schema, $type, undef, $dir, { no_comments => 1, %{ $sqltargs || {} } } );
1616   if (@statements > 1) {
1617     foreach my $statement (@statements) {
1618       $deploy->( $statement );
1619     }
1620   }
1621   elsif (@statements == 1) {
1622     foreach my $line ( split(";\n", $statements[0])) {
1623       $deploy->( $line );
1624     }
1625   }
1626 }
1627
1628 =head2 datetime_parser
1629
1630 Returns the datetime parser class
1631
1632 =cut
1633
1634 sub datetime_parser {
1635   my $self = shift;
1636   return $self->{datetime_parser} ||= do {
1637     $self->ensure_connected;
1638     $self->build_datetime_parser(@_);
1639   };
1640 }
1641
1642 =head2 datetime_parser_type
1643
1644 Defines (returns) the datetime parser class - currently hardwired to
1645 L<DateTime::Format::MySQL>
1646
1647 =cut
1648
1649 sub datetime_parser_type { "DateTime::Format::MySQL"; }
1650
1651 =head2 build_datetime_parser
1652
1653 See L</datetime_parser>
1654
1655 =cut
1656
1657 sub build_datetime_parser {
1658   my $self = shift;
1659   my $type = $self->datetime_parser_type(@_);
1660   eval "use ${type}";
1661   $self->throw_exception("Couldn't load ${type}: $@") if $@;
1662   return $type;
1663 }
1664
1665 {
1666     my $_check_sqlt_version; # private
1667     my $_check_sqlt_message; # private
1668     sub _check_sqlt_version {
1669         return $_check_sqlt_version if defined $_check_sqlt_version;
1670         eval 'use SQL::Translator "0.09003"';
1671         $_check_sqlt_message = $@ || '';
1672         $_check_sqlt_version = !$@;
1673     }
1674
1675     sub _check_sqlt_message {
1676         _check_sqlt_version if !defined $_check_sqlt_message;
1677         $_check_sqlt_message;
1678     }
1679 }
1680
1681 =head2 is_replicating
1682
1683 A boolean that reports if a particular L<DBIx::Class::Storage::DBI> is set to
1684 replicate from a master database.  Default is undef, which is the result
1685 returned by databases that don't support replication.
1686
1687 =cut
1688
1689 sub is_replicating {
1690     return;
1691
1692 }
1693
1694 =head2 lag_behind_master
1695
1696 Returns a number that represents a certain amount of lag behind a master db
1697 when a given storage is replicating.  The number is database dependent, but
1698 starts at zero and increases with the amount of lag. Default in undef
1699
1700 =cut
1701
1702 sub lag_behind_master {
1703     return;
1704 }
1705
1706 sub DESTROY {
1707   my $self = shift;
1708   return if !$self->_dbh;
1709   $self->_verify_pid;
1710   $self->_dbh(undef);
1711 }
1712
1713 1;
1714
1715 =head1 USAGE NOTES
1716
1717 =head2 DBIx::Class and AutoCommit
1718
1719 DBIx::Class can do some wonderful magic with handling exceptions,
1720 disconnections, and transactions when you use C<< AutoCommit => 1 >>
1721 combined with C<txn_do> for transaction support.
1722
1723 If you set C<< AutoCommit => 0 >> in your connect info, then you are always
1724 in an assumed transaction between commits, and you're telling us you'd
1725 like to manage that manually.  A lot of the magic protections offered by
1726 this module will go away.  We can't protect you from exceptions due to database
1727 disconnects because we don't know anything about how to restart your
1728 transactions.  You're on your own for handling all sorts of exceptional
1729 cases if you choose the C<< AutoCommit => 0 >> path, just as you would
1730 be with raw DBI.
1731
1732
1733
1734 =head1 AUTHORS
1735
1736 Matt S. Trout <mst@shadowcatsystems.co.uk>
1737
1738 Andy Grundman <andy@hybridized.org>
1739
1740 =head1 LICENSE
1741
1742 You may distribute this code under the same terms as Perl itself.
1743
1744 =cut