Amazon-DynamoDB

 view release on metacpan or  search on metacpan

lib/Amazon/DynamoDB/20120810.pm  view on Meta::CPAN

package Amazon::DynamoDB::20120810;
$Amazon::DynamoDB::20120810::VERSION = '0.35';
use strict;
use warnings;


use Future;
use Future::Utils qw(repeat try_repeat);
use POSIX qw(strftime);
use JSON::MaybeXS qw(decode_json encode_json);
use MIME::Base64;
use List::Util;
use List::MoreUtils;
use B qw(svref_2object);
use HTTP::Request;
use Kavorka;
use Amazon::DynamoDB::Types;
use Type::Registry;
use VM::EC2::Security::CredentialCache;
use AWS::Signature4;
   
BEGIN {
    my $reg = "Type::Registry"->for_me; 
    $reg->add_types(-Standard);
    $reg->add_types("Amazon::DynamoDB::Types");
};



sub new {
    my $class = shift;
    bless { @_ }, $class
}

sub implementation { shift->{implementation} }
sub host { shift->{host} }
sub port { shift->{port} }
sub ssl { shift->{ssl} }
sub algorithm { 'AWS4-HMAC-SHA256' }
sub scope { shift->{scope} }
sub access_key { shift->{access_key} }
sub secret_key { shift->{secret_key} }
sub debug_failures { shift->{debug} }

sub max_retries { shift->{max_retries} }





