File Coverage

blib/lib/InfluxDB.pm

Criterion	Covered	Total	%
statement	41	191	21.4
branch	6	66	9.0
condition	2	3	66.6
subroutine	13	40	32.5
pod	25	29	86.2
total	87	329	26.4

package InfluxDB;

use strict;
use warnings;
use 5.010_000;

our $VERSION = '1.005';

use Class::Accessor::Lite (
    new => 0,
    ro  => [qw(host port username database ssl json)],
);

use Carp;
use Data::Validator;
use Mouse::Util::TypeConstraints;
use Furl;
use URI::Escape;
use JSON 2;

enum 'TimePrecision' => qw(s m u);

subtype 'JSONBool' => as 'ScalarRef';
coerce 'JSONBool'
    => from 'Bool' => via { $_ ? \1 : \0 }
    => from 'Object' => via { JSON::is_bool($_) ? ($_ == 1 ? \1 : \0) : \0 }
;

sub new {
    state $rule = Data::Validator->new(
        host     => { isa => 'Str' },
        port     => { isa => 'Int', default => 8086 },
        username => { isa => 'Str', optional => 1 },
        password => { isa => 'Str', optional => 1 },
        database => { isa => 'Str' },
        ssl      => { isa => 'Bool', default => 0 },

        timeout  => { isa => 'Int', default => 120 },
        debug    => { isa => 'Bool', optional => 1 },
    )->with('NoRestricted')->with('Method');
    # Mouse::Util::apply_all_role leaks memory when takes 2 or more extensions.
    # so apply for each extension.
    my($class, $args) = $rule->validate(@_);

    if (delete $args->{debug}) {
        $ENV{IX_DEBUG} = 1;
    }

    my $self = bless {
        ua     => Furl->new(
            agent   => join('/', __PACKAGE__, $VERSION),
            timeout => $args->{timeout},
        ),
        json   => JSON->new,
        status => {},
        %$args
    }, $class;

    return $self;
}

sub debugf {
    return unless $ENV{IX_DEBUG};
    print STDERR "[DEBUG] ",sprintf(shift @_, @_),"\n";
}

sub status {
    my($self, $res) = @_;

    if ($res) {
        $self->{status} = {
            code        => $res->code,
            message     => $res->message,
            status_line => $res->status_line,
            content     => $res->content,
        };
        debugf("content: %s", $res->content);
    }

    return $self->{status};
}

sub errstr {
    my($self) = @_;
    my $errstr = "";

    if (substr($self->{status}{code}, 0, 2) ne "20") {
        $errstr = join("\n",
                       $self->{status}{status_line},
                       $self->{status}{content},
                   );
    }

    return $errstr;
}

sub as_hash {
    my(undef, $result) = @_;
    my $h;

    for my $r (@{ $result }) {
        my $series  = $r->{name};
        my @columns = @{ $r->{columns} };

        for my $p (@{ $r->{points} }) {
            my %ph;
            @ph{ @columns } = @$p;
            push @{ $h->{$series} }, \%ph;
        }
    }

    return $h;
}

sub switch_database {
    state $rule = Data::Validator->new(
        database => { isa => 'Str' },
    )->with('Method');
    my($self, $args) = $rule->validate(@_);

    $self->{database} = $args->{database};

    return 1;
}

sub switch_user {
    state $rule = Data::Validator->new(
        username => { isa => 'Str' },
        password => { isa => 'Str' },
    )->with('Method');
    my($self, $args) = $rule->validate(@_);

    $self->{username} = $args->{username};
    $self->{password} = $args->{password};

    return 1;
}

### database #############################################################
sub create_database {
    state $rule = Data::Validator->new(
        database => { isa => 'Str' },
    )->with('Method');
    my($self, $args) = $rule->validate(@_);

    my $url = $self->_build_url(
        path => '/db',
    );

    my $res = $self->{ua}->post($url, [], $self->json->encode({
        name => $args->{database},
    }));
    $self->status($res);

    return $res->is_success ? 1 : ();
}

sub list_database {
    my $self = shift;

    my $url = $self->_build_url(
        path => '/db',
    );

    my $res = $self->{ua}->get($url);
    $self->status($res);

    return $res->is_success ? $self->json->decode($res->content) : ();
}

sub delete_database {
    state $rule = Data::Validator->new(
        database => { isa => 'Str' },
    )->with('Method');
    my($self, $args) = $rule->validate(@_);

    my $url = $self->_build_url(
        path => '/db/'. $args->{database},
    );

    my $res = $self->{ua}->delete($url);
    $self->status($res);

    return $res->is_success ? 1 : ();
}

