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