MT#18663 MT#20893 row bulk processing framework WIP #14
+support for handling multiple ngcp table versions at runtime +setting allowed_ips Change-Id: I7e87456d561407fd35b4518ed53a3c435c2e7798changes/38/7538/4
Rene Krenn
9 years ago
parent
0332ec4fc3
commit
a2d365e3df
@ -0,0 +1,142 @@
|
||||
package NGCP::BulkProcessor::Dao::Trunk::provisioning::voip_aig_sequence;
|
||||
use strict;
|
||||
|
||||
## no critic
|
||||
|
||||
use NGCP::BulkProcessor::Logging qw(
|
||||
getlogger
|
||||
rowinserted
|
||||
rowinsertskipped
|
||||
rowupdateskipped
|
||||
rowupdated
|
||||
);
|
||||
|
||||
use NGCP::BulkProcessor::ConnectorPool qw(
|
||||
get_provisioning_db
|
||||
);
|
||||
|
||||
use NGCP::BulkProcessor::SqlProcessor qw(
|
||||
checktableinfo
|
||||
insert_stmt
|
||||
copy_row
|
||||
);
|
||||
use NGCP::BulkProcessor::SqlRecord qw();
|
||||
|
||||
require Exporter;
|
||||
our @ISA = qw(Exporter NGCP::BulkProcessor::SqlRecord);
|
||||
our @EXPORT_OK = qw(
|
||||
gettablename
|
||||
check_table
|
||||
|
||||
get_id
|
||||
forupdate_increment
|
||||
|
||||
);
|
||||
|
||||
my $tablename = 'voip_aig_sequence';
|
||||
my $get_db = \&get_provisioning_db;
|
||||
|
||||
my $expected_fieldnames = [
|
||||
'id',
|
||||
];
|
||||
|
||||
my $indexes = {};
|
||||
|
||||
my $insert_unique_fields = [];
|
||||
|
||||
my $start_value = 1;
|
||||
my $increment = 1;
|
||||
|
||||
sub new {
|
||||
|
||||
my $class = shift;
|
||||
my $self = NGCP::BulkProcessor::SqlRecord->new($class,$get_db,
|
||||
$tablename,$expected_fieldnames,$indexes);
|
||||
|
||||
copy_row($self,shift,$expected_fieldnames);
|
||||
|
||||
return $self;
|
||||
|
||||
}
|
||||
|
||||
sub get_id {
|
||||
|
||||
check_table();
|
||||
my $db = &$get_db();
|
||||
my $table = $db->tableidentifier($tablename);
|
||||
|
||||
my $stmt = 'SELECT ' . $db->columnidentifier('id') . ' FROM ' . $table;
|
||||
return $db->db_get_value($stmt);
|
||||
|
||||
}
|
||||
|
||||
sub forupdate_increment {
|
||||
|
||||
my ($xa_db) = @_;
|
||||
|
||||
check_table();
|
||||
my $db = &$get_db();
|
||||
$xa_db //= $db;
|
||||
my $table = $db->tableidentifier($tablename);
|
||||
|
||||
my $stmt = 'SELECT ' . $db->columnidentifier('id') . ' FROM ' . $table . ' FOR UPDATE';
|
||||
my $id = $xa_db->db_get_value($stmt);
|
||||
if (defined $id) {
|
||||
$stmt = 'UPDATE ' . $table . ' SET ' . $db->columnidentifier('id') . ' = ? WHERE ' . $db->columnidentifier('id') . ' = ?';
|
||||
if ($xa_db->db_do($stmt,$id + $increment,$id)) {
|
||||
rowupdated($db,$tablename,getlogger(__PACKAGE__));
|
||||
return $id + $increment;
|
||||
} else {
|
||||
rowupdateskipped($db,$tablename,0,getlogger(__PACKAGE__));
|
||||
return undef;
|
||||
}
|
||||
} else {
|
||||
$stmt = insert_stmt($db,__PACKAGE__);
|
||||
if ($xa_db->db_do($stmt,$start_value)) {
|
||||
rowinserted($db,$tablename,getlogger(__PACKAGE__));
|
||||
return $start_value;
|
||||
} else {
|
||||
rowinsertskipped($db,$tablename,getlogger(__PACKAGE__));
|
||||
return undef;
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
sub buildrecords_fromrows {
|
||||
|
||||
my ($rows,$load_recursive) = @_;
|
||||
|
||||
my @records = ();
|
||||
my $record;
|
||||
|
||||
if (defined $rows and ref $rows eq 'ARRAY') {
|
||||
foreach my $row (@$rows) {
|
||||
$record = __PACKAGE__->new($row);
|
||||
|
||||
# transformations go here ...
|
||||
|
||||
push @records,$record;
|
||||
}
|
||||
}
|
||||
|
||||
return \@records;
|
||||
|
||||
}
|
||||
|
||||
sub gettablename {
|
||||
|
||||
return $tablename;
|
||||
|
||||
}
|
||||
|
||||
sub check_table {
|
||||
|
||||
return checktableinfo($get_db,
|
||||
__PACKAGE__,$tablename,
|
||||
$expected_fieldnames,
|
||||
$indexes);
|
||||
|
||||
}
|
||||
|
||||
1;
|
||||
@ -0,0 +1,170 @@
|
||||
package NGCP::BulkProcessor::Dao::Trunk::provisioning::voip_allowed_ip_groups;
|
||||
use strict;
|
||||
|
||||
## no critic
|
||||
|
||||
use NGCP::BulkProcessor::Logging qw(
|
||||
getlogger
|
||||
rowsdeleted
|
||||
rowinserted
|
||||
);
|
||||
|
||||
use NGCP::BulkProcessor::ConnectorPool qw(
|
||||
get_provisioning_db
|
||||
);
|
||||
|
||||
use NGCP::BulkProcessor::SqlProcessor qw(
|
||||
checktableinfo
|
||||
insert_stmt
|
||||
copy_row
|
||||
);
|
||||
use NGCP::BulkProcessor::SqlRecord qw();
|
||||
|
||||
require Exporter;
|
||||
our @ISA = qw(Exporter NGCP::BulkProcessor::SqlRecord);
|
||||
our @EXPORT_OK = qw(
|
||||
gettablename
|
||||
check_table
|
||||
|
||||
insert_rows
|
||||
delete_groupid
|
||||
|
||||
countby_groupid_ipnet
|
||||
|
||||
);
|
||||
|
||||
my $tablename = 'voip_allowed_ip_groups';
|
||||
my $get_db = \&get_provisioning_db;
|
||||
|
||||
my $expected_fieldnames = [
|
||||
'id',
|
||||
'group_id',
|
||||
'ipnet',
|
||||
];
|
||||
|
||||
my $indexes = {};
|
||||
|
||||
my $insert_unique_fields = [];
|
||||
|
||||
sub new {
|
||||
|
||||
my $class = shift;
|
||||
my $self = NGCP::BulkProcessor::SqlRecord->new($class,$get_db,
|
||||
$tablename,$expected_fieldnames,$indexes);
|
||||
|
||||
copy_row($self,shift,$expected_fieldnames);
|
||||
|
||||
return $self;
|
||||
|
||||
}
|
||||
|
||||
sub insert_rows {
|
||||
|
||||
my ($xa_db,$group_id,$ipnets) = @_;
|
||||
|
||||
check_table();
|
||||
my $db = &$get_db();
|
||||
$xa_db //= $db;
|
||||
|
||||
my $stmt = insert_stmt($db,__PACKAGE__);
|
||||
|
||||
my @ids = ();
|
||||
foreach my $ipnet (@$ipnets) {
|
||||
if ($xa_db->db_do($stmt,undef,$group_id,$ipnet)) {
|
||||
rowinserted($db,$tablename,getlogger(__PACKAGE__));
|
||||
push(@ids,$xa_db->db_last_insert_id());
|
||||
}
|
||||
}
|
||||
|
||||
return \@ids;
|
||||
}
|
||||
|
||||
sub delete_groupid {
|
||||
|
||||
my ($xa_db,$group_id) = @_;
|
||||
|
||||
check_table();
|
||||
my $db = &$get_db();
|
||||
$xa_db //= $db;
|
||||
my $table = $db->tableidentifier($tablename);
|
||||
|
||||
my $stmt = 'DELETE FROM ' . $table . ' WHERE ' .
|
||||
$db->columnidentifier('group_id') . ' = ?';
|
||||
my @params = ($group_id);
|
||||
|
||||
my $count;
|
||||
if ($count = $xa_db->db_do($stmt,@params)) {
|
||||
rowsdeleted($db,$tablename,$count,$count,getlogger(__PACKAGE__));
|
||||
return 1;
|
||||
} else {
|
||||
rowsdeleted($db,$tablename,0,0,getlogger(__PACKAGE__));
|
||||
return 0;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
sub countby_groupid_ipnet {
|
||||
|
||||
my ($group_id,$ipnet) = @_;
|
||||
|
||||
check_table();
|
||||
my $db = &$get_db();
|
||||
my $table = $db->tableidentifier($tablename);
|
||||
|
||||
my $stmt = 'SELECT COUNT(*) FROM ' . $table;
|
||||
my @params = ();
|
||||
my @terms = ();
|
||||
if ($group_id) {
|
||||
push(@terms,$db->columnidentifier('group_id') . ' = ?');
|
||||
push(@params,$group_id);
|
||||
}
|
||||
if ($ipnet) {
|
||||
push(@terms,$db->columnidentifier('ipnet') . ' = ?');
|
||||
push(@params,$ipnet);
|
||||
}
|
||||
|
||||
if ((scalar @terms) > 0) {
|
||||
$stmt .= ' WHERE ' . join(' AND ',@terms);
|
||||
}
|
||||
|
||||
return $db->db_get_value($stmt,@params);
|
||||
|
||||
}
|
||||
|
||||
sub buildrecords_fromrows {
|
||||
|
||||
my ($rows,$load_recursive) = @_;
|
||||
|
||||
my @records = ();
|
||||
my $record;
|
||||
|
||||
if (defined $rows and ref $rows eq 'ARRAY') {
|
||||
foreach my $row (@$rows) {
|
||||
$record = __PACKAGE__->new($row);
|
||||
|
||||
# transformations go here ...
|
||||
|
||||
push @records,$record;
|
||||
}
|
||||
}
|
||||
|
||||
return \@records;
|
||||
|
||||
}
|
||||
|
||||
sub gettablename {
|
||||
|
||||
return $tablename;
|
||||
|
||||
}
|
||||
|
||||
sub check_table {
|
||||
|
||||
return checktableinfo($get_db,
|
||||
__PACKAGE__,$tablename,
|
||||
$expected_fieldnames,
|
||||
$indexes);
|
||||
|
||||
}
|
||||
|
||||
1;
|
||||
@ -0,0 +1,116 @@
|
||||
package NGCP::BulkProcessor::Dao::mr441::billing::lnp_providers;
|
||||
use strict;
|
||||
|
||||
## no critic
|
||||
|
||||
use NGCP::BulkProcessor::Logging qw(
|
||||
getlogger
|
||||
);
|
||||
|
||||
use NGCP::BulkProcessor::ConnectorPool qw(
|
||||
get_billing_db
|
||||
);
|
||||
|
||||
use NGCP::BulkProcessor::SqlProcessor qw(
|
||||
checktableinfo
|
||||
copy_row
|
||||
);
|
||||
use NGCP::BulkProcessor::SqlRecord qw();
|
||||
|
||||
require Exporter;
|
||||
our @ISA = qw(Exporter NGCP::BulkProcessor::SqlRecord);
|
||||
our @EXPORT_OK = qw(
|
||||
gettablename
|
||||
check_table
|
||||
|
||||
findby_prefix
|
||||
);
|
||||
|
||||
my $tablename = 'lnp_providers';
|
||||
my $get_db = \&get_billing_db;
|
||||
|
||||
my $expected_fieldnames = [
|
||||
'id',
|
||||
'name',
|
||||
'prefix',
|
||||
#'authoritative',
|
||||
#'skip_rewrite',
|
||||
];
|
||||
|
||||
my $indexes = {};
|
||||
|
||||
my $insert_unique_fields = [];
|
||||
|
||||
sub new {
|
||||
|
||||
my $class = shift;
|
||||
my $self = NGCP::BulkProcessor::SqlRecord->new($class,$get_db,
|
||||
$tablename,$expected_fieldnames,$indexes);
|
||||
|
||||
copy_row($self,shift,$expected_fieldnames);
|
||||
|
||||
return $self;
|
||||
|
||||
}
|
||||
|
||||
sub findby_prefix {
|
||||
|
||||
my ($prefix,$load_recursive) = @_;
|
||||
|
||||
check_table();
|
||||
my $db = &$get_db();
|
||||
my $table = $db->tableidentifier($tablename);
|
||||
|
||||
my $stmt = 'SELECT * FROM ' . $table;
|
||||
my @params = ();
|
||||
my @terms = ();
|
||||
if ($prefix) {
|
||||
push(@terms,$db->columnidentifier('prefix') . ' = ?');
|
||||
push(@params,$prefix);
|
||||
}
|
||||
if ((scalar @terms) > 0) {
|
||||
$stmt .= ' WHERE ' . join(' AND ',@terms);
|
||||
}
|
||||
my $rows = $db->db_get_all_arrayref($stmt,@params);
|
||||
|
||||
return buildrecords_fromrows($rows,$load_recursive);
|
||||
|
||||
}
|
||||
|
||||
sub buildrecords_fromrows {
|
||||
|
||||
my ($rows,$load_recursive) = @_;
|
||||
|
||||
my @records = ();
|
||||
my $record;
|
||||
|
||||
if (defined $rows and ref $rows eq 'ARRAY') {
|
||||
foreach my $row (@$rows) {
|
||||
$record = __PACKAGE__->new($row);
|
||||
|
||||
# transformations go here ...
|
||||
|
||||
push @records,$record;
|
||||
}
|
||||
}
|
||||
|
||||
return \@records;
|
||||
|
||||
}
|
||||
|
||||
sub gettablename {
|
||||
|
||||
return $tablename;
|
||||
|
||||
}
|
||||
|
||||
sub check_table {
|
||||
|
||||
return checktableinfo($get_db,
|
||||
__PACKAGE__,$tablename,
|
||||
$expected_fieldnames,
|
||||
$indexes);
|
||||
|
||||
}
|
||||
|
||||
1;
|
||||
Loading…
Reference in new issue