### series ###############################################################
## hmmm v0.5.1 (latest version) returns empty response
## https://github.com/FGRibreau/influxdb-cli/issues/8
sub list_series {
    my $self = shift;
    return $self->query(q => "list series");
}

### points ###############################################################
sub write_points {
    state $rule = Data::Validator->new(
        data           => { isa => 'ArrayRef|HashRef' },
        time_precision => { isa => 'TimePrecision', optional => 1 },
    )->with('Method');
    my($self, $args) = $rule->validate(@_);

    my $data = ref($args->{data}) eq 'ARRAY' ? $args->{data} : [$args->{data}];
    $data = $self->json->encode($data);
    debugf("data: %s", $data);

    my $url = $self->_build_url(
       path =>  '/db/' . $self->database . '/series',
       qs   => {
           time_precision => exists $args->{time_precision} ? $args->{time_precision} :  undef,
       },
    );

    my $res = $self->{ua}->post($url, [], $data);
    $self->status($res);

    return $res->is_success ? 1 : ();
}

sub delete_points {
    state $rule = Data::Validator->new(
        name  => { isa => 'Str' },
    )->with('Method');
    my($self, $args) = $rule->validate(@_);

    my $url = $self->_build_url(
       path =>  '/db/' . $self->database . '/series/' . $args->{name},
       qs   => {
           %$args,
       },
    );

    my $res = $self->{ua}->delete($url);
    $self->status($res);

    return $res->is_success ? 1 : ();
}

sub create_scheduled_deletes {
    croak "Not implemented in InfluxDB v0.5.1";
    # state $rule = Data::Validator->new(
    # )->with('Method');
    # my($self, $args) = $rule->validate(@_);
}

sub list_scheduled_deletes {
    croak "Not implemented in InfluxDB v0.5.1";
    # state $rule = Data::Validator->new(
    # )->with('Method');
    # my($self, $args) = $rule->validate(@_);
}

sub delete_scheduled_deletes {
    croak "Not implemented in InfluxDB v0.5.1";
    # state $rule = Data::Validator->new(
    # )->with('Method');
    # my($self, $args) = $rule->validate(@_);
}

sub query {
    state $rule = Data::Validator->new(
        q              => { isa => 'Str' },
        time_precision => { isa => 'TimePrecision', optional => 1 },
        chunked        => { isa => 'Bool', default => 0 },
        # order          => { isa => 'Str', optional => 1 }, # not implemented?
    )->with('Method');
    my($self, $args) = $rule->validate(@_);

    my $url = $self->_build_url(
        path => '/db/' . $self->database . '/series',
        qs   => {
            q              => $args->{q},
            time_precision => (exists $args->{time_precision} ? $args->{time_precision} :  undef),
            chunked        => ($args->{chunked} ? 'true' : 'false'),
            order          => (exists $args->{order} ? $args->{order} :  undef),
        },
    );

    my $res = $self->{ua}->get($url);
    $self->status($res);

    if ($res->is_success) {
        my $result;
        if ($args->{chunked}) {
            $result = [$self->json->incr_parse($res->content)];
        } else {
            $result = $self->json->decode($res->content);
        }
        return $result;
    } else {
        return;
    }
}

### Continuous Queries ###################################################
sub create_continuous_query {
    state $rule = Data::Validator->new(
        q    => { isa => 'Str' },
        name => { isa => 'Str' },
    )->with('Method');
    my($self, $args) = $rule->validate(@_);

    return $self->query(q => "$args->{q} into $args->{name}");
}

sub list_continuous_queries {
    my $self = shift;
    return $self->query(q => "list continuous queries");
}

sub drop_continuous_query {
    state $rule = Data::Validator->new(
        id => { isa => 'Str' },
    )->with('Method');
    my($self, $args) = $rule->validate(@_);

    return $self->query(q => "drop continuous query $args->{id}");
}

### user #################################################################
sub create_database_user {
    state $rule = Data::Validator->new(
        name     => { isa => 'Str' },
        password => { isa => 'Str' },
    )->with('Method');
    my($self, $args) = $rule->validate(@_);

    my $url = $self->_build_url(
        path => '/db/' . $self->database . '/users',
    );

    my $res = $self->{ua}->post($url, [], $self->json->encode({
        name     => $args->{name},
        password => $args->{password},
    }));
    $self->status($res);

    return $res->is_success ? 1 : ();
}

sub delete_database_user {
    state $rule = Data::Validator->new(
        name => { isa => 'Str' },
    )->with('Method');
    my($self, $args) = $rule->validate(@_);

    my $url = $self->_build_url(
        path => '/db/' . $self->database . '/users/' . $args->{name},
    );

    my $res = $self->{ua}->delete($url);
    $self->status($res);

    return $res->is_success ? 1 : ();
}

