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