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