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
  14 __PACKAGE__->mk_group_accessors(
  15   'simple' =>
  16     qw/_connect_info _dbh _sql_maker _sql_maker_opts _conn_pid _conn_tid
  17        cursor on_connect_do transaction_depth/
  18 );
  19
  20 BEGIN {
  21
  22 package DBIC::SQL::Abstract; # Would merge upstream, but nate doesn't reply :(
  23
  24 use base qw/SQL::Abstract::Limit/;
  25
  26 # This prevents the caching of $dbh in S::A::L, I believe
  27 sub new {
  28   my $self = shift->SUPER::new(@_);
  29
  30   # If limit_dialect is a ref (like a $dbh), go ahead and replace
  31   #   it with what it resolves to:
  32   $self->{limit_dialect} = $self->_find_syntax($self->{limit_dialect})
  33     if ref $self->{limit_dialect};
  34
  35   $self;
  36 }
  37
  38 sub _RowNumberOver {
  39   my ($self, $sql, $order, $rows, $offset ) = @_;
  40
  41   $offset += 1;
  42   my $last = $rows + $offset;
  43   my ( $order_by ) = $self->_order_by( $order );
  44
  45   $sql = <<"";
  46 SELECT * FROM
  47 (
  48    SELECT Q1.*, ROW_NUMBER() OVER( ) AS ROW_NUM FROM (
  49       $sql
  50       $order_by
  51    ) Q1
  52 ) Q2
  53 WHERE ROW_NUM BETWEEN $offset AND $last
  54
  55   return $sql;
  56 }
  57
  58
  59 # While we're at it, this should make LIMIT queries more efficient,
  60 #  without digging into things too deeply
  61 sub _find_syntax {
  62   my ($self, $syntax) = @_;
  63   my $dbhname = ref $syntax eq 'HASH' ? $syntax->{Driver}{Name} : '';
  64   if(ref($self) && $dbhname && $dbhname eq 'DB2') {
  65     return 'RowNumberOver';
  66   }
  67
  68   $self->{_cached_syntax} ||= $self->SUPER::_find_syntax($syntax);
  69 }
  70
  71 sub select {
  72   my ($self, $table, $fields, $where, $order, @rest) = @_;
  73   $table = $self->_quote($table) unless ref($table);
  74   local $self->{rownum_hack_count} = 1
  75     if (defined $rest[0] && $self->{limit_dialect} eq 'RowNum');
  76   @rest = (-1) unless defined $rest[0];
  77   die "LIMIT 0 Does Not Compute" if $rest[0] == 0;
  78     # and anyway, SQL::Abstract::Limit will cause a barf if we don't first
  79   local $self->{having_bind} = [];
  80   my ($sql, @ret) = $self->SUPER::select(
  81     $table, $self->_recurse_fields($fields), $where, $order, @rest
  82   );
  83   return wantarray ? ($sql, @ret, @{$self->{having_bind}}) : $sql;
  84 }
  85
  86 sub insert {
  87   my $self = shift;
  88   my $table = shift;
  89   $table = $self->_quote($table) unless ref($table);
  90   $self->SUPER::insert($table, @_);
  91 }
  92
  93 sub update {
  94   my $self = shift;
  95   my $table = shift;
  96   $table = $self->_quote($table) unless ref($table);
  97   $self->SUPER::update($table, @_);
  98 }
  99
 100 sub delete {
 101   my $self = shift;
 102   my $table = shift;
 103   $table = $self->_quote($table) unless ref($table);
 104   $self->SUPER::delete($table, @_);
 105 }
 106
 107 sub _emulate_limit {
 108   my $self = shift;
 109   if ($_[3] == -1) {
 110     return $_[1].$self->_order_by($_[2]);
 111   } else {
 112     return $self->SUPER::_emulate_limit(@_);
 113   }
 114 }
 115
 116 sub _recurse_fields {
 117   my ($self, $fields) = @_;
 118   my $ref = ref $fields;
 119   return $self->_quote($fields) unless $ref;
 120   return $$fields if $ref eq 'SCALAR';
 121
 122   if ($ref eq 'ARRAY') {
 123     return join(', ', map {
 124       $self->_recurse_fields($_)
 125       .(exists $self->{rownum_hack_count}
 126          ? ' AS col'.$self->{rownum_hack_count}++
 127          : '')
 128      } @$fields);
 129   } elsif ($ref eq 'HASH') {
 130     foreach my $func (keys %$fields) {
 131       return $self->_sqlcase($func)
 132         .'( '.$self->_recurse_fields($fields->{$func}).' )';
 133     }
 134   }
 135 }
 136
 137 sub _order_by {
 138   my $self = shift;
 139   my $ret = '';
 140   my @extra;
 141   if (ref $_[0] eq 'HASH') {
 142     if (defined $_[0]->{group_by}) {
 143       $ret = $self->_sqlcase(' group by ')
 144                .$self->_recurse_fields($_[0]->{group_by});
 145     }
 146     if (defined $_[0]->{having}) {
 147       my $frag;
 148       ($frag, @extra) = $self->_recurse_where($_[0]->{having});
 149       push(@{$self->{having_bind}}, @extra);
 150       $ret .= $self->_sqlcase(' having ').$frag;
 151     }
 152     if (defined $_[0]->{order_by}) {
 153       $ret .= $self->_order_by($_[0]->{order_by});
 154     }
 155   } elsif (ref $_[0] eq 'SCALAR') {
 156     $ret = $self->_sqlcase(' order by ').${ $_[0] };
 157   } elsif (ref $_[0] eq 'ARRAY' && @{$_[0]}) {
 158     my @order = @{+shift};
 159     $ret = $self->_sqlcase(' order by ')
 160           .join(', ', map {
 161                         my $r = $self->_order_by($_, @_);
 162                         $r =~ s/^ ?ORDER BY //i;
 163                         $r;
 164                       } @order);
 165   } else {
 166     $ret = $self->SUPER::_order_by(@_);
 167   }
 168   return $ret;
 169 }
 170
 171 sub _order_directions {
 172   my ($self, $order) = @_;
 173   $order = $order->{order_by} if ref $order eq 'HASH';
 174   return $self->SUPER::_order_directions($order);
 175 }
 176
 177 sub _table {
 178   my ($self, $from) = @_;
 179   if (ref $from eq 'ARRAY') {
 180     return $self->_recurse_from(@$from);
 181   } elsif (ref $from eq 'HASH') {
 182     return $self->_make_as($from);
 183   } else {
 184     return $from; # would love to quote here but _table ends up getting called
 185                   # twice during an ->select without a limit clause due to
 186                   # the way S::A::Limit->select works. should maybe consider
 187                   # bypassing this and doing S::A::select($self, ...) in
 188                   # our select method above. meantime, quoting shims have
 189                   # been added to select/insert/update/delete here
 190   }
 191 }
 192
 193 sub _recurse_from {
 194   my ($self, $from, @join) = @_;
 195   my @sqlf;
 196   push(@sqlf, $self->_make_as($from));
 197   foreach my $j (@join) {
 198     my ($to, $on) = @$j;
 199
 200     # check whether a join type exists
 201     my $join_clause = '';
 202     my $to_jt = ref($to) eq 'ARRAY' ? $to->[0] : $to;
 203     if (ref($to_jt) eq 'HASH' and exists($to_jt->{-join_type})) {
 204       $join_clause = ' '.uc($to_jt->{-join_type}).' JOIN ';
 205     } else {
 206       $join_clause = ' JOIN ';
 207     }
 208     push(@sqlf, $join_clause);
 209
 210     if (ref $to eq 'ARRAY') {
 211       push(@sqlf, '(', $self->_recurse_from(@$to), ')');
 212     } else {
 213       push(@sqlf, $self->_make_as($to));
 214     }
 215     push(@sqlf, ' ON ', $self->_join_condition($on));
 216   }
 217   return join('', @sqlf);
 218 }
 219
 220 sub _make_as {
 221   my ($self, $from) = @_;
 222   return join(' ', map { (ref $_ eq 'SCALAR' ? $$_ : $self->_quote($_)) }
 223                      reverse each %{$self->_skip_options($from)});
 224 }
 225
 226 sub _skip_options {
 227   my ($self, $hash) = @_;
 228   my $clean_hash = {};
 229   $clean_hash->{$_} = $hash->{$_}
 230     for grep {!/^-/} keys %$hash;
 231   return $clean_hash;
 232 }
 233
 234 sub _join_condition {
 235   my ($self, $cond) = @_;
 236   if (ref $cond eq 'HASH') {
 237     my %j;
 238     for (keys %$cond) {
 239       my $x = '= '.$self->_quote($cond->{$_}); $j{$_} = \$x;
 240     };
 241     return $self->_recurse_where(\%j);
 242   } elsif (ref $cond eq 'ARRAY') {
 243     return join(' OR ', map { $self->_join_condition($_) } @$cond);
 244   } else {
 245     die "Can't handle this yet!";
 246   }
 247 }
 248
 249 sub _quote {
 250   my ($self, $label) = @_;
 251   return '' unless defined $label;
 252   return "*" if $label eq '*';
 253   return $label unless $self->{quote_char};
 254   if(ref $self->{quote_char} eq "ARRAY"){
 255     return $self->{quote_char}->[0] . $label . $self->{quote_char}->[1]
 256       if !defined $self->{name_sep};
 257     my $sep = $self->{name_sep};
 258     return join($self->{name_sep},
 259         map { $self->{quote_char}->[0] . $_ . $self->{quote_char}->[1]  }
 260        split(/\Q$sep\E/,$label));
 261   }
 262   return $self->SUPER::_quote($label);
 263 }
 264
 265 sub limit_dialect {
 266     my $self = shift;
 267     $self->{limit_dialect} = shift if @_;
 268     return $self->{limit_dialect};
 269 }
 270
 271 sub quote_char {
 272     my $self = shift;
 273     $self->{quote_char} = shift if @_;
 274     return $self->{quote_char};
 275 }
 276
 277 sub name_sep {
 278     my $self = shift;
 279     $self->{name_sep} = shift if @_;
 280     return $self->{name_sep};
 281 }
 282
 283 } # End of BEGIN block
 284
 285 =head1 NAME
 286
 287 DBIx::Class::Storage::DBI - DBI storage handler
 288
 289 =head1 SYNOPSIS
 290
 291 =head1 DESCRIPTION
 292
 293 This class represents the connection to an RDBMS via L<DBI>.  See
 294 L<DBIx::Class::Storage> for general information.  This pod only
 295 documents DBI-specific methods and behaviors.
 296
 297 =head1 METHODS
 298
 299 =cut
 300
 301 sub new {
 302   my $new = shift->next::method(@_);
 303
 304   $new->cursor("DBIx::Class::Storage::DBI::Cursor");
 305   $new->transaction_depth(0);
 306   $new->_sql_maker_opts({});
 307
 308   $new;
 309 }
 310
 311 =head2 connect_info
 312
 313 The arguments of C<connect_info> are always a single array reference.
 314
 315 This is normally accessed via L<DBIx::Class::Schema/connection>, which
 316 encapsulates its argument list in an arrayref before calling
 317 C<connect_info> here.
 318
 319 The arrayref can either contain the same set of arguments one would
 320 normally pass to L<DBI/connect>, or a lone code reference which returns
 321 a connected database handle.
 322
 323 In either case, if the final argument in your connect_info happens
 324 to be a hashref, C<connect_info> will look there for several
 325 connection-specific options:
 326
 327 =over 4
 328
 329 =item on_connect_do
 330
 331 This can be set to an arrayref of literal sql statements, which will
 332 be executed immediately after making the connection to the database
 333 every time we [re-]connect.
 334
 335 =item limit_dialect
 336
 337 Sets the limit dialect. This is useful for JDBC-bridge among others
 338 where the remote SQL-dialect cannot be determined by the name of the
 339 driver alone.
 340
 341 =item quote_char
 342
 343 Specifies what characters to use to quote table and column names. If
 344 you use this you will want to specify L<name_sep> as well.
 345
 346 quote_char expects either a single character, in which case is it is placed
 347 on either side of the table/column, or an arrayref of length 2 in which case the
 348 table/column name is placed between the elements.
 349
 350 For example under MySQL you'd use C<quote_char =E<gt> '`'>, and user SQL Server you'd
 351 use C<quote_char =E<gt> [qw/[ ]/]>.
 352
 353 =item name_sep
 354
 355 This only needs to be used in conjunction with L<quote_char>, and is used to
 356 specify the charecter that seperates elements (schemas, tables, columns) from
 357 each other. In most cases this is simply a C<.>.
 358
 359 =back
 360
 361 These options can be mixed in with your other L<DBI> connection attributes,
 362 or placed in a seperate hashref after all other normal L<DBI> connection
 363 arguments.
 364
 365 Every time C<connect_info> is invoked, any previous settings for
 366 these options will be cleared before setting the new ones, regardless of
 367 whether any options are specified in the new C<connect_info>.
 368
 369 Important note:  DBIC expects the returned database handle provided by
 370 a subref argument to have RaiseError set on it.  If it doesn't, things
 371 might not work very well, YMMV.  If you don't use a subref, DBIC will
 372 force this setting for you anyways.  Setting HandleError to anything
 373 other than simple exception object wrapper might cause problems too.
 374
 375 Examples:
 376
 377   # Simple SQLite connection
 378   ->connect_info([ 'dbi:SQLite:./foo.db' ]);
 379
 380   # Connect via subref
 381   ->connect_info([ sub { DBI->connect(...) } ]);
 382
 383   # A bit more complicated
 384   ->connect_info(
 385     [
 386       'dbi:Pg:dbname=foo',
 387       'postgres',
 388       'my_pg_password',
 389       { AutoCommit => 0 },
 390       { quote_char => q{"}, name_sep => q{.} },
 391     ]
 392   );
 393
 394   # Equivalent to the previous example
 395   ->connect_info(
 396     [
 397       'dbi:Pg:dbname=foo',
 398       'postgres',
 399       'my_pg_password',
 400       { AutoCommit => 0, quote_char => q{"}, name_sep => q{.} },
 401     ]
 402   );
 403
 404   # Subref + DBIC-specific connection options
 405   ->connect_info(
 406     [
 407       sub { DBI->connect(...) },
 408       {
 409           quote_char => q{`},
 410           name_sep => q{@},
 411           on_connect_do => ['SET search_path TO myschema,otherschema,public'],
 412       },
 413     ]
 414   );
 415
 416 =cut
 417
 418 sub connect_info {
 419   my ($self, $info_arg) = @_;
 420
 421   return $self->_connect_info if !$info_arg;
 422
 423   # Kill sql_maker/_sql_maker_opts, so we get a fresh one with only
 424   #  the new set of options
 425   $self->_sql_maker(undef);
 426   $self->_sql_maker_opts({});
 427
 428   my $info = [ @$info_arg ]; # copy because we can alter it
 429   my $last_info = $info->[-1];
 430   if(ref $last_info eq 'HASH') {
 431     if(my $on_connect_do = delete $last_info->{on_connect_do}) {
 432       $self->on_connect_do($on_connect_do);
 433     }
 434     for my $sql_maker_opt (qw/limit_dialect quote_char name_sep/) {
 435       if(my $opt_val = delete $last_info->{$sql_maker_opt}) {
 436         $self->_sql_maker_opts->{$sql_maker_opt} = $opt_val;
 437       }
 438     }
 439
 440     # Get rid of any trailing empty hashref
 441     pop(@$info) if !keys %$last_info;
 442   }
 443
 444   $self->_connect_info($info);
 445 }
 446
 447 =head2 on_connect_do
 448
 449 This method is deprecated in favor of setting via L</connect_info>.
 450
 451 =head2 dbh_do
 452
 453 Arguments: $subref, @extra_coderef_args?
 454
 455 Execute the given subref with the underlying database handle as its
 456 first argument, using the new exception-based connection management.
 457
 458 Any additional arguments will be passed verbatim to the called subref
 459 as arguments 2 and onwards.
 460
 461 Example:
 462
 463   my @stuff = $schema->storage->dbh_do(
 464     sub {
 465       my $dbh = shift;
 466       my $cols = join(q{, }, @_);
 467       shift->selectrow_array("SELECT $cols FROM foo")
 468     },
 469     @column_list
 470   );
 471
 472 =cut
 473
 474 sub dbh_do {
 475   my $self = shift;
 476   my $coderef = shift;
 477
 478   return $coderef->($self->_dbh, @_) if $self->{_in_txn_do};
 479
 480   ref $coderef eq 'CODE' or $self->throw_exception
 481     ('$coderef must be a CODE reference');
 482
 483   my @result;
 484   my $want_array = wantarray;
 485
 486   eval {
 487     $self->_verify_pid if $self->_dbh;
 488     $self->_populate_dbh if !$self->_dbh;
 489     if($want_array) {
 490         @result = $coderef->($self->_dbh, @_);
 491     }
 492     elsif(defined $want_array) {
 493         $result[0] = $coderef->($self->_dbh, @_);
 494     }
 495     else {
 496         $coderef->($self->_dbh, @_);
 497     }
 498   };
 499
 500   my $exception = $@;
 501   if(!$exception) { return $want_array ? @result : $result[0] }
 502
 503   $self->throw_exception($exception) if $self->connected;
 504
 505   # We were not connected - reconnect and retry, but let any
 506   #  exception fall right through this time
 507   $self->_populate_dbh;
 508   $coderef->($self->_dbh, @_);
 509 }
 510
 511 # This is basically a blend of dbh_do above and DBIx::Class::Storage::txn_do.
 512 # It also informs dbh_do to bypass itself while under the direction of txn_do,
 513 #  via $self->{_in_txn_do} (this saves some redundant eval and errorcheck, etc)
 514 sub txn_do {
 515   my $self = shift;
 516   my $coderef = shift;
 517
 518   ref $coderef eq 'CODE' or $self->throw_exception
 519     ('$coderef must be a CODE reference');
 520
 521   local $self->{_in_txn_do} = 1;
 522
 523   my $tried = 0;
 524
 525   my @result;
 526   my $want_array = wantarray;
 527
 528   START_TXN: eval {
 529     $self->_verify_pid if $self->_dbh;
 530     $self->_populate_dbh if !$self->_dbh;
 531
 532     $self->txn_begin;
 533     if($want_array) {
 534         @result = $coderef->(@_);
 535     }
 536     elsif(defined $want_array) {
 537         $result[0] = $coderef->(@_);
 538     }
 539     else {
 540         $coderef->(@_);
 541     }
 542     $self->txn_commit;
 543   };
 544
 545   my $exception = $@;
 546   if(!$exception) { return $want_array ? @result : $result[0] }
 547
 548   if($tried++ > 0 || $self->connected) {
 549     eval { $self->txn_rollback };
 550     my $rollback_exception = $@;
 551     if($rollback_exception) {
 552       my $exception_class = "DBIx::Class::Storage::NESTED_ROLLBACK_EXCEPTION";
 553       $self->throw_exception($exception)  # propagate nested rollback
 554         if $rollback_exception =~ /$exception_class/;
 555
 556       $self->throw_exception(
 557         "Transaction aborted: ${exception}. "
 558         . "Rollback failed: ${rollback_exception}"
 559       );
 560     }
 561     $self->throw_exception($exception)
 562   }
 563
 564   # We were not connected, and was first try - reconnect and retry
 565   # XXX I know, gotos are evil.  If you can find a better way
 566   #  to write this that doesn't duplicate a lot of code/structure,
 567   #  and behaves identically, feel free...
 568
 569   $self->_populate_dbh;
 570   goto START_TXN;
 571 }
 572
 573 =head2 disconnect
 574
 575 Our C<disconnect> method also performs a rollback first if the
 576 database is not in C<AutoCommit> mode.
 577
 578 =cut
 579
 580 sub disconnect {
 581   my ($self) = @_;
 582
 583   if( $self->connected ) {
 584     $self->_dbh->rollback unless $self->_dbh->{AutoCommit};
 585     $self->_dbh->disconnect;
 586     $self->_dbh(undef);
 587   }
 588 }
 589
 590 sub connected {
 591   my ($self) = @_;
 592
 593   if(my $dbh = $self->_dbh) {
 594       if(defined $self->_conn_tid && $self->_conn_tid != threads->tid) {
 595           return $self->_dbh(undef);
 596       }
 597       else {
 598           $self->_verify_pid;
 599       }
 600       return ($dbh->FETCH('Active') && $dbh->ping);
 601   }
 602
 603   return 0;
 604 }
 605
 606 # handle pid changes correctly
 607 #  NOTE: assumes $self->_dbh is a valid $dbh
 608 sub _verify_pid {
 609   my ($self) = @_;
 610
 611   return if $self->_conn_pid == $$;
 612
 613   $self->_dbh->{InactiveDestroy} = 1;
 614   $self->_dbh(undef);
 615
 616   return;
 617 }
 618
 619 sub ensure_connected {
 620   my ($self) = @_;
 621
 622   unless ($self->connected) {
 623     $self->_populate_dbh;
 624   }
 625 }
 626
 627 =head2 dbh
 628
 629 Returns the dbh - a data base handle of class L<DBI>.
 630
 631 =cut
 632
 633 sub dbh {
 634   my ($self) = @_;
 635
 636   $self->ensure_connected;
 637   return $self->_dbh;
 638 }
 639
 640 sub _sql_maker_args {
 641     my ($self) = @_;
 642
 643     return ( limit_dialect => $self->dbh, %{$self->_sql_maker_opts} );
 644 }
 645
 646 sub sql_maker {
 647   my ($self) = @_;
 648   unless ($self->_sql_maker) {
 649     $self->_sql_maker(new DBIC::SQL::Abstract( $self->_sql_maker_args ));
 650   }
 651   return $self->_sql_maker;
 652 }
 653
 654 sub _populate_dbh {
 655   my ($self) = @_;
 656   my @info = @{$self->_connect_info || []};
 657   $self->_dbh($self->_connect(@info));
 658
 659   if(ref $self eq 'DBIx::Class::Storage::DBI') {
 660     my $driver = $self->_dbh->{Driver}->{Name};
 661     if ($self->load_optional_class("DBIx::Class::Storage::DBI::${driver}")) {
 662       bless $self, "DBIx::Class::Storage::DBI::${driver}";
 663       $self->_rebless() if $self->can('_rebless');
 664     }
 665   }
 666
 667   # if on-connect sql statements are given execute them
 668   foreach my $sql_statement (@{$self->on_connect_do || []}) {
 669     $self->debugobj->query_start($sql_statement) if $self->debug();
 670     $self->_dbh->do($sql_statement);
 671     $self->debugobj->query_end($sql_statement) if $self->debug();
 672   }
 673
 674   $self->_conn_pid($$);
 675   $self->_conn_tid(threads->tid) if $INC{'threads.pm'};
 676 }
 677
 678 sub _connect {
 679   my ($self, @info) = @_;
 680
 681   $self->throw_exception("You failed to provide any connection info")
 682       if !@info;
 683
 684   my ($old_connect_via, $dbh);
 685
 686   if ($INC{'Apache/DBI.pm'} && $ENV{MOD_PERL}) {
 687       $old_connect_via = $DBI::connect_via;
 688       $DBI::connect_via = 'connect';
 689   }
 690
 691   eval {
 692     if(ref $info[0] eq 'CODE') {
 693        $dbh = &{$info[0]}
 694     }
 695     else {
 696        $dbh = DBI->connect(@info);
 697        $dbh->{RaiseError} = 1;
 698        $dbh->{PrintError} = 0;
 699     }
 700   };
 701
 702   $DBI::connect_via = $old_connect_via if $old_connect_via;
 703
 704   if (!$dbh || $@) {
 705     $self->throw_exception("DBI Connection failed: " . ($@ || $DBI::errstr));
 706   }
 707
 708   $dbh;
 709 }
 710
 711 sub __txn_begin {
 712   my ($dbh, $self) = @_;
 713   if ($dbh->{AutoCommit}) {
 714     $self->debugobj->txn_begin()
 715       if ($self->debug);
 716     $dbh->begin_work;
 717   }
 718 }
 719
 720 sub txn_begin {
 721   my $self = shift;
 722   $self->dbh_do(\&__txn_begin, $self)
 723     if $self->{transaction_depth}++ == 0;
 724 }
 725
 726 sub __txn_commit {
 727   my ($dbh, $self) = @_;
 728   if ($self->{transaction_depth} == 0) {
 729     unless ($dbh->{AutoCommit}) {
 730       $self->debugobj->txn_commit()
 731         if ($self->debug);
 732       $dbh->commit;
 733     }
 734   }
 735   else {
 736     if (--$self->{transaction_depth} == 0) {
 737       $self->debugobj->txn_commit()
 738         if ($self->debug);
 739       $dbh->commit;
 740     }
 741   }
 742 }
 743
 744 sub txn_commit {
 745   my $self = shift;
 746   $self->dbh_do(\&__txn_commit, $self);
 747 }
 748
 749 sub __txn_rollback {
 750   my ($dbh, $self) = @_;
 751   if ($self->{transaction_depth} == 0) {
 752     unless ($dbh->{AutoCommit}) {
 753       $self->debugobj->txn_rollback()
 754         if ($self->debug);
 755       $dbh->rollback;
 756     }
 757   }
 758   else {
 759     if (--$self->{transaction_depth} == 0) {
 760       $self->debugobj->txn_rollback()
 761         if ($self->debug);
 762       $dbh->rollback;
 763     }
 764     else {
 765       die DBIx::Class::Storage::NESTED_ROLLBACK_EXCEPTION->new;
 766     }
 767   }
 768 }
 769
 770 sub txn_rollback {
 771   my $self = shift;
 772   eval { $self->dbh_do(\&__txn_rollback, $self) };
 773   if ($@) {
 774     my $error = $@;
 775     my $exception_class = "DBIx::Class::Storage::NESTED_ROLLBACK_EXCEPTION";
 776     $error =~ /$exception_class/ and $self->throw_exception($error);
 777     $self->{transaction_depth} = 0;          # ensure that a failed rollback
 778     $self->throw_exception($error);          # resets the transaction depth
 779   }
 780 }
 781
 782 sub _execute {
 783   my ($self, $op, $extra_bind, $ident, @args) = @_;
 784   my ($sql, @bind) = $self->sql_maker->$op($ident, @args);
 785   unshift(@bind, @$extra_bind) if $extra_bind;
 786   if ($self->debug) {
 787       my @debug_bind = map { defined $_ ? qq{'$_'} : q{'NULL'} } @bind;
 788       $self->debugobj->query_start($sql, @debug_bind);
 789   }
 790   my $sth = eval { $self->sth($sql,$op) };
 791
 792   if (!$sth || $@) {
 793     $self->throw_exception(
 794       'no sth generated via sql (' . ($@ || $self->_dbh->errstr) . "): $sql"
 795     );
 796   }
 797   @bind = map { ref $_ ? ''.$_ : $_ } @bind; # stringify args
 798   my $rv;
 799   if ($sth) {
 800     my $time = time();
 801     $rv = eval { $sth->execute(@bind) };
 802
 803     if ($@ || !$rv) {
 804       $self->throw_exception("Error executing '$sql': ".($@ || $sth->errstr));
 805     }
 806   } else {
 807     $self->throw_exception("'$sql' did not generate a statement.");
 808   }
 809   if ($self->debug) {
 810       my @debug_bind = map { defined $_ ? qq{`$_'} : q{`NULL'} } @bind;
 811       $self->debugobj->query_end($sql, @debug_bind);
 812   }
 813   return (wantarray ? ($rv, $sth, @bind) : $rv);
 814 }
 815
 816 sub insert {
 817   my ($self, $ident, $to_insert) = @_;
 818   $self->throw_exception(
 819     "Couldn't insert ".join(', ',
 820       map "$_ => $to_insert->{$_}", keys %$to_insert
 821     )." into ${ident}"
 822   ) unless ($self->_execute('insert' => [], $ident, $to_insert));
 823   return $to_insert;
 824 }
 825
 826 sub update {
 827   return shift->_execute('update' => [], @_);
 828 }
 829
 830 sub delete {
 831   return shift->_execute('delete' => [], @_);
 832 }
 833
 834 sub _select {
 835   my ($self, $ident, $select, $condition, $attrs) = @_;
 836   my $order = $attrs->{order_by};
 837   if (ref $condition eq 'SCALAR') {
 838     $order = $1 if $$condition =~ s/ORDER BY (.*)$//i;
 839   }
 840   if (exists $attrs->{group_by} || $attrs->{having}) {
 841     $order = {
 842       group_by => $attrs->{group_by},
 843       having => $attrs->{having},
 844       ($order ? (order_by => $order) : ())
 845     };
 846   }
 847   my @args = ('select', $attrs->{bind}, $ident, $select, $condition, $order);
 848   if ($attrs->{software_limit} ||
 849       $self->sql_maker->_default_limit_syntax eq "GenericSubQ") {
 850         $attrs->{software_limit} = 1;
 851   } else {
 852     $self->throw_exception("rows attribute must be positive if present")
 853       if (defined($attrs->{rows}) && !($attrs->{rows} > 0));
 854     push @args, $attrs->{rows}, $attrs->{offset};
 855   }
 856   return $self->_execute(@args);
 857 }
 858
 859 sub select {
 860   my $self = shift;
 861   my ($ident, $select, $condition, $attrs) = @_;
 862   return $self->cursor->new($self, \@_, $attrs);
 863 }
 864
 865 sub select_single {
 866   my $self = shift;
 867   my ($rv, $sth, @bind) = $self->_select(@_);
 868   my @row = $sth->fetchrow_array;
 869   # Need to call finish() to work round broken DBDs
 870   $sth->finish();
 871   return @row;
 872 }
 873
 874 =head2 sth
 875
 876 Returns a L<DBI> sth (statement handle) for the supplied SQL.
 877
 878 =cut
 879
 880 sub __sth {
 881   my ($dbh, $sql) = @_;
 882   # 3 is the if_active parameter which avoids active sth re-use
 883   $dbh->prepare_cached($sql, {}, 3);
 884 }
 885
 886 sub sth {
 887   my ($self, $sql) = @_;
 888   $self->dbh_do(\&__sth, $sql);
 889 }
 890
 891
 892 sub __columns_info_for {
 893   my ($dbh, $self, $table) = @_;
 894
 895   if ($dbh->can('column_info')) {
 896     my %result;
 897     eval {
 898       my ($schema,$tab) = $table =~ /^(.+?)\.(.+)$/ ? ($1,$2) : (undef,$table);
 899       my $sth = $dbh->column_info( undef,$schema, $tab, '%' );
 900       $sth->execute();
 901       while ( my $info = $sth->fetchrow_hashref() ){
 902         my %column_info;
 903         $column_info{data_type}   = $info->{TYPE_NAME};
 904         $column_info{size}      = $info->{COLUMN_SIZE};
 905         $column_info{is_nullable}   = $info->{NULLABLE} ? 1 : 0;
 906         $column_info{default_value} = $info->{COLUMN_DEF};
 907         my $col_name = $info->{COLUMN_NAME};
 908         $col_name =~ s/^\"(.*)\"$/$1/;
 909
 910         $result{$col_name} = \%column_info;
 911       }
 912     };
 913     return \%result if !$@ && scalar keys %result;
 914   }
 915
 916   my %result;
 917   my $sth = $dbh->prepare("SELECT * FROM $table WHERE 1=0");
 918   $sth->execute;
 919   my @columns = @{$sth->{NAME_lc}};
 920   for my $i ( 0 .. $#columns ){
 921     my %column_info;
 922     my $type_num = $sth->{TYPE}->[$i];
 923     my $type_name;
 924     if(defined $type_num && $dbh->can('type_info')) {
 925       my $type_info = $dbh->type_info($type_num);
 926       $type_name = $type_info->{TYPE_NAME} if $type_info;
 927     }
 928     $column_info{data_type} = $type_name ? $type_name : $type_num;
 929     $column_info{size} = $sth->{PRECISION}->[$i];
 930     $column_info{is_nullable} = $sth->{NULLABLE}->[$i] ? 1 : 0;
 931
 932     if ($column_info{data_type} =~ m/^(.*?)\((.*?)\)$/) {
 933       $column_info{data_type} = $1;
 934       $column_info{size}    = $2;
 935     }
 936
 937     $result{$columns[$i]} = \%column_info;
 938   }
 939
 940   return \%result;
 941 }
 942
 943 sub columns_info_for {
 944   my ($self, $table) = @_;
 945   $self->dbh_do(\&__columns_info_for, $self, $table);
 946 }
 947
 948 =head2 last_insert_id
 949
 950 Return the row id of the last insert.
 951
 952 =cut
 953
 954 sub last_insert_id {
 955   my ($self, $row) = @_;
 956
 957   $self->dbh_do(sub { shift->func('last_insert_rowid') });
 958 }
 959
 960 =head2 sqlt_type
 961
 962 Returns the database driver name.
 963
 964 =cut
 965
 966 sub sqlt_type { shift->dbh_do(sub { shift->{Driver}->{Name} }) }
 967
 968 =head2 create_ddl_dir (EXPERIMENTAL)
 969
 970 =over 4
 971
 972 =item Arguments: $schema \@databases, $version, $directory, $sqlt_args
 973
 974 =back
 975
 976 Creates an SQL file based on the Schema, for each of the specified
 977 database types, in the given directory.
 978
 979 Note that this feature is currently EXPERIMENTAL and may not work correctly
 980 across all databases, or fully handle complex relationships.
 981
 982 =cut
 983
 984 sub create_ddl_dir
 985 {
 986   my ($self, $schema, $databases, $version, $dir, $sqltargs) = @_;
 987
 988   if(!$dir || !-d $dir)
 989   {
 990     warn "No directory given, using ./\n";
 991     $dir = "./";
 992   }
 993   $databases ||= ['MySQL', 'SQLite', 'PostgreSQL'];
 994   $databases = [ $databases ] if(ref($databases) ne 'ARRAY');
 995   $version ||= $schema->VERSION || '1.x';
 996   $sqltargs = { ( add_drop_table => 1 ), %{$sqltargs || {}} };
 997
 998   eval "use SQL::Translator";
 999   $self->throw_exception("Can't deploy without SQL::Translator: $@") if $@;
1000
1001   my $sqlt = SQL::Translator->new($sqltargs);
1002   foreach my $db (@$databases)
1003   {
1004     $sqlt->reset();
1005     $sqlt->parser('SQL::Translator::Parser::DBIx::Class');
1006 #    $sqlt->parser_args({'DBIx::Class' => $schema);
1007     $sqlt->data($schema);
1008     $sqlt->producer($db);
1009
1010     my $file;
1011     my $filename = $schema->ddl_filename($db, $dir, $version);
1012     if(-e $filename)
1013     {
1014       $self->throw_exception("$filename already exists, skipping $db");
1015       next;
1016     }
1017     open($file, ">$filename")
1018       or $self->throw_exception("Can't open $filename for writing ($!)");
1019     my $output = $sqlt->translate;
1020 #use Data::Dumper;
1021 #    print join(":", keys %{$schema->source_registrations});
1022 #    print Dumper($sqlt->schema);
1023     if(!$output)
1024     {
1025       $self->throw_exception("Failed to translate to $db. (" . $sqlt->error . ")");
1026       next;
1027     }
1028     print $file $output;
1029     close($file);
1030   }
1031
1032 }
1033
1034 =head2 deployment_statements
1035
1036 Create the statements for L</deploy> and
1037 L<DBIx::Class::Schema/deploy>.
1038
1039 =cut
1040
1041 sub deployment_statements {
1042   my ($self, $schema, $type, $version, $dir, $sqltargs) = @_;
1043   # Need to be connected to get the correct sqlt_type
1044   $self->ensure_connected() unless $type;
1045   $type ||= $self->sqlt_type;
1046   $version ||= $schema->VERSION || '1.x';
1047   $dir ||= './';
1048   eval "use SQL::Translator";
1049   if(!$@)
1050   {
1051     eval "use SQL::Translator::Parser::DBIx::Class;";
1052     $self->throw_exception($@) if $@;
1053     eval "use SQL::Translator::Producer::${type};";
1054     $self->throw_exception($@) if $@;
1055     my $tr = SQL::Translator->new(%$sqltargs);
1056     SQL::Translator::Parser::DBIx::Class::parse( $tr, $schema );
1057     return "SQL::Translator::Producer::${type}"->can('produce')->($tr);
1058   }
1059
1060   my $filename = $schema->ddl_filename($type, $dir, $version);
1061   if(!-f $filename)
1062   {
1063 #      $schema->create_ddl_dir([ $type ], $version, $dir, $sqltargs);
1064       $self->throw_exception("No SQL::Translator, and no Schema file found, aborting deploy");
1065       return;
1066   }
1067   my $file;
1068   open($file, "<$filename")
1069       or $self->throw_exception("Can't open $filename ($!)");
1070   my @rows = <$file>;
1071   close($file);
1072
1073   return join('', @rows);
1074
1075 }
1076
1077 sub deploy {
1078   my ($self, $schema, $type, $sqltargs, $dir) = @_;
1079   foreach my $statement ( $self->deployment_statements($schema, $type, undef, $dir, { no_comments => 1, %{ $sqltargs || {} } } ) ) {
1080     for ( split(";\n", $statement)) {
1081       next if($_ =~ /^--/);
1082       next if(!$_);
1083 #      next if($_ =~ /^DROP/m);
1084       next if($_ =~ /^BEGIN TRANSACTION/m);
1085       next if($_ =~ /^COMMIT/m);
1086       next if $_ =~ /^\s+$/; # skip whitespace only
1087       $self->debugobj->query_start($_) if $self->debug;
1088       $self->dbh->do($_) or warn "SQL was:\n $_"; # XXX exceptions?
1089       $self->debugobj->query_end($_) if $self->debug;
1090     }
1091   }
1092 }
1093
1094 =head2 datetime_parser
1095
1096 Returns the datetime parser class
1097
1098 =cut
1099
1100 sub datetime_parser {
1101   my $self = shift;
1102   return $self->{datetime_parser} ||= $self->build_datetime_parser(@_);
1103 }
1104
1105 =head2 datetime_parser_type
1106
1107 Defines (returns) the datetime parser class - currently hardwired to
1108 L<DateTime::Format::MySQL>
1109
1110 =cut
1111
1112 sub datetime_parser_type { "DateTime::Format::MySQL"; }
1113
1114 =head2 build_datetime_parser
1115
1116 See L</datetime_parser>
1117
1118 =cut
1119
1120 sub build_datetime_parser {
1121   my $self = shift;
1122   my $type = $self->datetime_parser_type(@_);
1123   eval "use ${type}";
1124   $self->throw_exception("Couldn't load ${type}: $@") if $@;
1125   return $type;
1126 }
1127
1128 sub DESTROY {
1129   my $self = shift;
1130   return if !$self->_dbh;
1131   $self->_verify_pid;
1132   $self->_dbh(undef);
1133 }
1134
1135 1;
1136
1137 =head1 SQL METHODS
1138
1139 The module defines a set of methods within the DBIC::SQL::Abstract
1140 namespace.  These build on L<SQL::Abstract::Limit> to provide the
1141 SQL query functions.
1142
1143 The following methods are extended:-
1144
1145 =over 4
1146
1147 =item delete
1148
1149 =item insert
1150
1151 =item select
1152
1153 =item update
1154
1155 =item limit_dialect
1156
1157 See L</connect_info> for details.
1158 For setting, this method is deprecated in favor of L</connect_info>.
1159
1160 =item quote_char
1161
1162 See L</connect_info> for details.
1163 For setting, this method is deprecated in favor of L</connect_info>.
1164
1165 =item name_sep
1166
1167 See L</connect_info> for details.
1168 For setting, this method is deprecated in favor of L</connect_info>.
1169
1170 =back
1171
1172 =head1 AUTHORS
1173
1174 Matt S. Trout <mst@shadowcatsystems.co.uk>
1175
1176 Andy Grundman <andy@hybridized.org>
1177
1178 =head1 LICENSE
1179
1180 You may distribute this code under the same terms as Perl itself.
1181
1182 =cut
1183