Mojo::Pg::Cursor - Minimal Prototype

Cursor.pm

package Mojo::Pg::Cursor;

require Mojo::Pg::CursorIterator;
require Mojo::Pg::Results;
require Scalar::Util;

use Mojo::Util 'monkey_patch';

use Mojo::Base -base;

has [qw(bind db name query)];

sub DESTROY {
    my $self = shift;
    if ($self->{close} && $self->{dbh}) { $self->close }
}

sub new {
    my $self = shift->SUPER::new(@_);
    my $sql = sprintf('DECLARE %s CURSOR WITH HOLD FOR %s', $self->name, $self->query);
    $self->{dbh} = $self->db->dbh;
    $self->db->query($sql, @{$self->bind || [] });
    $self->{close} = 1;
    return $self;
}

sub close {
    my $self = shift;
    return unless delete $self->{close};
    my $sql = sprintf('CLOSE %s', $self->name);
    $self->{dbh}->do($sql);
    return 1;
}

sub fetch {
    my ($self, $count) = (shift, shift);
    my $sql = sprintf('FETCH %d FROM %s', $count, $self->name);
    my $sth = $self->{dbh}->prepare($sql);
    $sth->execute;
    return Mojo::Pg::Results->new(sth => $sth);
}

sub iterator {
    my ($self, $batch_size) = (shift, (shift || 1));
    return Mojo::Pg::CursorIterator->new(batch_size => $batch_size, cursor => $self);
}

monkey_patch 'Mojo::Pg::Database', 'cursor', sub {
    my $self = shift;
    my $cursor = Mojo::Pg::Cursor->new(db => $self, @_);
    Scalar::Util::weaken $cursor->{db};
    return $cursor;
};

1;

=encoding utf8

=head1 NAME

Mojo::Pg::Cursor

=head1 SYNOPSIS

  # Using a cursor in iterator form:
  my $pg = Mojo::Pg->new(...);
  my $cursor = $pg->db->cursor(
      name  => 'all_big_table',
      query => 'SELECT * FROM big_table',
  )->iterator;
  while (my $row = $cursor->next) {
      say $row->{id};
  }

  # Using a cursor manually (fetch):
  my $pg = Mojo::Pg->new(...);
  my $cursor = $pg->db->cursor(
      name  => 'all_big_table',
      query => 'SELECT * FROM big_table',
  );
  my $count;
  do {
      $count = 0;
      my $results = $cursor->fetch(100);
      while (my $row = $results->hash) {
          $count++;
          say $row->{id};
      }
  } while ($count > 0);

=head1 DESCRIPTION

L<Mojo::Pg::Cursor> is a scope guard for L<DBD::Pg> cursors.

=head1 ATTRIBUTES

=head2 bind

An arrayref of the bind values to use with C<query>.

=head2 db

L<Mojo::Pg::Database> that will be used to issue SQL statements.

=head2 name

Name to use for the cursor.

Note: I don't think there is any protection about using the same name twice.
Hopefully cursors are contained within the session.

=head2 query

The SQL query the cursor should perform.

=head1 METHODS

=head2 close

  $cursor->close;

Close the cursor.  This is run when the L<Mojo::Pg::Cursor> object leaves scope.

=head2 fetch

  my $results = $cursor->fetch($batch_size);

Fetch a batch of rows.

=head2 iterator

  my $iter = $cursor->iterator;
  my $iter = $cursor->iterator($batch_size);

Create an iterator object that handles reloading batches automatically.

=head2 new

  my $cursor = $pg->cursor(
    name  => 'my_cursor',
    query => 'select * from big_table',
  );
  my $cursor = $pg->cursor(
    name  => 'my_cursor',
    query => 'select * from big_table where id < (?)',
    bind  => [10],
  );

Create a L<Mojo::Pg::Cursor>.

=head1 SEE ALSO

L<Mojo::Pg>

=cut

CursorIterator.pm

package Mojo::Pg::CursorIterator;

use Mojo::Base -base;

has [qw(batch_size cursor expand)];

sub next {
    my $self = shift;
    if (--$self->{count} <= 0) {
        $self->{count} = $self->batch_size;
        $self->{results} = $self->cursor->fetch($self->{batch_size});
        $self->{results}->expand if $self->expand;
    }
    return $self->{results}->hash;
}

1;

=encoding utf8

=head1 NAME

Mojo::Pg::CursorIterator

=head1 SYNOPSIS

  my $iter = $cursor->iterator;
  while (my $row = $iter->next) {
      say $row->{id};
  }

=head1 DESCRIPTION

L<Mojo::Pg::CursorIterator> is an iterator that automatically handles reloading
row batches from a cursor.

=head1 ATTRIBUTES

=head2 batch_size

The number of rows to fetch for each batch.  Defaults to 1.

=head2 cursor

L<Mojo::Pg::Cursor> to manage.

=head2 expand

Option to expand JSON on the results.  This iterator uses L<Mojo::Pg::Results>
internally; this option causes C<expand()> to be called on the results.

=head1 METHODS

=head2 next

Returns the next row as a hash.

=head1 SEE ALSO

L<Mojo::Pg>

=cut