method create_table(TableNameType :$TableName!,
                    Int :$ReadCapacityUnits = 2, 
                    Int :$WriteCapacityUnits = 2,
                    AttributeDefinitionsType :$AttributeDefinitions,
                    KeySchemaType :$KeySchema!,
                    ArrayRef[GlobalSecondaryIndexType] :$GlobalSecondaryIndexes where { scalar(@$_) <= 5 },
                    ArrayRef[LocalSecondaryIndexType] :$LocalSecondaryIndexes
                ) {
    my %payload = (
        TableName => $TableName,
        ProvisionedThroughput => {
            ReadCapacityUnits => int($ReadCapacityUnits),
            WriteCapacityUnits => int($WriteCapacityUnits),
        }
    );

    if (defined($AttributeDefinitions)) {
        foreach my $field_name (keys %$AttributeDefinitions) {
            my $type = $AttributeDefinitions->{$field_name};

            push @{$payload{AttributeDefinitions}}, {
                AttributeName => $field_name,
                AttributeType => $type // 'S',
            }
        }
    }

    $payload{KeySchema} = _create_key_schema($KeySchema, $AttributeDefinitions);

    foreach my $index_record (['GlobalSecondaryIndexes', $GlobalSecondaryIndexes], 
                              ['LocalSecondaryIndexes', $LocalSecondaryIndexes]) {
        my $index_type = $index_record->[0];
        my $index = $index_record->[1];
        
        if (defined($index)) {
            foreach my $i (@$index) {
                my $r = {
                    IndexName => $i->{IndexName},
                    (($index_type eq 'GlobalSecondaryIndexes') ? 
                         (ProvisionedThroughput => {
                             ReadCapacityUnits => int($i->{ProvisionedThroughput}->{ReadCapacityUnits} // 1),
                             WriteCapacityUnits => int($i->{ProvisionedThroughput}->{WriteCapacityUnits} // 1),
                         }) : ()),
                    KeySchema => _create_key_schema($i->{KeySchema}, $AttributeDefinitions),
                };

                my $type = $i->{Projection}->{ProjectionType};
                $r->{Projection}->{ProjectionType} = $type;

lib/Amazon/DynamoDB/20120810.pm  view on Meta::CPAN

                                'ReturnConsumedCapacity' => $ReturnConsumedCapacity,
                                'ScanIndexForward' => $ScanIndexForward,
                                'Select' => $Select,
                                'TableName' => $TableName
                            });
    

    foreach my $key_name (keys %$KeyConditions) {
        my $key_details = $KeyConditions->{$key_name};
        $payload->{KeyConditions}->{$key_name} = {
            AttributeValueList => _encode_attribute_value_list($key_details->{AttributeValueList}, $key_details->{ComparisonOperator}),
            ComparisonOperator => $key_details->{ComparisonOperator}
        };
    }

    $self->_scan_or_query_process('Query', $payload, $code, { ResultLimit => $Limit});
}





method scan (CodeRef $code,
             AttributesToGetType :$AttributesToGet,
             KeyType :$ExclusiveStartKey,
             Int :$Limit where { $_ >= 0},
             ReturnConsumedCapacityType :$ReturnConsumedCapacity,
             ScanFilterType :$ScanFilter,
             Int :$Segment where { $_ >= 0 },
             SelectType :$Select,
             TableNameType :$TableName!,
             Int :$TotalSegments where { $_ >= 1 && $_ <= 1000000 },
             Str :$FilterExpression,
             ExpressionAttributeValuesType :$ExpressionAttributeValues,
             ExpressionAttributeNamesType :$ExpressionAttributeNames,
         ) {
    my $payload = _make_payload({
                                'AttributesToGet' => $AttributesToGet,
                                'ExclusiveStartKey' => $ExclusiveStartKey,
                                'ExpressionAttributeValues' => $ExpressionAttributeValues,
                                'ExpressionAttributeNames' => $ExpressionAttributeNames,
                                'FilterExpression' => $FilterExpression,
                                'ReturnConsumedCapacity' => $ReturnConsumedCapacity,
                                'ScanFilter' => $ScanFilter,
                                'Segment' => $Segment,
                                'Select' => $Select,
                                'TableName' => $TableName,
                                'TotalSegments' => $TotalSegments
                            });

    $self->_scan_or_query_process('Scan', $payload, $code, { ResultLimit => $Limit});
}


method make_request(Str :$target,
                    HashRef :$payload,
                ) {
    my $api_version = '20120810';
    my $host = $self->host;
    my $req = HTTP::Request->new(
        POST => (($self->ssl) ? 'https' : 'http') . '://' . $self->host . ($self->port ? (':' . $self->port) : '') . '/'
    );
    $req->header( host => $host );
    # Amazon requires ISO-8601 basic format
    my $now = time;
    my $http_date = strftime('%Y%m%dT%H%M%SZ', gmtime($now));
    my $date = strftime('%Y%m%d', gmtime($now));

    $req->protocol('HTTP/1.1');
    $req->header( 'Date' => $http_date );
    $req->header( 'x-amz-target', 'DynamoDB_'. $api_version. '.'. $target );
    $req->header( 'content-type' => 'application/x-amz-json-1.0' );
    $payload = encode_json($payload);
    $req->content($payload);
    $req->header( 'Content-Length' => length($payload));
    
    if ($self->{use_iam_role}) {
        my $creds = VM::EC2::Security::CredentialCache->get();
        defined($creds) || die("Unable to retrieve IAM role credentials");
        $self->{access_key} = $creds->accessKeyId;
        $self->{secret_key} = $creds->secretAccessKey;
        $req->header('x-amz-security-token' => $creds->sessionToken);
    }        

    my $signer = AWS::Signature4->new(-access_key => $self->access_key,
                                      -secret_key => $self->secret_key);
    
    $signer->sign($req);
    return $req;
}

method _request(HTTP::Request $req) {
    $self->implementation->request($req);
}


# Since scan and query have the same type of responses share the processing.
method _scan_or_query_process (Str $target,
                               HashRef $payload,
                               CodeRef $code,
                               HashRef $args) {
    my $finished = 0;
    my $records_seen = 0;
    my $repeat = try_repeat {
        
        # Since we're may be making more than one request in this repeat loop
        # decrease our limit of results to scan in each call by the number 
        # of records remaining that the overall request wanted ot pull.
        if (defined($args->{ResultLimit})) {
            $payload->{Limit} = $args->{ResultLimit} - $records_seen;
        }

        my $req = $self->make_request(
            target => $target,
            payload => $payload,
        );
        
        $self->_process_request(
            $req,
            sub {
                my $result = shift;

lib/Amazon/DynamoDB/20120810.pm  view on Meta::CPAN

    if (defined($r->{ItemCollectionMetrics})) {
        foreach my $key (keys %{$r->{ItemCollectionMetrics}}) {
            foreach my $key_part (keys %{$r->{ItemCollectionMetrics}->{$key}}) {
                $r->{ItemCollectionMetrics}->{$key}->{$key_part} = _decode_item_attributes($r->{ItemCollectionMetrics}->{$key})
            }
        }
    }    
    return $r;
}


fun _create_key_schema(ArrayRef $source, HashRef $known_fields) {
    defined($source) || die("No source passed to create_key_schema");
    defined($known_fields) || die("No known fields passed to create_key_schmea");
    my @r;
    foreach my $field_name (@$source) {
        defined($known_fields->{$field_name}) || Carp::confess("Unknown field specified '$field_name' in schema, must be defined in fields.  schema:" . Data::Dumper->Dump([$source]));
        push @r, {
            AttributeName => $field_name,
            KeyType       => (scalar(@r) ? 'RANGE' : 'HASH')
        };
    }
    return \@r;
};



1;

__END__

=pod

=encoding UTF-8

=head1 NAME

Amazon::DynamoDB::20120810

=head1 VERSION

version 0.35

=head1 DESCRIPTION

=head2 new

Instantiates the API object.

Expects the following named parameters:

=over 4

=item * implementation - the object which provides a Future-returning C<request> method,
see L<Amazon::DynamoDB::NaHTTP> for example.

=item * host - the host (IP or hostname) to communicate with

=item * port - the port to use for HTTP(S) requests

=item * ssl - true for HTTPS, false for HTTP

=item * algorithm - which signing algorithm to use, default AWS4-HMAC-SHA256

=item * scope - the scope for requests, typically C<region/host/aws4_request>

=item * access_key - the access key for signing requests

=item * secret_key - the secret key for signing requests

=item * debug_failures - print errors if they occur

=item * max_retries - maximum number of retries for a request

=back

=head2 create_table

Creates a new table. It may take some time before the table is marked
as active - use L</wait_for_table_status> to poll until the status changes.

Amazon Documentation:

L<http://docs.aws.amazon.com/amazondynamodb/latest/APIReference/API_CreateTable.html>

  $ddb->create_table(
     TableName => $table_name,
     ReadCapacityUnits => 2,
     WriteCapacityUnits => 2,
     AttributeDefinitions => {
         user_id => 'N',
         date => 'N',
     },
     KeySchema => ['user_id', 'date'],
     LocalSecondaryIndexes => [
         {
             IndexName => 'UserDateIndex',
             KeySchema => ['user_id', 'date'],
             Projection => {
                 ProjectionType => 'KEYS_ONLY',
             },
             ProvisionedThroughput => {
                 ReadCapacityUnits => 2,
                 WriteCapacityUnits => 2,
             }
         }
     ]
  );

=back

=head2 describe_table

Describes the given table.

Amazon Documentation:

L<http://docs.aws.amazon.com/amazondynamodb/latest/APIReference/API_DescribeTable.html>

  $ddb->describe_table(TableName => $table_name);