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