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