lib/DBIx/Class/Storage/DBI.pm

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