lib/DBIx/Class/Storage/DBI.pm

   1 package DBIx::Class::Storage::DBI;
   2 # -*- mode: cperl; cperl-indent-level: 2 -*-
   3
   4 use strict;
   5 use warnings;
   6
   7 use base qw/DBIx::Class::Storage::DBIHacks DBIx::Class::Storage/;
   8 use mro 'c3';
   9
  10 use Carp::Clan qw/^DBIx::Class/;
  11 use DBI;
  12 use DBIx::Class::Storage::DBI::Cursor;
  13 use DBIx::Class::Storage::Statistics;
  14 use Scalar::Util();
  15 use List::Util();
  16 use Data::Dumper::Concise();
  17 use Sub::Name ();
  18
  19 __PACKAGE__->mk_group_accessors('simple' =>
  20   qw/_connect_info _dbi_connect_info _dbh _sql_maker _sql_maker_opts _conn_pid
  21      _conn_tid transaction_depth _dbh_autocommit _driver_determined savepoints/
  22 );
  23
  24 # the values for these accessors are picked out (and deleted) from
  25 # the attribute hashref passed to connect_info
  26 my @storage_options = qw/
  27   on_connect_call on_disconnect_call on_connect_do on_disconnect_do
  28   disable_sth_caching unsafe auto_savepoint
  29 /;
  30 __PACKAGE__->mk_group_accessors('simple' => @storage_options);
  31
  32
  33 # default cursor class, overridable in connect_info attributes
  34 __PACKAGE__->cursor_class('DBIx::Class::Storage::DBI::Cursor');
  35
  36 __PACKAGE__->mk_group_accessors('inherited' => qw/sql_maker_class/);
  37 __PACKAGE__->sql_maker_class('DBIx::Class::SQLAHacks');
  38
  39
  40 # Each of these methods need _determine_driver called before itself
  41 # in order to function reliably. This is a purely DRY optimization
  42 my @rdbms_specific_methods = qw/
  43   deployment_statements
  44   sqlt_type
  45   build_datetime_parser
  46   datetime_parser_type
  47
  48   insert
  49   insert_bulk
  50   update
  51   delete
  52   select
  53   select_single
  54 /;
  55
  56 for my $meth (@rdbms_specific_methods) {
  57
  58   my $orig = __PACKAGE__->can ($meth)
  59     or next;
  60
  61   no strict qw/refs/;
  62   no warnings qw/redefine/;
  63   *{__PACKAGE__ ."::$meth"} = Sub::Name::subname $meth => sub {
  64     if (not $_[0]->_driver_determined) {
  65       $_[0]->_determine_driver;
  66       goto $_[0]->can($meth);
  67     }
  68     $orig->(@_);
  69   };
  70 }
  71
  72
  73 =head1 NAME
  74
  75 DBIx::Class::Storage::DBI - DBI storage handler
  76
  77 =head1 SYNOPSIS
  78
  79   my $schema = MySchema->connect('dbi:SQLite:my.db');
  80
  81   $schema->storage->debug(1);
  82
  83   my @stuff = $schema->storage->dbh_do(
  84     sub {
  85       my ($storage, $dbh, @args) = @_;
  86       $dbh->do("DROP TABLE authors");
  87     },
  88     @column_list
  89   );
  90
  91   $schema->resultset('Book')->search({
  92      written_on => $schema->storage->datetime_parser(DateTime->now)
  93   });
  94
  95 =head1 DESCRIPTION
  96
  97 This class represents the connection to an RDBMS via L<DBI>.  See
  98 L<DBIx::Class::Storage> for general information.  This pod only
  99 documents DBI-specific methods and behaviors.
 100
 101 =head1 METHODS
 102
 103 =cut
 104
 105 sub new {
 106   my $new = shift->next::method(@_);
 107
 108   $new->transaction_depth(0);
 109   $new->_sql_maker_opts({});
 110   $new->{savepoints} = [];
 111   $new->{_in_dbh_do} = 0;
 112   $new->{_dbh_gen} = 0;
 113
 114   $new;
 115 }
 116
 117 =head2 connect_info
 118
 119 This method is normally called by L<DBIx::Class::Schema/connection>, which
 120 encapsulates its argument list in an arrayref before passing them here.
 121
 122 The argument list may contain:
 123
 124 =over
 125
 126 =item *
 127
 128 The same 4-element argument set one would normally pass to
 129 L<DBI/connect>, optionally followed by
 130 L<extra attributes|/DBIx::Class specific connection attributes>
 131 recognized by DBIx::Class:
 132
 133   $connect_info_args = [ $dsn, $user, $password, \%dbi_attributes?, \%extra_attributes? ];
 134
 135 =item *
 136
 137 A single code reference which returns a connected
 138 L<DBI database handle|DBI/connect> optionally followed by
 139 L<extra attributes|/DBIx::Class specific connection attributes> recognized
 140 by DBIx::Class:
 141
 142   $connect_info_args = [ sub { DBI->connect (...) }, \%extra_attributes? ];
 143
 144 =item *
 145
 146 A single hashref with all the attributes and the dsn/user/password
 147 mixed together:
 148
 149   $connect_info_args = [{
 150     dsn => $dsn,
 151     user => $user,
 152     password => $pass,
 153     %dbi_attributes,
 154     %extra_attributes,
 155   }];
 156
 157   $connect_info_args = [{
 158     dbh_maker => sub { DBI->connect (...) },
 159     %dbi_attributes,
 160     %extra_attributes,
 161   }];
 162
 163 This is particularly useful for L<Catalyst> based applications, allowing the
 164 following config (L<Config::General> style):
 165
 166   <Model::DB>
 167     schema_class   App::DB
 168     <connect_info>
 169       dsn          dbi:mysql:database=test
 170       user         testuser
 171       password     TestPass
 172       AutoCommit   1
 173     </connect_info>
 174   </Model::DB>
 175
 176 The C<dsn>/C<user>/C<password> combination can be substituted by the
 177 C<dbh_maker> key whose value is a coderef that returns a connected
 178 L<DBI database handle|DBI/connect>
 179
 180 =back
 181
 182 Please note that the L<DBI> docs recommend that you always explicitly
 183 set C<AutoCommit> to either I<0> or I<1>.  L<DBIx::Class> further
 184 recommends that it be set to I<1>, and that you perform transactions
 185 via our L<DBIx::Class::Schema/txn_do> method.  L<DBIx::Class> will set it
 186 to I<1> if you do not do explicitly set it to zero.  This is the default
 187 for most DBDs. See L</DBIx::Class and AutoCommit> for details.
 188
 189 =head3 DBIx::Class specific connection attributes
 190
 191 In addition to the standard L<DBI|DBI/ATTRIBUTES_COMMON_TO_ALL_HANDLES>
 192 L<connection|DBI/Database_Handle_Attributes> attributes, DBIx::Class recognizes
 193 the following connection options. These options can be mixed in with your other
 194 L<DBI> connection attributes, or placed in a separate hashref
 195 (C<\%extra_attributes>) as shown above.
 196
 197 Every time C<connect_info> is invoked, any previous settings for
 198 these options will be cleared before setting the new ones, regardless of
 199 whether any options are specified in the new C<connect_info>.
 200
 201
 202 =over
 203
 204 =item on_connect_do
 205
 206 Specifies things to do immediately after connecting or re-connecting to
 207 the database.  Its value may contain:
 208
 209 =over
 210
 211 =item a scalar
 212
 213 This contains one SQL statement to execute.
 214
 215 =item an array reference
 216
 217 This contains SQL statements to execute in order.  Each element contains
 218 a string or a code reference that returns a string.
 219
 220 =item a code reference
 221
 222 This contains some code to execute.  Unlike code references within an
 223 array reference, its return value is ignored.
 224
 225 =back
 226
 227 =item on_disconnect_do
 228
 229 Takes arguments in the same form as L</on_connect_do> and executes them
 230 immediately before disconnecting from the database.
 231
 232 Note, this only runs if you explicitly call L</disconnect> on the
 233 storage object.
 234
 235 =item on_connect_call
 236
 237 A more generalized form of L</on_connect_do> that calls the specified
 238 C<connect_call_METHOD> methods in your storage driver.
 239
 240   on_connect_do => 'select 1'
 241
 242 is equivalent to:
 243
 244   on_connect_call => [ [ do_sql => 'select 1' ] ]
 245
 246 Its values may contain:
 247
 248 =over
 249
 250 =item a scalar
 251
 252 Will call the C<connect_call_METHOD> method.
 253
 254 =item a code reference
 255
 256 Will execute C<< $code->($storage) >>
 257
 258 =item an array reference
 259
 260 Each value can be a method name or code reference.
 261
 262 =item an array of arrays
 263
 264 For each array, the first item is taken to be the C<connect_call_> method name
 265 or code reference, and the rest are parameters to it.
 266
 267 =back
 268
 269 Some predefined storage methods you may use:
 270
 271 =over
 272
 273 =item do_sql
 274
 275 Executes a SQL string or a code reference that returns a SQL string. This is
 276 what L</on_connect_do> and L</on_disconnect_do> use.
 277
 278 It can take:
 279
 280 =over
 281
 282 =item a scalar
 283
 284 Will execute the scalar as SQL.
 285
 286 =item an arrayref
 287
 288 Taken to be arguments to L<DBI/do>, the SQL string optionally followed by the
 289 attributes hashref and bind values.
 290
 291 =item a code reference
 292
 293 Will execute C<< $code->($storage) >> and execute the return array refs as
 294 above.
 295
 296 =back
 297
 298 =item datetime_setup
 299
 300 Execute any statements necessary to initialize the database session to return
 301 and accept datetime/timestamp values used with
 302 L<DBIx::Class::InflateColumn::DateTime>.
 303
 304 Only necessary for some databases, see your specific storage driver for
 305 implementation details.
 306
 307 =back
 308
 309 =item on_disconnect_call
 310
 311 Takes arguments in the same form as L</on_connect_call> and executes them
 312 immediately before disconnecting from the database.
 313
 314 Calls the C<disconnect_call_METHOD> methods as opposed to the
 315 C<connect_call_METHOD> methods called by L</on_connect_call>.
 316
 317 Note, this only runs if you explicitly call L</disconnect> on the
 318 storage object.
 319
 320 =item disable_sth_caching
 321
 322 If set to a true value, this option will disable the caching of
 323 statement handles via L<DBI/prepare_cached>.
 324
 325 =item limit_dialect
 326
 327 Sets the limit dialect. This is useful for JDBC-bridge among others
 328 where the remote SQL-dialect cannot be determined by the name of the
 329 driver alone. See also L<SQL::Abstract::Limit>.
 330
 331 =item quote_char
 332
 333 Specifies what characters to use to quote table and column names. If
 334 you use this you will want to specify L</name_sep> as well.
 335
 336 C<quote_char> expects either a single character, in which case is it
 337 is placed on either side of the table/column name, or an arrayref of length
 338 2 in which case the table/column name is placed between the elements.
 339
 340 For example under MySQL you should use C<< quote_char => '`' >>, and for
 341 SQL Server you should use C<< quote_char => [qw/[ ]/] >>.
 342
 343 =item name_sep
 344
 345 This only needs to be used in conjunction with C<quote_char>, and is used to
 346 specify the character that separates elements (schemas, tables, columns) from
 347 each other. In most cases this is simply a C<.>.
 348
 349 The consequences of not supplying this value is that L<SQL::Abstract>
 350 will assume DBIx::Class' uses of aliases to be complete column
 351 names. The output will look like I<"me.name"> when it should actually
 352 be I<"me"."name">.
 353
 354 =item unsafe
 355
 356 This Storage driver normally installs its own C<HandleError>, sets
 357 C<RaiseError> and C<ShowErrorStatement> on, and sets C<PrintError> off on
 358 all database handles, including those supplied by a coderef.  It does this
 359 so that it can have consistent and useful error behavior.
 360
 361 If you set this option to a true value, Storage will not do its usual
 362 modifications to the database handle's attributes, and instead relies on
 363 the settings in your connect_info DBI options (or the values you set in
 364 your connection coderef, in the case that you are connecting via coderef).
 365
 366 Note that your custom settings can cause Storage to malfunction,
 367 especially if you set a C<HandleError> handler that suppresses exceptions
 368 and/or disable C<RaiseError>.
 369
 370 =item auto_savepoint
 371
 372 If this option is true, L<DBIx::Class> will use savepoints when nesting
 373 transactions, making it possible to recover from failure in the inner
 374 transaction without having to abort all outer transactions.
 375
 376 =item cursor_class
 377
 378 Use this argument to supply a cursor class other than the default
 379 L<DBIx::Class::Storage::DBI::Cursor>.
 380
 381 =back
 382
 383 Some real-life examples of arguments to L</connect_info> and
 384 L<DBIx::Class::Schema/connect>
 385
 386   # Simple SQLite connection
 387   ->connect_info([ 'dbi:SQLite:./foo.db' ]);
 388
 389   # Connect via subref
 390   ->connect_info([ sub { DBI->connect(...) } ]);
 391
 392   # Connect via subref in hashref
 393   ->connect_info([{
 394     dbh_maker => sub { DBI->connect(...) },
 395     on_connect_do => 'alter session ...',
 396   }]);
 397
 398   # A bit more complicated
 399   ->connect_info(
 400     [
 401       'dbi:Pg:dbname=foo',
 402       'postgres',
 403       'my_pg_password',
 404       { AutoCommit => 1 },
 405       { quote_char => q{"}, name_sep => q{.} },
 406     ]
 407   );
 408
 409   # Equivalent to the previous example
 410   ->connect_info(
 411     [
 412       'dbi:Pg:dbname=foo',
 413       'postgres',
 414       'my_pg_password',
 415       { AutoCommit => 1, quote_char => q{"}, name_sep => q{.} },
 416     ]
 417   );
 418
 419   # Same, but with hashref as argument
 420   # See parse_connect_info for explanation
 421   ->connect_info(
 422     [{
 423       dsn         => 'dbi:Pg:dbname=foo',
 424       user        => 'postgres',
 425       password    => 'my_pg_password',
 426       AutoCommit  => 1,
 427       quote_char  => q{"},
 428       name_sep    => q{.},
 429     }]
 430   );
 431
 432   # Subref + DBIx::Class-specific connection options
 433   ->connect_info(
 434     [
 435       sub { DBI->connect(...) },
 436       {
 437           quote_char => q{`},
 438           name_sep => q{@},
 439           on_connect_do => ['SET search_path TO myschema,otherschema,public'],
 440           disable_sth_caching => 1,
 441       },
 442     ]
 443   );
 444
 445
 446
 447 =cut
 448
 449 sub connect_info {
 450   my ($self, $info) = @_;
 451
 452   return $self->_connect_info if !$info;
 453
 454   $self->_connect_info($info); # copy for _connect_info
 455
 456   $info = $self->_normalize_connect_info($info)
 457     if ref $info eq 'ARRAY';
 458
 459   for my $storage_opt (keys %{ $info->{storage_options} }) {
 460     my $value = $info->{storage_options}{$storage_opt};
 461
 462     $self->$storage_opt($value);
 463   }
 464
 465   # Kill sql_maker/_sql_maker_opts, so we get a fresh one with only
 466   #  the new set of options
 467   $self->_sql_maker(undef);
 468   $self->_sql_maker_opts({});
 469
 470   for my $sql_maker_opt (keys %{ $info->{sql_maker_options} }) {
 471     my $value = $info->{sql_maker_options}{$sql_maker_opt};
 472
 473     $self->_sql_maker_opts->{$sql_maker_opt} = $value;
 474   }
 475
 476   my %attrs = (
 477     %{ $self->_default_dbi_connect_attributes || {} },
 478     %{ $info->{attributes} || {} },
 479   );
 480
 481   my @args = @{ $info->{arguments} };
 482
 483   $self->_dbi_connect_info([@args,
 484     %attrs && !(ref $args[0] eq 'CODE') ? \%attrs : ()]);
 485
 486   return $self->_connect_info;
 487 }
 488
 489 sub _normalize_connect_info {
 490   my ($self, $info_arg) = @_;
 491   my %info;
 492
 493   my @args = @$info_arg;  # take a shallow copy for further mutilation
 494
 495   # combine/pre-parse arguments depending on invocation style
 496
 497   my %attrs;
 498   if (ref $args[0] eq 'CODE') {     # coderef with optional \%extra_attributes
 499     %attrs = %{ $args[1] || {} };
 500     @args = $args[0];
 501   }
 502   elsif (ref $args[0] eq 'HASH') { # single hashref (i.e. Catalyst config)
 503     %attrs = %{$args[0]};
 504     @args = ();
 505     if (my $code = delete $attrs{dbh_maker}) {
 506       @args = $code;
 507
 508       my @ignored = grep { delete $attrs{$_} } (qw/dsn user password/);
 509       if (@ignored) {
 510         carp sprintf (
 511             'Attribute(s) %s in connect_info were ignored, as they can not be applied '
 512           . "to the result of 'dbh_maker'",
 513
 514           join (', ', map { "'$_'" } (@ignored) ),
 515         );
 516       }
 517     }
 518     else {
 519       @args = delete @attrs{qw/dsn user password/};
 520     }
 521   }
 522   else {                # otherwise assume dsn/user/password + \%attrs + \%extra_attrs
 523     %attrs = (
 524       % { $args[3] || {} },
 525       % { $args[4] || {} },
 526     );
 527     @args = @args[0,1,2];
 528   }
 529
 530   $info{arguments} = \@args;
 531
 532   my @storage_opts = grep exists $attrs{$_},
 533     @storage_options, 'cursor_class';
 534
 535   @{ $info{storage_options} }{@storage_opts} =
 536     delete @attrs{@storage_opts} if @storage_opts;
 537
 538   my @sql_maker_opts = grep exists $attrs{$_},
 539     qw/limit_dialect quote_char name_sep/;
 540
 541   @{ $info{sql_maker_options} }{@sql_maker_opts} =
 542     delete @attrs{@sql_maker_opts} if @sql_maker_opts;
 543
 544   $info{attributes} = \%attrs if %attrs;
 545
 546   return \%info;
 547 }
 548
 549 sub _default_dbi_connect_attributes {
 550   return {
 551     AutoCommit => 1,
 552     RaiseError => 1,
 553     PrintError => 0,
 554   };
 555 }
 556
 557 =head2 on_connect_do
 558
 559 This method is deprecated in favour of setting via L</connect_info>.
 560
 561 =cut
 562
 563 =head2 on_disconnect_do
 564
 565 This method is deprecated in favour of setting via L</connect_info>.
 566
 567 =cut
 568
 569 sub _parse_connect_do {
 570   my ($self, $type) = @_;
 571
 572   my $val = $self->$type;
 573   return () if not defined $val;
 574
 575   my @res;
 576
 577   if (not ref($val)) {
 578     push @res, [ 'do_sql', $val ];
 579   } elsif (ref($val) eq 'CODE') {
 580     push @res, $val;
 581   } elsif (ref($val) eq 'ARRAY') {
 582     push @res, map { [ 'do_sql', $_ ] } @$val;
 583   } else {
 584     $self->throw_exception("Invalid type for $type: ".ref($val));
 585   }
 586
 587   return \@res;
 588 }
 589
 590 =head2 dbh_do
 591
 592 Arguments: ($subref | $method_name), @extra_coderef_args?
 593
 594 Execute the given $subref or $method_name using the new exception-based
 595 connection management.
 596
 597 The first two arguments will be the storage object that C<dbh_do> was called
 598 on and a database handle to use.  Any additional arguments will be passed
 599 verbatim to the called subref as arguments 2 and onwards.
 600
 601 Using this (instead of $self->_dbh or $self->dbh) ensures correct
 602 exception handling and reconnection (or failover in future subclasses).
 603
 604 Your subref should have no side-effects outside of the database, as
 605 there is the potential for your subref to be partially double-executed
 606 if the database connection was stale/dysfunctional.
 607
 608 Example:
 609
 610   my @stuff = $schema->storage->dbh_do(
 611     sub {
 612       my ($storage, $dbh, @cols) = @_;
 613       my $cols = join(q{, }, @cols);
 614       $dbh->selectrow_array("SELECT $cols FROM foo");
 615     },
 616     @column_list
 617   );
 618
 619 =cut
 620
 621 sub dbh_do {
 622   my $self = shift;
 623   my $code = shift;
 624
 625   my $dbh = $self->_get_dbh;
 626
 627   return $self->$code($dbh, @_) if $self->{_in_dbh_do}
 628       || $self->{transaction_depth};
 629
 630   local $self->{_in_dbh_do} = 1;
 631
 632   my @result;
 633   my $want_array = wantarray;
 634
 635   eval {
 636
 637     if($want_array) {
 638         @result = $self->$code($dbh, @_);
 639     }
 640     elsif(defined $want_array) {
 641         $result[0] = $self->$code($dbh, @_);
 642     }
 643     else {
 644         $self->$code($dbh, @_);
 645     }
 646   };
 647
 648   # ->connected might unset $@ - copy
 649   my $exception = $@;
 650   if(!$exception) { return $want_array ? @result : $result[0] }
 651
 652   $self->throw_exception($exception) if $self->connected;
 653
 654   # We were not connected - reconnect and retry, but let any
 655   #  exception fall right through this time
 656   carp "Retrying $code after catching disconnected exception: $exception"
 657     if $ENV{DBIC_DBIRETRY_DEBUG};
 658   $self->_populate_dbh;
 659   $self->$code($self->_dbh, @_);
 660 }
 661
 662 # This is basically a blend of dbh_do above and DBIx::Class::Storage::txn_do.
 663 # It also informs dbh_do to bypass itself while under the direction of txn_do,
 664 #  via $self->{_in_dbh_do} (this saves some redundant eval and errorcheck, etc)
 665 sub txn_do {
 666   my $self = shift;
 667   my $coderef = shift;
 668
 669   ref $coderef eq 'CODE' or $self->throw_exception
 670     ('$coderef must be a CODE reference');
 671
 672   return $coderef->(@_) if $self->{transaction_depth} && ! $self->auto_savepoint;
 673
 674   local $self->{_in_dbh_do} = 1;
 675
 676   my @result;
 677   my $want_array = wantarray;
 678
 679   my $tried = 0;
 680   while(1) {
 681     eval {
 682       $self->_get_dbh;
 683
 684       $self->txn_begin;
 685       if($want_array) {
 686           @result = $coderef->(@_);
 687       }
 688       elsif(defined $want_array) {
 689           $result[0] = $coderef->(@_);
 690       }
 691       else {
 692           $coderef->(@_);
 693       }
 694       $self->txn_commit;
 695     };
 696
 697     # ->connected might unset $@ - copy
 698     my $exception = $@;
 699     if(!$exception) { return $want_array ? @result : $result[0] }
 700
 701     if($tried++ || $self->connected) {
 702       eval { $self->txn_rollback };
 703       my $rollback_exception = $@;
 704       if($rollback_exception) {
 705         my $exception_class = "DBIx::Class::Storage::NESTED_ROLLBACK_EXCEPTION";
 706         $self->throw_exception($exception)  # propagate nested rollback
 707           if $rollback_exception =~ /$exception_class/;
 708
 709         $self->throw_exception(
 710           "Transaction aborted: ${exception}. "
 711           . "Rollback failed: ${rollback_exception}"
 712         );
 713       }
 714       $self->throw_exception($exception)
 715     }
 716
 717     # We were not connected, and was first try - reconnect and retry
 718     # via the while loop
 719     carp "Retrying $coderef after catching disconnected exception: $exception"
 720       if $ENV{DBIC_DBIRETRY_DEBUG};
 721     $self->_populate_dbh;
 722   }
 723 }
 724
 725 =head2 disconnect
 726
 727 Our C<disconnect> method also performs a rollback first if the
 728 database is not in C<AutoCommit> mode.
 729
 730 =cut
 731
 732 sub disconnect {
 733   my ($self) = @_;
 734
 735   if( $self->_dbh ) {
 736     my @actions;
 737
 738     push @actions, ( $self->on_disconnect_call || () );
 739     push @actions, $self->_parse_connect_do ('on_disconnect_do');
 740
 741     $self->_do_connection_actions(disconnect_call_ => $_) for @actions;
 742
 743     $self->_dbh_rollback unless $self->_dbh_autocommit;
 744
 745     %{ $self->_dbh->{CachedKids} } = ();
 746     $self->_dbh->disconnect;
 747     $self->_dbh(undef);
 748     $self->{_dbh_gen}++;
 749   }
 750 }
 751
 752 =head2 with_deferred_fk_checks
 753
 754 =over 4
 755
 756 =item Arguments: C<$coderef>
 757
 758 =item Return Value: The return value of $coderef
 759
 760 =back
 761
 762 Storage specific method to run the code ref with FK checks deferred or
 763 in MySQL's case disabled entirely.
 764
 765 =cut
 766
 767 # Storage subclasses should override this
 768 sub with_deferred_fk_checks {
 769   my ($self, $sub) = @_;
 770   $sub->();
 771 }
 772
 773 =head2 connected
 774
 775 =over
 776
 777 =item Arguments: none
 778
 779 =item Return Value: 1|0
 780
 781 =back
 782
 783 Verifies that the current database handle is active and ready to execute
 784 an SQL statement (e.g. the connection did not get stale, server is still
 785 answering, etc.) This method is used internally by L</dbh>.
 786
 787 =cut
 788
 789 sub connected {
 790   my $self = shift;
 791   return 0 unless $self->_seems_connected;
 792
 793   #be on the safe side
 794   local $self->_dbh->{RaiseError} = 1;
 795
 796   return $self->_ping;
 797 }
 798
 799 sub _seems_connected {
 800   my $self = shift;
 801
 802   my $dbh = $self->_dbh
 803     or return 0;
 804
 805   if(defined $self->_conn_tid && $self->_conn_tid != threads->tid) {
 806     $self->_dbh(undef);
 807     $self->{_dbh_gen}++;
 808     return 0;
 809   }
 810   else {
 811     $self->_verify_pid;
 812     return 0 if !$self->_dbh;
 813   }
 814
 815   return $dbh->FETCH('Active');
 816 }
 817
 818 sub _ping {
 819   my $self = shift;
 820
 821   my $dbh = $self->_dbh or return 0;
 822
 823   return $dbh->ping;
 824 }
 825
 826 # handle pid changes correctly
 827 #  NOTE: assumes $self->_dbh is a valid $dbh
 828 sub _verify_pid {
 829   my ($self) = @_;
 830
 831   return if defined $self->_conn_pid && $self->_conn_pid == $$;
 832
 833   $self->_dbh->{InactiveDestroy} = 1;
 834   $self->_dbh(undef);
 835   $self->{_dbh_gen}++;
 836
 837   return;
 838 }
 839
 840 sub ensure_connected {
 841   my ($self) = @_;
 842
 843   unless ($self->connected) {
 844     $self->_populate_dbh;
 845   }
 846 }
 847
 848 =head2 dbh
 849
 850 Returns a C<$dbh> - a data base handle of class L<DBI>. The returned handle
 851 is guaranteed to be healthy by implicitly calling L</connected>, and if
 852 necessary performing a reconnection before returning. Keep in mind that this
 853 is very B<expensive> on some database engines. Consider using L<dbh_do>
 854 instead.
 855
 856 =cut
 857
 858 sub dbh {
 859   my ($self) = @_;
 860
 861   if (not $self->_dbh) {
 862     $self->_populate_dbh;
 863   } else {
 864     $self->ensure_connected;
 865   }
 866   return $self->_dbh;
 867 }
 868
 869 # this is the internal "get dbh or connect (don't check)" method
 870 sub _get_dbh {
 871   my $self = shift;
 872   $self->_verify_pid if $self->_dbh;
 873   $self->_populate_dbh unless $self->_dbh;
 874   return $self->_dbh;
 875 }
 876
 877 sub _sql_maker_args {
 878     my ($self) = @_;
 879
 880     return (
 881       bindtype=>'columns',
 882       array_datatypes => 1,
 883       limit_dialect => $self->_get_dbh,
 884       %{$self->_sql_maker_opts}
 885     );
 886 }
 887
 888 sub sql_maker {
 889   my ($self) = @_;
 890   unless ($self->_sql_maker) {
 891     my $sql_maker_class = $self->sql_maker_class;
 892     $self->ensure_class_loaded ($sql_maker_class);
 893     $self->_sql_maker($sql_maker_class->new( $self->_sql_maker_args ));
 894   }
 895   return $self->_sql_maker;
 896 }
 897
 898 # nothing to do by default
 899 sub _rebless {}
 900 sub _init {}
 901
 902 sub _populate_dbh {
 903   my ($self) = @_;
 904
 905   my @info = @{$self->_dbi_connect_info || []};
 906   $self->_dbh(undef); # in case ->connected failed we might get sent here
 907   $self->_dbh($self->_connect(@info));
 908
 909   $self->_conn_pid($$);
 910   $self->_conn_tid(threads->tid) if $INC{'threads.pm'};
 911
 912   $self->_determine_driver;
 913
 914   # Always set the transaction depth on connect, since
 915   #  there is no transaction in progress by definition
 916   $self->{transaction_depth} = $self->_dbh_autocommit ? 0 : 1;
 917
 918   $self->_run_connection_actions unless $self->{_in_determine_driver};
 919 }
 920
 921 sub _run_connection_actions {
 922   my $self = shift;
 923   my @actions;
 924
 925   push @actions, ( $self->on_connect_call || () );
 926   push @actions, $self->_parse_connect_do ('on_connect_do');
 927
 928   $self->_do_connection_actions(connect_call_ => $_) for @actions;
 929 }
 930
 931 sub _determine_driver {
 932   my ($self) = @_;
 933
 934   if ((not $self->_driver_determined) && (not $self->{_in_determine_driver})) {
 935     my $started_connected = 0;
 936     local $self->{_in_determine_driver} = 1;
 937
 938     if (ref($self) eq __PACKAGE__) {
 939       my $driver;
 940       if ($self->_dbh) { # we are connected
 941         $driver = $self->_dbh->{Driver}{Name};
 942         $started_connected = 1;
 943       } else {
 944         # if connect_info is a CODEREF, we have no choice but to connect
 945         if (ref $self->_dbi_connect_info->[0] &&
 946             Scalar::Util::reftype($self->_dbi_connect_info->[0]) eq 'CODE') {
 947           $self->_populate_dbh;
 948           $driver = $self->_dbh->{Driver}{Name};
 949         }
 950         else {
 951           # try to use dsn to not require being connected, the driver may still
 952           # force a connection in _rebless to determine version
 953           # (dsn may not be supplied at all if all we do is make a mock-schema)
 954           my $dsn = $self->_dbi_connect_info->[0] || '';
 955           ($driver) = $dsn =~ /dbi:([^:]+):/i;
 956         }
 957       }
 958
 959       if ($driver) {
 960         my $storage_class = "DBIx::Class::Storage::DBI::${driver}";
 961         if ($self->load_optional_class($storage_class)) {
 962           mro::set_mro($storage_class, 'c3');
 963           bless $self, $storage_class;
 964           $self->_rebless();
 965         }
 966       }
 967     }
 968
 969     $self->_driver_determined(1);
 970
 971     $self->_init; # run driver-specific initializations
 972
 973     $self->_run_connection_actions
 974         if !$started_connected && defined $self->_dbh;
 975   }
 976 }
 977
 978 sub _do_connection_actions {
 979   my $self          = shift;
 980   my $method_prefix = shift;
 981   my $call          = shift;
 982
 983   if (not ref($call)) {
 984     my $method = $method_prefix . $call;
 985     $self->$method(@_);
 986   } elsif (ref($call) eq 'CODE') {
 987     $self->$call(@_);
 988   } elsif (ref($call) eq 'ARRAY') {
 989     if (ref($call->[0]) ne 'ARRAY') {
 990       $self->_do_connection_actions($method_prefix, $_) for @$call;
 991     } else {
 992       $self->_do_connection_actions($method_prefix, @$_) for @$call;
 993     }
 994   } else {
 995     $self->throw_exception (sprintf ("Don't know how to process conection actions of type '%s'", ref($call)) );
 996   }
 997
 998   return $self;
 999 }
1000
1001 sub connect_call_do_sql {
1002   my $self = shift;
1003   $self->_do_query(@_);
1004 }
1005
1006 sub disconnect_call_do_sql {
1007   my $self = shift;
1008   $self->_do_query(@_);
1009 }
1010
1011 # override in db-specific backend when necessary
1012 sub connect_call_datetime_setup { 1 }
1013
1014 sub _do_query {
1015   my ($self, $action) = @_;
1016
1017   if (ref $action eq 'CODE') {
1018     $action = $action->($self);
1019     $self->_do_query($_) foreach @$action;
1020   }
1021   else {
1022     # Most debuggers expect ($sql, @bind), so we need to exclude
1023     # the attribute hash which is the second argument to $dbh->do
1024     # furthermore the bind values are usually to be presented
1025     # as named arrayref pairs, so wrap those here too
1026     my @do_args = (ref $action eq 'ARRAY') ? (@$action) : ($action);
1027     my $sql = shift @do_args;
1028     my $attrs = shift @do_args;
1029     my @bind = map { [ undef, $_ ] } @do_args;
1030
1031     $self->_query_start($sql, @bind);
1032     $self->_get_dbh->do($sql, $attrs, @do_args);
1033     $self->_query_end($sql, @bind);
1034   }
1035
1036   return $self;
1037 }
1038
1039 sub _connect {
1040   my ($self, @info) = @_;
1041
1042   $self->throw_exception("You failed to provide any connection info")
1043     if !@info;
1044
1045   my ($old_connect_via, $dbh);
1046
1047   if ($INC{'Apache/DBI.pm'} && $ENV{MOD_PERL}) {
1048     $old_connect_via = $DBI::connect_via;
1049     $DBI::connect_via = 'connect';
1050   }
1051
1052   eval {
1053     if(ref $info[0] eq 'CODE') {
1054        $dbh = $info[0]->();
1055     }
1056     else {
1057        $dbh = DBI->connect(@info);
1058     }
1059
1060     if($dbh && !$self->unsafe) {
1061       my $weak_self = $self;
1062       Scalar::Util::weaken($weak_self);
1063       $dbh->{HandleError} = sub {
1064           if ($weak_self) {
1065             $weak_self->throw_exception("DBI Exception: $_[0]");
1066           }
1067           else {
1068             # the handler may be invoked by something totally out of
1069             # the scope of DBIC
1070             croak ("DBI Exception: $_[0]");
1071           }
1072       };
1073       $dbh->{ShowErrorStatement} = 1;
1074       $dbh->{RaiseError} = 1;
1075       $dbh->{PrintError} = 0;
1076     }
1077   };
1078
1079   $DBI::connect_via = $old_connect_via if $old_connect_via;
1080
1081   $self->throw_exception("DBI Connection failed: " . ($@||$DBI::errstr))
1082     if !$dbh || $@;
1083
1084   $self->_dbh_autocommit($dbh->{AutoCommit});
1085
1086   $dbh;
1087 }
1088
1089 sub svp_begin {
1090   my ($self, $name) = @_;
1091
1092   $name = $self->_svp_generate_name
1093     unless defined $name;
1094
1095   $self->throw_exception ("You can't use savepoints outside a transaction")
1096     if $self->{transaction_depth} == 0;
1097
1098   $self->throw_exception ("Your Storage implementation doesn't support savepoints")
1099     unless $self->can('_svp_begin');
1100
1101   push @{ $self->{savepoints} }, $name;
1102
1103   $self->debugobj->svp_begin($name) if $self->debug;
1104
1105   return $self->_svp_begin($name);
1106 }
1107
1108 sub svp_release {
1109   my ($self, $name) = @_;
1110
1111   $self->throw_exception ("You can't use savepoints outside a transaction")
1112     if $self->{transaction_depth} == 0;
1113
1114   $self->throw_exception ("Your Storage implementation doesn't support savepoints")
1115     unless $self->can('_svp_release');
1116
1117   if (defined $name) {
1118     $self->throw_exception ("Savepoint '$name' does not exist")
1119       unless grep { $_ eq $name } @{ $self->{savepoints} };
1120
1121     # Dig through the stack until we find the one we are releasing.  This keeps
1122     # the stack up to date.
1123     my $svp;
1124
1125     do { $svp = pop @{ $self->{savepoints} } } while $svp ne $name;
1126   } else {
1127     $name = pop @{ $self->{savepoints} };
1128   }
1129
1130   $self->debugobj->svp_release($name) if $self->debug;
1131
1132   return $self->_svp_release($name);
1133 }
1134
1135 sub svp_rollback {
1136   my ($self, $name) = @_;
1137
1138   $self->throw_exception ("You can't use savepoints outside a transaction")
1139     if $self->{transaction_depth} == 0;
1140
1141   $self->throw_exception ("Your Storage implementation doesn't support savepoints")
1142     unless $self->can('_svp_rollback');
1143
1144   if (defined $name) {
1145       # If they passed us a name, verify that it exists in the stack
1146       unless(grep({ $_ eq $name } @{ $self->{savepoints} })) {
1147           $self->throw_exception("Savepoint '$name' does not exist!");
1148       }
1149
1150       # Dig through the stack until we find the one we are releasing.  This keeps
1151       # the stack up to date.
1152       while(my $s = pop(@{ $self->{savepoints} })) {
1153           last if($s eq $name);
1154       }
1155       # Add the savepoint back to the stack, as a rollback doesn't remove the
1156       # named savepoint, only everything after it.
1157       push(@{ $self->{savepoints} }, $name);
1158   } else {
1159       # We'll assume they want to rollback to the last savepoint
1160       $name = $self->{savepoints}->[-1];
1161   }
1162
1163   $self->debugobj->svp_rollback($name) if $self->debug;
1164
1165   return $self->_svp_rollback($name);
1166 }
1167
1168 sub _svp_generate_name {
1169     my ($self) = @_;
1170
1171     return 'savepoint_'.scalar(@{ $self->{'savepoints'} });
1172 }
1173
1174 sub txn_begin {
1175   my $self = shift;
1176
1177   # this means we have not yet connected and do not know the AC status
1178   # (e.g. coderef $dbh)
1179   $self->ensure_connected if (! defined $self->_dbh_autocommit);
1180
1181   if($self->{transaction_depth} == 0) {
1182     $self->debugobj->txn_begin()
1183       if $self->debug;
1184     $self->_dbh_begin_work;
1185   }
1186   elsif ($self->auto_savepoint) {
1187     $self->svp_begin;
1188   }
1189   $self->{transaction_depth}++;
1190 }
1191
1192 sub _dbh_begin_work {
1193   my $self = shift;
1194
1195   # if the user is utilizing txn_do - good for him, otherwise we need to
1196   # ensure that the $dbh is healthy on BEGIN.
1197   # We do this via ->dbh_do instead of ->dbh, so that the ->dbh "ping"
1198   # will be replaced by a failure of begin_work itself (which will be
1199   # then retried on reconnect)
1200   if ($self->{_in_dbh_do}) {
1201     $self->_dbh->begin_work;
1202   } else {
1203     $self->dbh_do(sub { $_[1]->begin_work });
1204   }
1205 }
1206
1207 sub txn_commit {
1208   my $self = shift;
1209   if ($self->{transaction_depth} == 1) {
1210     $self->debugobj->txn_commit()
1211       if ($self->debug);
1212     $self->_dbh_commit;
1213     $self->{transaction_depth} = 0
1214       if $self->_dbh_autocommit;
1215   }
1216   elsif($self->{transaction_depth} > 1) {
1217     $self->{transaction_depth}--;
1218     $self->svp_release
1219       if $self->auto_savepoint;
1220   }
1221 }
1222
1223 sub _dbh_commit {
1224   my $self = shift;
1225   my $dbh  = $self->_dbh
1226     or $self->throw_exception('cannot COMMIT on a disconnected handle');
1227   $dbh->commit;
1228 }
1229
1230 sub txn_rollback {
1231   my $self = shift;
1232   my $dbh = $self->_dbh;
1233   eval {
1234     if ($self->{transaction_depth} == 1) {
1235       $self->debugobj->txn_rollback()
1236         if ($self->debug);
1237       $self->{transaction_depth} = 0
1238         if $self->_dbh_autocommit;
1239       $self->_dbh_rollback;
1240     }
1241     elsif($self->{transaction_depth} > 1) {
1242       $self->{transaction_depth}--;
1243       if ($self->auto_savepoint) {
1244         $self->svp_rollback;
1245         $self->svp_release;
1246       }
1247     }
1248     else {
1249       die DBIx::Class::Storage::NESTED_ROLLBACK_EXCEPTION->new;
1250     }
1251   };
1252   if ($@) {
1253     my $error = $@;
1254     my $exception_class = "DBIx::Class::Storage::NESTED_ROLLBACK_EXCEPTION";
1255     $error =~ /$exception_class/ and $self->throw_exception($error);
1256     # ensure that a failed rollback resets the transaction depth
1257     $self->{transaction_depth} = $self->_dbh_autocommit ? 0 : 1;
1258     $self->throw_exception($error);
1259   }
1260 }
1261
1262 sub _dbh_rollback {
1263   my $self = shift;
1264   my $dbh  = $self->_dbh
1265     or $self->throw_exception('cannot ROLLBACK on a disconnected handle');
1266   $dbh->rollback;
1267 }
1268
1269 # This used to be the top-half of _execute.  It was split out to make it
1270 #  easier to override in NoBindVars without duping the rest.  It takes up
1271 #  all of _execute's args, and emits $sql, @bind.
1272 sub _prep_for_execute {
1273   my ($self, $op, $extra_bind, $ident, $args) = @_;
1274
1275   if( Scalar::Util::blessed($ident) && $ident->isa("DBIx::Class::ResultSource") ) {
1276     $ident = $ident->from();
1277   }
1278
1279   my ($sql, @bind) = $self->sql_maker->$op($ident, @$args);
1280
1281   unshift(@bind,
1282     map { ref $_ eq 'ARRAY' ? $_ : [ '!!dummy', $_ ] } @$extra_bind)
1283       if $extra_bind;
1284   return ($sql, \@bind);
1285 }
1286
1287
1288 sub _fix_bind_params {
1289     my ($self, @bind) = @_;
1290
1291     ### Turn @bind from something like this:
1292     ###   ( [ "artist", 1 ], [ "cdid", 1, 3 ] )
1293     ### to this:
1294     ###   ( "'1'", "'1'", "'3'" )
1295     return
1296         map {
1297             if ( defined( $_ && $_->[1] ) ) {
1298                 map { qq{'$_'}; } @{$_}[ 1 .. $#$_ ];
1299             }
1300             else { q{'NULL'}; }
1301         } @bind;
1302 }
1303
1304 sub _query_start {
1305     my ( $self, $sql, @bind ) = @_;
1306
1307     if ( $self->debug ) {
1308         @bind = $self->_fix_bind_params(@bind);
1309
1310         $self->debugobj->query_start( $sql, @bind );
1311     }
1312 }
1313
1314 sub _query_end {
1315     my ( $self, $sql, @bind ) = @_;
1316
1317     if ( $self->debug ) {
1318         @bind = $self->_fix_bind_params(@bind);
1319         $self->debugobj->query_end( $sql, @bind );
1320     }
1321 }
1322
1323 sub _dbh_execute {
1324   my ($self, $dbh, $op, $extra_bind, $ident, $bind_attributes, @args) = @_;
1325
1326   my ($sql, $bind) = $self->_prep_for_execute($op, $extra_bind, $ident, \@args);
1327
1328   $self->_query_start( $sql, @$bind );
1329
1330   my $sth = $self->sth($sql,$op);
1331
1332   my $placeholder_index = 1;
1333
1334   foreach my $bound (@$bind) {
1335     my $attributes = {};
1336     my($column_name, @data) = @$bound;
1337
1338     if ($bind_attributes) {
1339       $attributes = $bind_attributes->{$column_name}
1340       if defined $bind_attributes->{$column_name};
1341     }
1342
1343     foreach my $data (@data) {
1344       my $ref = ref $data;
1345       $data = $ref && $ref ne 'ARRAY' ? ''.$data : $data; # stringify args (except arrayrefs)
1346
1347       $sth->bind_param($placeholder_index, $data, $attributes);
1348       $placeholder_index++;
1349     }
1350   }
1351
1352   # Can this fail without throwing an exception anyways???
1353   my $rv = $sth->execute();
1354   $self->throw_exception($sth->errstr) if !$rv;
1355
1356   $self->_query_end( $sql, @$bind );
1357
1358   return (wantarray ? ($rv, $sth, @$bind) : $rv);
1359 }
1360
1361 sub _execute {
1362     my $self = shift;
1363     $self->dbh_do('_dbh_execute', @_);  # retry over disconnects
1364 }
1365
1366 sub insert {
1367   my ($self, $source, $to_insert) = @_;
1368
1369   my $ident = $source->from;
1370   my $bind_attributes = $self->source_bind_attributes($source);
1371
1372   my $updated_cols = {};
1373
1374   foreach my $col ( $source->columns ) {
1375     if ( !defined $to_insert->{$col} ) {
1376       my $col_info = $source->column_info($col);
1377
1378       if ( $col_info->{auto_nextval} ) {
1379         $updated_cols->{$col} = $to_insert->{$col} = $self->_sequence_fetch(
1380           'nextval',
1381           $col_info->{sequence} ||=
1382             $self->_dbh_get_autoinc_seq($self->_get_dbh, $source, $col)
1383         );
1384       }
1385     }
1386   }
1387
1388   $self->_execute('insert' => [], $source, $bind_attributes, $to_insert);
1389
1390   return $updated_cols;
1391 }
1392
1393 ## Currently it is assumed that all values passed will be "normal", i.e. not
1394 ## scalar refs, or at least, all the same type as the first set, the statement is
1395 ## only prepped once.
1396 sub insert_bulk {
1397   my ($self, $source, $cols, $data) = @_;
1398
1399   my %colvalues;
1400   @colvalues{@$cols} = (0..$#$cols);
1401
1402   for my $i (0..$#$cols) {
1403     my $first_val = $data->[0][$i];
1404     next unless ref $first_val eq 'SCALAR';
1405
1406     $colvalues{ $cols->[$i] } = $first_val;
1407   }
1408
1409   # check for bad data and stringify stringifiable objects
1410   my $bad_slice = sub {
1411     my ($msg, $col_idx, $slice_idx) = @_;
1412     $self->throw_exception(sprintf "%s for column '%s' in populate slice:\n%s",
1413       $msg,
1414       $cols->[$col_idx],
1415       do {
1416         local $Data::Dumper::Maxdepth = 1; # don't dump objects, if any
1417         Data::Dumper::Concise::Dumper({
1418           map { $cols->[$_] => $data->[$slice_idx][$_] } (0 .. $#$cols)
1419         }),
1420       }
1421     );
1422   };
1423
1424   for my $datum_idx (0..$#$data) {
1425     my $datum = $data->[$datum_idx];
1426
1427     for my $col_idx (0..$#$cols) {
1428       my $val            = $datum->[$col_idx];
1429       my $sqla_bind      = $colvalues{ $cols->[$col_idx] };
1430       my $is_literal_sql = (ref $sqla_bind) eq 'SCALAR';
1431
1432       if ($is_literal_sql) {
1433         if (not ref $val) {
1434           $bad_slice->('bind found where literal SQL expected', $col_idx, $datum_idx);
1435         }
1436         elsif ((my $reftype = ref $val) ne 'SCALAR') {
1437           $bad_slice->("$reftype reference found where literal SQL expected",
1438             $col_idx, $datum_idx);
1439         }
1440         elsif ($$val ne $$sqla_bind){
1441           $bad_slice->("inconsistent literal SQL value, expecting: '$$sqla_bind'",
1442             $col_idx, $datum_idx);
1443         }
1444       }
1445       elsif (my $reftype = ref $val) {
1446         require overload;
1447         if (overload::Method($val, '""')) {
1448           $datum->[$col_idx] = "".$val;
1449         }
1450         else {
1451           $bad_slice->("$reftype reference found where bind expected",
1452             $col_idx, $datum_idx);
1453         }
1454       }
1455     }
1456   }
1457
1458   my ($sql, $bind) = $self->_prep_for_execute (
1459     'insert', undef, $source, [\%colvalues]
1460   );
1461   my @bind = @$bind;
1462
1463   my $empty_bind = 1 if (not @bind) &&
1464     (grep { ref $_ eq 'SCALAR' } values %colvalues) == @$cols;
1465
1466   if ((not @bind) && (not $empty_bind)) {
1467     $self->throw_exception(
1468       'Cannot insert_bulk without support for placeholders'
1469     );
1470   }
1471
1472   # neither _execute_array, nor _execute_inserts_with_no_binds are
1473   # atomic (even if _execute _array is a single call). Thus a safety
1474   # scope guard
1475   my $guard = $self->txn_scope_guard;
1476
1477   $self->_query_start( $sql, ['__BULK__'] );
1478   my $sth = $self->sth($sql);
1479   my $rv = do {
1480     if ($empty_bind) {
1481       # bind_param_array doesn't work if there are no binds
1482       $self->_dbh_execute_inserts_with_no_binds( $sth, scalar @$data );
1483     }
1484     else {
1485 #      @bind = map { ref $_ ? ''.$_ : $_ } @bind; # stringify args
1486       $self->_execute_array( $source, $sth, \@bind, $cols, $data );
1487     }
1488   };
1489
1490   $self->_query_end( $sql, ['__BULK__'] );
1491
1492   $guard->commit;
1493
1494   return (wantarray ? ($rv, $sth, @bind) : $rv);
1495 }
1496
1497 sub _execute_array {
1498   my ($self, $source, $sth, $bind, $cols, $data, @extra) = @_;
1499
1500   ## This must be an arrayref, else nothing works!
1501   my $tuple_status = [];
1502
1503   ## Get the bind_attributes, if any exist
1504   my $bind_attributes = $self->source_bind_attributes($source);
1505
1506   ## Bind the values and execute
1507   my $placeholder_index = 1;
1508
1509   foreach my $bound (@$bind) {
1510
1511     my $attributes = {};
1512     my ($column_name, $data_index) = @$bound;
1513
1514     if( $bind_attributes ) {
1515       $attributes = $bind_attributes->{$column_name}
1516       if defined $bind_attributes->{$column_name};
1517     }
1518
1519     my @data = map { $_->[$data_index] } @$data;
1520
1521     $sth->bind_param_array(
1522       $placeholder_index,
1523       [@data],
1524       (%$attributes ?  $attributes : ()),
1525     );
1526     $placeholder_index++;
1527   }
1528
1529   my $rv = eval {
1530     $self->_dbh_execute_array($sth, $tuple_status, @extra);
1531   };
1532   my $err = $@ || $sth->errstr;
1533
1534 # Statement must finish even if there was an exception.
1535   eval { $sth->finish };
1536   $err = $@ unless $err;
1537
1538   if ($err) {
1539     my $i = 0;
1540     ++$i while $i <= $#$tuple_status && !ref $tuple_status->[$i];
1541
1542     $self->throw_exception("Unexpected populate error: $err")
1543       if ($i > $#$tuple_status);
1544
1545     $self->throw_exception(sprintf "%s for populate slice:\n%s",
1546       ($tuple_status->[$i][1] || $err),
1547       Data::Dumper::Concise::Dumper({
1548         map { $cols->[$_] => $data->[$i][$_] } (0 .. $#$cols)
1549       }),
1550     );
1551   }
1552   return $rv;
1553 }
1554
1555 sub _dbh_execute_array {
1556     my ($self, $sth, $tuple_status, @extra) = @_;
1557
1558     return $sth->execute_array({ArrayTupleStatus => $tuple_status});
1559 }
1560
1561 sub _dbh_execute_inserts_with_no_binds {
1562   my ($self, $sth, $count) = @_;
1563
1564   eval {
1565     my $dbh = $self->_get_dbh;
1566     local $dbh->{RaiseError} = 1;
1567     local $dbh->{PrintError} = 0;
1568
1569     $sth->execute foreach 1..$count;
1570   };
1571   my $exception = $@;
1572
1573 # Make sure statement is finished even if there was an exception.
1574   eval { $sth->finish };
1575   $exception = $@ unless $exception;
1576
1577   $self->throw_exception($exception) if $exception;
1578
1579   return $count;
1580 }
1581
1582 sub update {
1583   my ($self, $source, @args) = @_;
1584
1585   my $bind_attrs = $self->source_bind_attributes($source);
1586
1587   return $self->_execute('update' => [], $source, $bind_attrs, @args);
1588 }
1589
1590
1591 sub delete {
1592   my ($self, $source, @args) = @_;
1593
1594   my $bind_attrs = $self->source_bind_attributes($source);
1595
1596   return $self->_execute('delete' => [], $source, $bind_attrs, @args);
1597 }
1598
1599 # We were sent here because the $rs contains a complex search
1600 # which will require a subquery to select the correct rows
1601 # (i.e. joined or limited resultsets, or non-introspectable conditions)
1602 #
1603 # Generating a single PK column subquery is trivial and supported
1604 # by all RDBMS. However if we have a multicolumn PK, things get ugly.
1605 # Look at _multipk_update_delete()
1606 sub _subq_update_delete {
1607   my $self = shift;
1608   my ($rs, $op, $values) = @_;
1609
1610   my $rsrc = $rs->result_source;
1611
1612   # quick check if we got a sane rs on our hands
1613   my @pcols = $rsrc->_pri_cols;
1614
1615   my $sel = $rs->_resolved_attrs->{select};
1616   $sel = [ $sel ] unless ref $sel eq 'ARRAY';
1617
1618   if (
1619       join ("\x00", map { join '.', $rs->{attrs}{alias}, $_ } sort @pcols)
1620         ne
1621       join ("\x00", sort @$sel )
1622   ) {
1623     $self->throw_exception (
1624       '_subq_update_delete can not be called on resultsets selecting columns other than the primary keys'
1625     );
1626   }
1627
1628   if (@pcols == 1) {
1629     return $self->$op (
1630       $rsrc,
1631       $op eq 'update' ? $values : (),
1632       { $pcols[0] => { -in => $rs->as_query } },
1633     );
1634   }
1635
1636   else {
1637     return $self->_multipk_update_delete (@_);
1638   }
1639 }
1640
1641 # ANSI SQL does not provide a reliable way to perform a multicol-PK
1642 # resultset update/delete involving subqueries. So by default resort
1643 # to simple (and inefficient) delete_all style per-row opearations,
1644 # while allowing specific storages to override this with a faster
1645 # implementation.
1646 #
1647 sub _multipk_update_delete {
1648   return shift->_per_row_update_delete (@_);
1649 }
1650
1651 # This is the default loop used to delete/update rows for multi PK
1652 # resultsets, and used by mysql exclusively (because it can't do anything
1653 # else).
1654 #
1655 # We do not use $row->$op style queries, because resultset update/delete
1656 # is not expected to cascade (this is what delete_all/update_all is for).
1657 #
1658 # There should be no race conditions as the entire operation is rolled
1659 # in a transaction.
1660 #
1661 sub _per_row_update_delete {
1662   my $self = shift;
1663   my ($rs, $op, $values) = @_;
1664
1665   my $rsrc = $rs->result_source;
1666   my @pcols = $rsrc->_pri_cols;
1667
1668   my $guard = $self->txn_scope_guard;
1669
1670   # emulate the return value of $sth->execute for non-selects
1671   my $row_cnt = '0E0';
1672
1673   my $subrs_cur = $rs->cursor;
1674   my @all_pk = $subrs_cur->all;
1675   for my $pks ( @all_pk) {
1676
1677     my $cond;
1678     for my $i (0.. $#pcols) {
1679       $cond->{$pcols[$i]} = $pks->[$i];
1680     }
1681
1682     $self->$op (
1683       $rsrc,
1684       $op eq 'update' ? $values : (),
1685       $cond,
1686     );
1687
1688     $row_cnt++;
1689   }
1690
1691   $guard->commit;
1692
1693   return $row_cnt;
1694 }
1695
1696 sub _select {
1697   my $self = shift;
1698
1699   # localization is neccessary as
1700   # 1) there is no infrastructure to pass this around before SQLA2
1701   # 2) _select_args sets it and _prep_for_execute consumes it
1702   my $sql_maker = $self->sql_maker;
1703   local $sql_maker->{_dbic_rs_attrs};
1704
1705   return $self->_execute($self->_select_args(@_));
1706 }
1707
1708 sub _select_args_to_query {
1709   my $self = shift;
1710
1711   # localization is neccessary as
1712   # 1) there is no infrastructure to pass this around before SQLA2
1713   # 2) _select_args sets it and _prep_for_execute consumes it
1714   my $sql_maker = $self->sql_maker;
1715   local $sql_maker->{_dbic_rs_attrs};
1716
1717   # my ($op, $bind, $ident, $bind_attrs, $select, $cond, $order, $rows, $offset)
1718   #  = $self->_select_args($ident, $select, $cond, $attrs);
1719   my ($op, $bind, $ident, $bind_attrs, @args) =
1720     $self->_select_args(@_);
1721
1722   # my ($sql, $prepared_bind) = $self->_prep_for_execute($op, $bind, $ident, [ $select, $cond, $order, $rows, $offset ]);
1723   my ($sql, $prepared_bind) = $self->_prep_for_execute($op, $bind, $ident, \@args);
1724   $prepared_bind ||= [];
1725
1726   return wantarray
1727     ? ($sql, $prepared_bind, $bind_attrs)
1728     : \[ "($sql)", @$prepared_bind ]
1729   ;
1730 }
1731
1732 sub _select_args {
1733   my ($self, $ident, $select, $where, $attrs) = @_;
1734
1735   my ($alias2source, $rs_alias) = $self->_resolve_ident_sources ($ident);
1736
1737   my $sql_maker = $self->sql_maker;
1738   $sql_maker->{_dbic_rs_attrs} = {
1739     %$attrs,
1740     select => $select,
1741     from => $ident,
1742     where => $where,
1743     $rs_alias && $alias2source->{$rs_alias}
1744       ? ( _source_handle => $alias2source->{$rs_alias}->handle )
1745       : ()
1746     ,
1747   };
1748
1749   # calculate bind_attrs before possible $ident mangling
1750   my $bind_attrs = {};
1751   for my $alias (keys %$alias2source) {
1752     my $bindtypes = $self->source_bind_attributes ($alias2source->{$alias}) || {};
1753     for my $col (keys %$bindtypes) {
1754
1755       my $fqcn = join ('.', $alias, $col);
1756       $bind_attrs->{$fqcn} = $bindtypes->{$col} if $bindtypes->{$col};
1757
1758       # Unqialified column names are nice, but at the same time can be
1759       # rather ambiguous. What we do here is basically go along with
1760       # the loop, adding an unqualified column slot to $bind_attrs,
1761       # alongside the fully qualified name. As soon as we encounter
1762       # another column by that name (which would imply another table)
1763       # we unset the unqualified slot and never add any info to it
1764       # to avoid erroneous type binding. If this happens the users
1765       # only choice will be to fully qualify his column name
1766
1767       if (exists $bind_attrs->{$col}) {
1768         $bind_attrs->{$col} = {};
1769       }
1770       else {
1771         $bind_attrs->{$col} = $bind_attrs->{$fqcn};
1772       }
1773     }
1774   }
1775
1776   # adjust limits
1777   if (
1778     $attrs->{software_limit}
1779       ||
1780     $sql_maker->_default_limit_syntax eq "GenericSubQ"
1781   ) {
1782     $attrs->{software_limit} = 1;
1783   }
1784   else {
1785     $self->throw_exception("rows attribute must be positive if present")
1786       if (defined($attrs->{rows}) && !($attrs->{rows} > 0));
1787
1788     # MySQL actually recommends this approach.  I cringe.
1789     $attrs->{rows} = 2**48 if not defined $attrs->{rows} and defined $attrs->{offset};
1790   }
1791
1792   my @limit;
1793
1794   # see if we need to tear the prefetch apart otherwise delegate the limiting to the
1795   # storage, unless software limit was requested
1796   if (
1797     #limited has_many
1798     ( $attrs->{rows} && keys %{$attrs->{collapse}} )
1799        ||
1800     # limited prefetch with RNO subqueries
1801     (
1802       $attrs->{rows}
1803         &&
1804       $sql_maker->limit_dialect eq 'RowNumberOver'
1805         &&
1806       $attrs->{_prefetch_select}
1807         &&
1808       @{$attrs->{_prefetch_select}}
1809     )
1810       ||
1811     # grouped prefetch
1812     ( $attrs->{group_by}
1813         &&
1814       @{$attrs->{group_by}}
1815         &&
1816       $attrs->{_prefetch_select}
1817         &&
1818       @{$attrs->{_prefetch_select}}
1819     )
1820   ) {
1821     ($ident, $select, $where, $attrs)
1822       = $self->_adjust_select_args_for_complex_prefetch ($ident, $select, $where, $attrs);
1823   }
1824
1825   elsif (
1826     ($attrs->{rows} || $attrs->{offset})
1827       &&
1828     $sql_maker->limit_dialect eq 'RowNumberOver'
1829       &&
1830     (ref $ident eq 'ARRAY' && @$ident > 1)  # indicates a join
1831       &&
1832     scalar $self->_parse_order_by ($attrs->{order_by})
1833   ) {
1834     # the RNO limit dialect above mangles the SQL such that the join gets lost
1835     # wrap a subquery here
1836
1837     push @limit, delete @{$attrs}{qw/rows offset/};
1838
1839     my $subq = $self->_select_args_to_query (
1840       $ident,
1841       $select,
1842       $where,
1843       $attrs,
1844     );
1845
1846     $ident = {
1847       -alias => $attrs->{alias},
1848       -source_handle => $ident->[0]{-source_handle},
1849       $attrs->{alias} => $subq,
1850     };
1851
1852     # all part of the subquery now
1853     delete @{$attrs}{qw/order_by group_by having/};
1854     $where = undef;
1855   }
1856
1857   elsif (! $attrs->{software_limit} ) {
1858     push @limit, $attrs->{rows}, $attrs->{offset};
1859   }
1860
1861   # try to simplify the joinmap further (prune unreferenced type-single joins)
1862   $ident = $self->_prune_unused_joins ($ident, $select, $where, $attrs);
1863
1864 ###
1865   # This would be the point to deflate anything found in $where
1866   # (and leave $attrs->{bind} intact). Problem is - inflators historically
1867   # expect a row object. And all we have is a resultsource (it is trivial
1868   # to extract deflator coderefs via $alias2source above).
1869   #
1870   # I don't see a way forward other than changing the way deflators are
1871   # invoked, and that's just bad...
1872 ###
1873
1874   my $order = { map
1875     { $attrs->{$_} ? ( $_ => $attrs->{$_} ) : ()  }
1876     (qw/order_by group_by having/ )
1877   };
1878
1879   return ('select', $attrs->{bind}, $ident, $bind_attrs, $select, $where, $order, @limit);
1880 }
1881
1882 # Returns a counting SELECT for a simple count
1883 # query. Abstracted so that a storage could override
1884 # this to { count => 'firstcol' } or whatever makes
1885 # sense as a performance optimization
1886 sub _count_select {
1887   #my ($self, $source, $rs_attrs) = @_;
1888   return { count => '*' };
1889 }
1890
1891 # Returns a SELECT which will end up in the subselect
1892 # There may or may not be a group_by, as the subquery
1893 # might have been called to accomodate a limit
1894 #
1895 # Most databases would be happy with whatever ends up
1896 # here, but some choke in various ways.
1897 #
1898 sub _subq_count_select {
1899   my ($self, $source, $rs_attrs) = @_;
1900
1901   if (my $groupby = $rs_attrs->{group_by}) {
1902
1903     my $avail_columns = $self->_resolve_column_info ($rs_attrs->{from});
1904
1905     my $sel_index;
1906     for my $sel (@{$rs_attrs->{select}}) {
1907       if (ref $sel eq 'HASH' and $sel->{-as}) {
1908         $sel_index->{$sel->{-as}} = $sel;
1909       }
1910     }
1911
1912     my @selection;
1913     for my $g_part (@$groupby) {
1914       if (ref $g_part or $avail_columns->{$g_part}) {
1915         push @selection, $g_part;
1916       }
1917       elsif ($sel_index->{$g_part}) {
1918         push @selection, $sel_index->{$g_part};
1919       }
1920       else {
1921         $self->throw_exception ("group_by criteria '$g_part' not contained within current resultset source(s)");
1922       }
1923     }
1924
1925     return \@selection;
1926   }
1927
1928   my @pcols = map { join '.', $rs_attrs->{alias}, $_ } ($source->primary_columns);
1929   return @pcols ? \@pcols : [ 1 ];
1930 }
1931
1932 sub source_bind_attributes {
1933   my ($self, $source) = @_;
1934
1935   my $bind_attributes;
1936   foreach my $column ($source->columns) {
1937
1938     my $data_type = $source->column_info($column)->{data_type} || '';
1939     $bind_attributes->{$column} = $self->bind_attribute_by_data_type($data_type)
1940      if $data_type;
1941   }
1942
1943   return $bind_attributes;
1944 }
1945
1946 =head2 select
1947
1948 =over 4
1949
1950 =item Arguments: $ident, $select, $condition, $attrs
1951
1952 =back
1953
1954 Handle a SQL select statement.
1955
1956 =cut
1957
1958 sub select {
1959   my $self = shift;
1960   my ($ident, $select, $condition, $attrs) = @_;
1961   return $self->cursor_class->new($self, \@_, $attrs);
1962 }
1963
1964 sub select_single {
1965   my $self = shift;
1966   my ($rv, $sth, @bind) = $self->_select(@_);
1967   my @row = $sth->fetchrow_array;
1968   my @nextrow = $sth->fetchrow_array if @row;
1969   if(@row && @nextrow) {
1970     carp "Query returned more than one row.  SQL that returns multiple rows is DEPRECATED for ->find and ->single";
1971   }
1972   # Need to call finish() to work round broken DBDs
1973   $sth->finish();
1974   return @row;
1975 }
1976
1977 =head2 sth
1978
1979 =over 4
1980
1981 =item Arguments: $sql
1982
1983 =back
1984
1985 Returns a L<DBI> sth (statement handle) for the supplied SQL.
1986
1987 =cut
1988
1989 sub _dbh_sth {
1990   my ($self, $dbh, $sql) = @_;
1991
1992   # 3 is the if_active parameter which avoids active sth re-use
1993   my $sth = $self->disable_sth_caching
1994     ? $dbh->prepare($sql)
1995     : $dbh->prepare_cached($sql, {}, 3);
1996
1997   # XXX You would think RaiseError would make this impossible,
1998   #  but apparently that's not true :(
1999   $self->throw_exception($dbh->errstr) if !$sth;
2000
2001   $sth;
2002 }
2003
2004 sub sth {
2005   my ($self, $sql) = @_;
2006   $self->dbh_do('_dbh_sth', $sql);  # retry over disconnects
2007 }
2008
2009 sub _dbh_columns_info_for {
2010   my ($self, $dbh, $table) = @_;
2011
2012   if ($dbh->can('column_info')) {
2013     my %result;
2014     eval {
2015       my ($schema,$tab) = $table =~ /^(.+?)\.(.+)$/ ? ($1,$2) : (undef,$table);
2016       my $sth = $dbh->column_info( undef,$schema, $tab, '%' );
2017       $sth->execute();
2018       while ( my $info = $sth->fetchrow_hashref() ){
2019         my %column_info;
2020         $column_info{data_type}   = $info->{TYPE_NAME};
2021         $column_info{size}      = $info->{COLUMN_SIZE};
2022         $column_info{is_nullable}   = $info->{NULLABLE} ? 1 : 0;
2023         $column_info{default_value} = $info->{COLUMN_DEF};
2024         my $col_name = $info->{COLUMN_NAME};
2025         $col_name =~ s/^\"(.*)\"$/$1/;
2026
2027         $result{$col_name} = \%column_info;
2028       }
2029     };
2030     return \%result if !$@ && scalar keys %result;
2031   }
2032
2033   my %result;
2034   my $sth = $dbh->prepare($self->sql_maker->select($table, undef, \'1 = 0'));
2035   $sth->execute;
2036   my @columns = @{$sth->{NAME_lc}};
2037   for my $i ( 0 .. $#columns ){
2038     my %column_info;
2039     $column_info{data_type} = $sth->{TYPE}->[$i];
2040     $column_info{size} = $sth->{PRECISION}->[$i];
2041     $column_info{is_nullable} = $sth->{NULLABLE}->[$i] ? 1 : 0;
2042
2043     if ($column_info{data_type} =~ m/^(.*?)\((.*?)\)$/) {
2044       $column_info{data_type} = $1;
2045       $column_info{size}    = $2;
2046     }
2047
2048     $result{$columns[$i]} = \%column_info;
2049   }
2050   $sth->finish;
2051
2052   foreach my $col (keys %result) {
2053     my $colinfo = $result{$col};
2054     my $type_num = $colinfo->{data_type};
2055     my $type_name;
2056     if(defined $type_num && $dbh->can('type_info')) {
2057       my $type_info = $dbh->type_info($type_num);
2058       $type_name = $type_info->{TYPE_NAME} if $type_info;
2059       $colinfo->{data_type} = $type_name if $type_name;
2060     }
2061   }
2062
2063   return \%result;
2064 }
2065
2066 sub columns_info_for {
2067   my ($self, $table) = @_;
2068   $self->_dbh_columns_info_for ($self->_get_dbh, $table);
2069 }
2070
2071 =head2 last_insert_id
2072
2073 Return the row id of the last insert.
2074
2075 =cut
2076
2077 sub _dbh_last_insert_id {
2078     my ($self, $dbh, $source, $col) = @_;
2079
2080     my $id = eval { $dbh->last_insert_id (undef, undef, $source->name, $col) };
2081
2082     return $id if defined $id;
2083
2084     my $class = ref $self;
2085     $self->throw_exception ("No storage specific _dbh_last_insert_id() method implemented in $class, and the generic DBI::last_insert_id() failed");
2086 }
2087
2088 sub last_insert_id {
2089   my $self = shift;
2090   $self->_dbh_last_insert_id ($self->_dbh, @_);
2091 }
2092
2093 =head2 _native_data_type
2094
2095 =over 4
2096
2097 =item Arguments: $type_name
2098
2099 =back
2100
2101 This API is B<EXPERIMENTAL>, will almost definitely change in the future, and
2102 currently only used by L<::AutoCast|DBIx::Class::Storage::DBI::AutoCast> and
2103 L<::Sybase::ASE|DBIx::Class::Storage::DBI::Sybase::ASE>.
2104
2105 The default implementation returns C<undef>, implement in your Storage driver if
2106 you need this functionality.
2107
2108 Should map types from other databases to the native RDBMS type, for example
2109 C<VARCHAR2> to C<VARCHAR>.
2110
2111 Types with modifiers should map to the underlying data type. For example,
2112 C<INTEGER AUTO_INCREMENT> should become C<INTEGER>.
2113
2114 Composite types should map to the container type, for example
2115 C<ENUM(foo,bar,baz)> becomes C<ENUM>.
2116
2117 =cut
2118
2119 sub _native_data_type {
2120   #my ($self, $data_type) = @_;
2121   return undef
2122 }
2123
2124 # Check if placeholders are supported at all
2125 sub _placeholders_supported {
2126   my $self = shift;
2127   my $dbh  = $self->_get_dbh;
2128
2129   # some drivers provide a $dbh attribute (e.g. Sybase and $dbh->{syb_dynamic_supported})
2130   # but it is inaccurate more often than not
2131   eval {
2132     local $dbh->{PrintError} = 0;
2133     local $dbh->{RaiseError} = 1;
2134     $dbh->do('select ?', {}, 1);
2135   };
2136   return $@ ? 0 : 1;
2137 }
2138
2139 # Check if placeholders bound to non-string types throw exceptions
2140 #
2141 sub _typeless_placeholders_supported {
2142   my $self = shift;
2143   my $dbh  = $self->_get_dbh;
2144
2145   eval {
2146     local $dbh->{PrintError} = 0;
2147     local $dbh->{RaiseError} = 1;
2148     # this specifically tests a bind that is NOT a string
2149     $dbh->do('select 1 where 1 = ?', {}, 1);
2150   };
2151   return $@ ? 0 : 1;
2152 }
2153
2154 =head2 sqlt_type
2155
2156 Returns the database driver name.
2157
2158 =cut
2159
2160 sub sqlt_type {
2161   shift->_get_dbh->{Driver}->{Name};
2162 }
2163
2164 =head2 bind_attribute_by_data_type
2165
2166 Given a datatype from column info, returns a database specific bind
2167 attribute for C<< $dbh->bind_param($val,$attribute) >> or nothing if we will
2168 let the database planner just handle it.
2169
2170 Generally only needed for special case column types, like bytea in postgres.
2171
2172 =cut
2173
2174 sub bind_attribute_by_data_type {
2175     return;
2176 }
2177
2178 =head2 is_datatype_numeric
2179
2180 Given a datatype from column_info, returns a boolean value indicating if
2181 the current RDBMS considers it a numeric value. This controls how
2182 L<DBIx::Class::Row/set_column> decides whether to mark the column as
2183 dirty - when the datatype is deemed numeric a C<< != >> comparison will
2184 be performed instead of the usual C<eq>.
2185
2186 =cut
2187
2188 sub is_datatype_numeric {
2189   my ($self, $dt) = @_;
2190
2191   return 0 unless $dt;
2192
2193   return $dt =~ /^ (?:
2194     numeric | int(?:eger)? | (?:tiny|small|medium|big)int | dec(?:imal)? | real | float | double (?: \s+ precision)? | (?:big)?serial
2195   ) $/ix;
2196 }
2197
2198
2199 =head2 create_ddl_dir
2200
2201 =over 4
2202
2203 =item Arguments: $schema \@databases, $version, $directory, $preversion, \%sqlt_args
2204
2205 =back
2206
2207 Creates a SQL file based on the Schema, for each of the specified
2208 database engines in C<\@databases> in the given directory.
2209 (note: specify L<SQL::Translator> names, not L<DBI> driver names).
2210
2211 Given a previous version number, this will also create a file containing
2212 the ALTER TABLE statements to transform the previous schema into the
2213 current one. Note that these statements may contain C<DROP TABLE> or
2214 C<DROP COLUMN> statements that can potentially destroy data.
2215
2216 The file names are created using the C<ddl_filename> method below, please
2217 override this method in your schema if you would like a different file
2218 name format. For the ALTER file, the same format is used, replacing
2219 $version in the name with "$preversion-$version".
2220
2221 See L<SQL::Translator/METHODS> for a list of values for C<\%sqlt_args>.
2222 The most common value for this would be C<< { add_drop_table => 1 } >>
2223 to have the SQL produced include a C<DROP TABLE> statement for each table
2224 created. For quoting purposes supply C<quote_table_names> and
2225 C<quote_field_names>.
2226
2227 If no arguments are passed, then the following default values are assumed:
2228
2229 =over 4
2230
2231 =item databases  - ['MySQL', 'SQLite', 'PostgreSQL']
2232
2233 =item version    - $schema->schema_version
2234
2235 =item directory  - './'
2236
2237 =item preversion - <none>
2238
2239 =back
2240
2241 By default, C<\%sqlt_args> will have
2242
2243  { add_drop_table => 1, ignore_constraint_names => 1, ignore_index_names => 1 }
2244
2245 merged with the hash passed in. To disable any of those features, pass in a
2246 hashref like the following
2247
2248  { ignore_constraint_names => 0, # ... other options }
2249
2250
2251 WARNING: You are strongly advised to check all SQL files created, before applying
2252 them.
2253
2254 =cut
2255
2256 sub create_ddl_dir {
2257   my ($self, $schema, $databases, $version, $dir, $preversion, $sqltargs) = @_;
2258
2259   unless ($dir) {
2260     carp "No directory given, using ./\n";
2261     $dir = './';
2262   }
2263
2264   $self->throw_exception ("Directory '$dir' does not exist\n") unless(-d $dir);
2265
2266   $databases ||= ['MySQL', 'SQLite', 'PostgreSQL'];
2267   $databases = [ $databases ] if(ref($databases) ne 'ARRAY');
2268
2269   my $schema_version = $schema->schema_version || '1.x';
2270   $version ||= $schema_version;
2271
2272   $sqltargs = {
2273     add_drop_table => 1,
2274     ignore_constraint_names => 1,
2275     ignore_index_names => 1,
2276     %{$sqltargs || {}}
2277   };
2278
2279   unless (DBIx::Class::Optional::Dependencies->req_ok_for ('deploy')) {
2280     $self->throw_exception("Can't create a ddl file without " . DBIx::Class::Optional::Dependencies->req_missing_for ('deploy') );
2281   }
2282
2283   my $sqlt = SQL::Translator->new( $sqltargs );
2284
2285   $sqlt->parser('SQL::Translator::Parser::DBIx::Class');
2286   my $sqlt_schema = $sqlt->translate({ data => $schema })
2287     or $self->throw_exception ($sqlt->error);
2288
2289   foreach my $db (@$databases) {
2290     $sqlt->reset();
2291     $sqlt->{schema} = $sqlt_schema;
2292     $sqlt->producer($db);
2293
2294     my $file;
2295     my $filename = $schema->ddl_filename($db, $version, $dir);
2296     if (-e $filename && ($version eq $schema_version )) {
2297       # if we are dumping the current version, overwrite the DDL
2298       carp "Overwriting existing DDL file - $filename";
2299       unlink($filename);
2300     }
2301
2302     my $output = $sqlt->translate;
2303     if(!$output) {
2304       carp("Failed to translate to $db, skipping. (" . $sqlt->error . ")");
2305       next;
2306     }
2307     if(!open($file, ">$filename")) {
2308       $self->throw_exception("Can't open $filename for writing ($!)");
2309       next;
2310     }
2311     print $file $output;
2312     close($file);
2313
2314     next unless ($preversion);
2315
2316     require SQL::Translator::Diff;
2317
2318     my $prefilename = $schema->ddl_filename($db, $preversion, $dir);
2319     if(!-e $prefilename) {
2320       carp("No previous schema file found ($prefilename)");
2321       next;
2322     }
2323
2324     my $difffile = $schema->ddl_filename($db, $version, $dir, $preversion);
2325     if(-e $difffile) {
2326       carp("Overwriting existing diff file - $difffile");
2327       unlink($difffile);
2328     }
2329
2330     my $source_schema;
2331     {
2332       my $t = SQL::Translator->new($sqltargs);
2333       $t->debug( 0 );
2334       $t->trace( 0 );
2335
2336       $t->parser( $db )
2337         or $self->throw_exception ($t->error);
2338
2339       my $out = $t->translate( $prefilename )
2340         or $self->throw_exception ($t->error);
2341
2342       $source_schema = $t->schema;
2343
2344       $source_schema->name( $prefilename )
2345         unless ( $source_schema->name );
2346     }
2347
2348     # The "new" style of producers have sane normalization and can support
2349     # diffing a SQL file against a DBIC->SQLT schema. Old style ones don't
2350     # And we have to diff parsed SQL against parsed SQL.
2351     my $dest_schema = $sqlt_schema;
2352
2353     unless ( "SQL::Translator::Producer::$db"->can('preprocess_schema') ) {
2354       my $t = SQL::Translator->new($sqltargs);
2355       $t->debug( 0 );
2356       $t->trace( 0 );
2357
2358       $t->parser( $db )
2359         or $self->throw_exception ($t->error);
2360
2361       my $out = $t->translate( $filename )
2362         or $self->throw_exception ($t->error);
2363
2364       $dest_schema = $t->schema;
2365
2366       $dest_schema->name( $filename )
2367         unless $dest_schema->name;
2368     }
2369
2370     my $diff = SQL::Translator::Diff::schema_diff($source_schema, $db,
2371                                                   $dest_schema,   $db,
2372                                                   $sqltargs
2373                                                  );
2374     if(!open $file, ">$difffile") {
2375       $self->throw_exception("Can't write to $difffile ($!)");
2376       next;
2377     }
2378     print $file $diff;
2379     close($file);
2380   }
2381 }
2382
2383 =head2 deployment_statements
2384
2385 =over 4
2386
2387 =item Arguments: $schema, $type, $version, $directory, $sqlt_args
2388
2389 =back
2390
2391 Returns the statements used by L</deploy> and L<DBIx::Class::Schema/deploy>.
2392
2393 The L<SQL::Translator> (not L<DBI>) database driver name can be explicitly
2394 provided in C<$type>, otherwise the result of L</sqlt_type> is used as default.
2395
2396 C<$directory> is used to return statements from files in a previously created
2397 L</create_ddl_dir> directory and is optional. The filenames are constructed
2398 from L<DBIx::Class::Schema/ddl_filename>, the schema name and the C<$version>.
2399
2400 If no C<$directory> is specified then the statements are constructed on the
2401 fly using L<SQL::Translator> and C<$version> is ignored.
2402
2403 See L<SQL::Translator/METHODS> for a list of values for C<$sqlt_args>.
2404
2405 =cut
2406
2407 sub deployment_statements {
2408   my ($self, $schema, $type, $version, $dir, $sqltargs) = @_;
2409   $type ||= $self->sqlt_type;
2410   $version ||= $schema->schema_version || '1.x';
2411   $dir ||= './';
2412   my $filename = $schema->ddl_filename($type, $version, $dir);
2413   if(-f $filename)
2414   {
2415       my $file;
2416       open($file, "<$filename")
2417         or $self->throw_exception("Can't open $filename ($!)");
2418       my @rows = <$file>;
2419       close($file);
2420       return join('', @rows);
2421   }
2422
2423   unless (DBIx::Class::Optional::Dependencies->req_ok_for ('deploy') ) {
2424     $self->throw_exception("Can't deploy without a ddl_dir or " . DBIx::Class::Optional::Dependencies->req_missing_for ('deploy') );
2425   }
2426
2427   # sources needs to be a parser arg, but for simplicty allow at top level
2428   # coming in
2429   $sqltargs->{parser_args}{sources} = delete $sqltargs->{sources}
2430       if exists $sqltargs->{sources};
2431
2432   my $tr = SQL::Translator->new(
2433     producer => "SQL::Translator::Producer::${type}",
2434     %$sqltargs,
2435     parser => 'SQL::Translator::Parser::DBIx::Class',
2436     data => $schema,
2437   );
2438
2439   my @ret;
2440   my $wa = wantarray;
2441   if ($wa) {
2442     @ret = $tr->translate;
2443   }
2444   else {
2445     $ret[0] = $tr->translate;
2446   }
2447
2448   $self->throw_exception( 'Unable to produce deployment statements: ' . $tr->error)
2449     unless (@ret && defined $ret[0]);
2450
2451   return $wa ? @ret : $ret[0];
2452 }
2453
2454 sub deploy {
2455   my ($self, $schema, $type, $sqltargs, $dir) = @_;
2456   my $deploy = sub {
2457     my $line = shift;
2458     return if($line =~ /^--/);
2459     return if(!$line);
2460     # next if($line =~ /^DROP/m);
2461     return if($line =~ /^BEGIN TRANSACTION/m);
2462     return if($line =~ /^COMMIT/m);
2463     return if $line =~ /^\s+$/; # skip whitespace only
2464     $self->_query_start($line);
2465     eval {
2466       # do a dbh_do cycle here, as we need some error checking in
2467       # place (even though we will ignore errors)
2468       $self->dbh_do (sub { $_[1]->do($line) });
2469     };
2470     if ($@) {
2471       carp qq{$@ (running "${line}")};
2472     }
2473     $self->_query_end($line);
2474   };
2475   my @statements = $schema->deployment_statements($type, undef, $dir, { %{ $sqltargs || {} }, no_comments => 1 } );
2476   if (@statements > 1) {
2477     foreach my $statement (@statements) {
2478       $deploy->( $statement );
2479     }
2480   }
2481   elsif (@statements == 1) {
2482     foreach my $line ( split(";\n", $statements[0])) {
2483       $deploy->( $line );
2484     }
2485   }
2486 }
2487
2488 =head2 datetime_parser
2489
2490 Returns the datetime parser class
2491
2492 =cut
2493
2494 sub datetime_parser {
2495   my $self = shift;
2496   return $self->{datetime_parser} ||= do {
2497     $self->build_datetime_parser(@_);
2498   };
2499 }
2500
2501 =head2 datetime_parser_type
2502
2503 Defines (returns) the datetime parser class - currently hardwired to
2504 L<DateTime::Format::MySQL>
2505
2506 =cut
2507
2508 sub datetime_parser_type { "DateTime::Format::MySQL"; }
2509
2510 =head2 build_datetime_parser
2511
2512 See L</datetime_parser>
2513
2514 =cut
2515
2516 sub build_datetime_parser {
2517   my $self = shift;
2518   my $type = $self->datetime_parser_type(@_);
2519   $self->ensure_class_loaded ($type);
2520   return $type;
2521 }
2522
2523
2524 =head2 is_replicating
2525
2526 A boolean that reports if a particular L<DBIx::Class::Storage::DBI> is set to
2527 replicate from a master database.  Default is undef, which is the result
2528 returned by databases that don't support replication.
2529
2530 =cut
2531
2532 sub is_replicating {
2533     return;
2534
2535 }
2536
2537 =head2 lag_behind_master
2538
2539 Returns a number that represents a certain amount of lag behind a master db
2540 when a given storage is replicating.  The number is database dependent, but
2541 starts at zero and increases with the amount of lag. Default in undef
2542
2543 =cut
2544
2545 sub lag_behind_master {
2546     return;
2547 }
2548
2549 =head2 relname_to_table_alias
2550
2551 =over 4
2552
2553 =item Arguments: $relname, $join_count
2554
2555 =back
2556
2557 L<DBIx::Class> uses L<DBIx::Class::Relationship> names as table aliases in
2558 queries.
2559
2560 This hook is to allow specific L<DBIx::Class::Storage> drivers to change the
2561 way these aliases are named.
2562
2563 The default behavior is C<< "$relname_$join_count" if $join_count > 1 >>,
2564 otherwise C<"$relname">.
2565
2566 =cut
2567
2568 sub relname_to_table_alias {
2569   my ($self, $relname, $join_count) = @_;
2570
2571   my $alias = ($join_count && $join_count > 1 ?
2572     join('_', $relname, $join_count) : $relname);
2573
2574   return $alias;
2575 }
2576
2577 sub DESTROY {
2578   my $self = shift;
2579
2580   $self->_verify_pid if $self->_dbh;
2581
2582   # some databases need this to stop spewing warnings
2583   if (my $dbh = $self->_dbh) {
2584     local $@;
2585     eval {
2586       %{ $dbh->{CachedKids} } = ();
2587       $dbh->disconnect;
2588     };
2589   }
2590
2591   $self->_dbh(undef);
2592 }
2593
2594 1;
2595
2596 =head1 USAGE NOTES
2597
2598 =head2 DBIx::Class and AutoCommit
2599
2600 DBIx::Class can do some wonderful magic with handling exceptions,
2601 disconnections, and transactions when you use C<< AutoCommit => 1 >>
2602 (the default) combined with C<txn_do> for transaction support.
2603
2604 If you set C<< AutoCommit => 0 >> in your connect info, then you are always
2605 in an assumed transaction between commits, and you're telling us you'd
2606 like to manage that manually.  A lot of the magic protections offered by
2607 this module will go away.  We can't protect you from exceptions due to database
2608 disconnects because we don't know anything about how to restart your
2609 transactions.  You're on your own for handling all sorts of exceptional
2610 cases if you choose the C<< AutoCommit => 0 >> path, just as you would
2611 be with raw DBI.
2612
2613
2614 =head1 AUTHORS
2615
2616 Matt S. Trout <mst@shadowcatsystems.co.uk>
2617
2618 Andy Grundman <andy@hybridized.org>
2619
2620 =head1 LICENSE
2621
2622 You may distribute this code under the same terms as Perl itself.
2623
2624 =cut