sub update_database_user {
    state $rule = Data::Validator->new(
        name     => { isa => 'Str' },
        password => { isa => 'Str', optional => 1 },
        admin    => { isa => 'JSONBool', optional => 1 },
    )->with('Method');
    my($self, $args) = $rule->validate(@_);

    my $url = $self->_build_url(
        path => '/db/' . $self->database . '/users/' . $args->{name},
    );

    my $res = $self->{ua}->post($url, [], $self->json->encode({
        exists $args->{password} ? (password => $args->{password}) : (),
        exists $args->{admin} ? (admin => $args->{admin}) : (),
    }));
    $self->status($res);

    return $res->is_success ? 1 : ();
}

sub list_database_users {
    my $self = shift;

    my $url = $self->_build_url(
        path => '/db/' . $self->database . '/users',
    );

    my $res = $self->{ua}->get($url);
    $self->status($res);

    return $res->is_success ? $self->json->decode($res->content) : ();
}

sub show_database_user {
    state $rule = Data::Validator->new(
        name => { isa => 'Str' },
    )->with('Method');
    my($self, $args) = $rule->validate(@_);

    my $url = $self->_build_url(
        path => '/db/' . $self->database . '/users/' . $args->{name},
    );

    my $res = $self->{ua}->get($url);
    $self->status($res);

    return $res->is_success ? $self->json->decode($res->content) : ();
}

sub create_cluster_admin {
    state $rule = Data::Validator->new(
        name     => { isa => 'Str' },
        password => { isa => 'Str' },
    )->with('Method');
    my($self, $args) = $rule->validate(@_);

    my $url = $self->_build_url(
        path => '/cluster_admins',
    );

    my $res = $self->{ua}->post($url, [], $self->json->encode({
        name     => $args->{name},
        password => $args->{password},
    }));
    $self->status($res);

    return $res->is_success ? 1 : ();
}

sub delete_cluster_admin {
    state $rule = Data::Validator->new(
        name => { isa => 'Str' },
    )->with('Method');
    my($self, $args) = $rule->validate(@_);

    my $url = $self->_build_url(
        path => '/cluster_admins/' . $args->{name},
    );

    my $res = $self->{ua}->delete($url);
    $self->status($res);

    return $res->is_success ? 1 : ();
}

sub update_cluster_admin {
    state $rule = Data::Validator->new(
        name     => { isa => 'Str' },
        password => { isa => 'Str' },
    )->with('Method');
    my($self, $args) = $rule->validate(@_);

    my $url = $self->_build_url(
        path => '/cluster_admins/' . $args->{name},
    );

    my $res = $self->{ua}->post($url, [], $self->json->encode({
        exists $args->{password} ? (password => $args->{password}) : (),
    }));
    $self->status($res);

    return $res->is_success ? 1 : ();
}

sub list_cluster_admins {
    my $self = shift;

    my $url = $self->_build_url(
        path => '/cluster_admins',
    );

    my $res = $self->{ua}->get($url);
    $self->status($res);

    return $res->is_success ? $self->json->decode($res->content) : ();
}

# sub show_cluster_admin {
#     state $rule = Data::Validator->new(
#         name => { isa => 'Str' },
#     )->with('Method');
#     my($self, $args) = $rule->validate(@_);

#     my $url = $self->_build_url(
#         path => '/cluster_admins/' . $args->{name},
#     );

#     my $res = $self->{ua}->get($url);
#     $self->status($res);

#     return $res->is_success ? $self->json->decode($res->content) : ();
# }

### utils ################################################################
sub _build_url {
    state $rule = Data::Validator->new(
        path => { isa => 'Str' },
        qs   => { isa => 'HashRef', optional => 1 },
    )->with('Method');
    my($self, $args) = $rule->validate(@_);

    my $url = sprintf("%s://%s%s:%d%s",
                      ($self->ssl ? 'https' : 'http'),
                      ($self->username and $self->{password})
                          ? sprintf('%s:%s@', $self->username, $self->{password})
                          : '',
                      $self->host,
                      $self->port,
                      $args->{path},
                  );
    if (exists $args->{qs}) {
        my @qs;
        for my $k (keys %{ $args->{qs} }) {
            next unless defined $args->{qs}{$k};
            push @qs, join('=', uri_escape($k), uri_escape($args->{qs}{$k}));
        }
        $url .= '?' . join('&', @qs) if @qs;
    }

    debugf("url: %s", $url);
    return $url
}

1;

__END__