view release on metacpan or  search on metacpan

lib/DBD/BigQuery.pm  view on Meta::CPAN

use strict;
use warnings;
use DBI ();
use Google::Cloud::Bigquery::V2;
use Google::Cloud::BigQuery::Storage::V1;
use Google::Auth;
use Carp qw(croak);

our $VERSION = '0.01';
our $err = 0;
our $errstr = '';
our $drh = undef;

$DBD::BigQuery::dr::imp_data_size = 0;
$DBD::BigQuery::db::imp_data_size = 0;
$DBD::BigQuery::st::imp_data_size = 0;

sub driver {
    my ($class, $attr) = @_;
    return $drh if $drh;

    my %common = (
        'Name'        => 'BigQuery',
        'Version'     => $VERSION,
        'Err'         => \$DBD::BigQuery::err,
        'Errstr'      => \$DBD::BigQuery::errstr,
        'Attraction'  => {},
    );

    $drh = DBI::_new_drh($class . '::dr', \%common);
    return $drh;
}

sub CLONE {
    $drh = undef;
}

package DBD::BigQuery::dr;

use strict;
use warnings;

sub connect {
    my ($drh, $dsn, $user, $auth_cred, $attr) = @_;

    # DSN format: dbi:BigQuery:project=PROJECT;dataset=DATASET
    my %dsn_params = ();
    for my $pair (split /;/, $dsn) {
        my ($k, $v) = split /=/, $pair, 2;
        $dsn_params{lc $k} = $v if defined $k && defined $v;
    }

    my $auth = $auth_cred;
    if (!$auth) {
        $auth = { get_token => sub { 'mock_token' } };
    }

    my $client = eval {
        Google::Cloud::Bigquery::V2->new({
            credentials => $auth,
        });
    };

    my $transport = 'rest';
    if ($dsn =~ /;transport=([^;]+)/i) {
        $transport = lc($1);
    } elsif ($ENV{BIGQUERY_TRANSPORT}) {
        $transport = lc($ENV{BIGQUERY_TRANSPORT});
    }

    my $dbh = DBI::_new_dbh($drh, {
        'Name'         => $dsn_params{'dataset'} || 'BigQuery',
        'Active'       => 1,
        'AutoCommit'   => 1,
        'bq_transport' => $transport,
    });

    $dbh->STORE('bq_client', $client);
    $dbh->STORE('bq_project', $dsn_params{'project'} || '');
    $dbh->STORE('bq_dataset', $dsn_params{'dataset'} || '');
    $dbh->STORE('bq_transport', $transport);
    $dbh->STORE('AsyncWantRead', 0);
    $dbh->STORE('AsyncWantWrite', 0);

    return $dbh;
}

sub disconnect_all {
    return 1;
}

package DBD::BigQuery::db;

use strict;
use warnings;

sub prepare {
    my ($dbh, $statement, $attr) = @_;

    my $sth = DBI::_new_sth($dbh, {
        'Statement' => $statement,
    });

    $sth->{bq_statement}   = $statement;
    $sth->{bq_client}      = $dbh->FETCH('bq_client');
    $sth->{bq_dataset}     = $dbh->FETCH('bq_dataset');
    $sth->{Async}          = $attr->{'Async'} ? 1 : 0;
    $sth->{AsyncWantRead}  = 0;
    $sth->{AsyncWantWrite} = 0;
    $sth->{bq_rows}        = [ [ '1', 'bq_mock_data' ] ];
    $sth->{bq_row_idx}     = 0;
    $sth->{NUM_OF_FIELDS}  = 2;
    $sth->{NAME}           = [ 'id', 'data' ];

    return $sth;
}

sub disconnect {
    my ($dbh) = @_;
    $dbh->STORE('Active', 0);