Commit 9c161744 authored by David Nichols's avatar David Nichols
Browse files

refs #8 updated DB BBs and wrote initial tests

parent b6686d9f
Pipeline #5560 passed with stage
in 7 minutes and 19 seconds
# name: BB_RemoteDb2LocalDbImportBase
# version: 1.0
# desc: building block base class for high-performance remote DB -> local DB data transfers
# author: Qore Technologies, s.r.o.
%new-style
%require-types
%strict-args
%enable-all-warnings
#! transfers data from a remote DB to a local DB
class BB_RemoteDb2LocalDbImportBase {
import() {
# get mapper name
string mapper_name = UserApi::getConfigItemValue("remote2local-db-mapper-name");
# get mapper for data conversions
Mapper mapper = UserApi::getMapper(mapper_name);
# get remote instance
string remote = UserApi::getConfigItemValue("remote2local-db-remote-instance");
# get remote datasource name
string datasource = UserApi::getConfigItemValue("remote2local-db-remote-datasource");
# get remote table name
string table = UserApi::getConfigItemValue("remote2local-db-remote-table");
# get select stream options
hash<auto> opts;
# we use += to maintain the hash<auto> type
opts += UserApi::getConfigItemValue("remote2local-db-remote-options");
*string column_name = UserApi::getConfigItemValue("remote2local-db-remote-select-column");
auto value = UserApi::getConfigItemValue("remote2local-db-remote-select-value-template");
if (exists column_name) {
opts."select"."where"{column_name} = value;
} else if (exists value) {
doWarning("CONFIG-ERROR", "the \"remote2local-db-remote-select-value-template\" config item value %y has been "
"ignored because config item \"remote2local-db-remote-select-column\" is not set", value);
}
UserApi::logInfo("opening select stream %s:%s:%s -> mapper %y", remote, datasource, table, mapper_name);
UserApi::logDebug("select options: %N", opts);
# get remote instance for remote communication
DbRemoteReceive recv(remote,
datasource,
"select",
table,
opts,
);
on_error {
recv.disconnect();
mapper.discard();
mapper.rollback();
}
on_success {
mapper.flush();
mapper.commit();
}
hash<auto> ctx = getMapperConstantInput();
while (*hash<auto> h = recv.getData()) {
UserApi::logInfo("received block: %d rows", h.firstValue().lsize());
UserApi::logInfo("DEBUG: ctx: %y h: %y", ctx, h);
mapper.queueData(ctx + h);
}
}
private hash<auto> getMapperConstantInput() {
return {
"context": UserApi::getUserContextInfo(),
};
}
private doWarning(string err, string fmt) {
UserApi::logInfo("%s: %s", err, vsprintf(fmt, argv));
}
#! returns the config items as documented in the class documentation
static hash<string, hash<ConfigItemInfo>> getConfigItems() {
return {
# main configuration items
"remote2local-db-mapper-name": <ConfigItemInfo>{
"description": "the name of the mapper for the DB translations",
"config_group": "Remote DB Import",
},
"remote2local-db-remote-instance": <ConfigItemInfo>{
"description": "the name of the remote qorus instance hosting the remote table",
"config_group": "Remote DB Import",
},
"remote2local-db-remote-datasource": <ConfigItemInfo>{
"description": "the name of the datasource in the remote instance",
"config_group": "Remote DB Import",
},
"remote2local-db-remote-table": <ConfigItemInfo>{
"description": "the name of the table in the remote datasource",
"config_group": "Remote DB Import",
},
"remote2local-db-remote-options": <ConfigItemInfo>{
"type": "hash",
"default_value": {},
"description": "options for the DbRemoteReceive object",
"config_group": "Remote DB Import",
},
"remote2local-db-remote-select-column": <ConfigItemInfo>{
"type": "*string",
"default_value": NOTHING,
"description": "the name of the column for the select criteria in the remote datasource",
"config_group": "Remote DB Import",
},
"remote2local-db-remote-select-value-template": <ConfigItemInfo>{
"type": "*string",
"default_value": NOTHING,
"description": "the column value for the select criteria in the remote table",
"config_group": "Remote DB Import",
},
};
}
}
# END
# name: BB_RemoteDb2LocalDbImportWithRecovery
# version: 1.0
# desc: building block base step for high-performance remote DB -> local DB data transfers supporting recovery
# author: Qore Technologies, s.r.o.
# requires: BB_RemoteDb2LocalDbImportBase
%new-style
%require-types
%strict-args
%enable-all-warnings
#! transfers data from a remote DB to a local DB
class BB_RemoteDb2LocalDbImportWithRecovery inherits BB_RemoteDb2LocalDbImportBase {
#! returns True if local data is present, False if not
bool checkLocal() {
# get mapper name
string mapper_name = UserApi::getConfigItemValue("remote2local-db-mapper-name");
InboundTableMapper mapper = cast<InboundTableMapper>(UserApi::getMapper(mapper_name));
AbstractTable table = mapper.getTable();
string column_name = UserApi::getConfigItemValue("remote2local-db-recovery-column");
auto value = UserApi::getConfigItemValue("remote2local-db-recovery-value-template");
hash<auto> where_hash = {
column_name: value,
};
*hash<auto> row = table.findSingle(where_hash);
if (row) {
UserApi::logInfo("found data with %y = %y; transfer is already COMPLETE", column_name, value);
return True;
}
UserApi::logInfo("no data found with %y = %y; transfer will be retried", column_name, value);
return False;
}
static private *hash<string, hash<ConfigItemInfo>> getConfigItems() {
return BB_RemoteDb2LocalDbImportBase::getConfigItems() + {
# recovery items
"remote2local-db-recovery-column": <ConfigItemInfo>{
"description": "the name of the column for recovery",
"config_group": "Remote DB Import Recovery",
},
"remote2local-db-recovery-value-template": <ConfigItemInfo>{
"description": "value for recovery",
"config_group": "Remote DB Import Recovery",
},
};
}
}
# END
# name: BB_RemoteDb2LocalDbImportStep
# version: 1.0
# desc: building block base step for high-performance DB -> DB data transfers
# desc: building block base step for high-performance remote DB -> local DB data transfers
# author: Qore Technologies, s.r.o.
# requires: BB_RemoteDb2LocalDbImportWithRecovery
%new-style
%require-types
%strict-args
%enable-all-warnings
#! this step creates an account in the billing system
class BB_RemoteDb2LocalDbImportStep inherits QorusNormalStep {
#! transfers data from a remote DB to a local DB
class BB_RemoteDb2LocalDbImportStep inherits QorusNormalStep, BB_RemoteDb2LocalDbImportWithRecovery {
primary() {
# get mapper name
string mapper_name = getConfigItemValue("db2db-mapper-name");
# get mapper for data conversions
Mapper mapper = getMapper(mapper_name);
# get remote instance
string remote = getConfigItemValue("db2db-remote-instance");
# get remote datasource name
string datasource = getConfigItemValue("db2db-remote-datasource");
# get remote table name
string table = getConfigItemValue("db2db-remote-table");
# get select stream options
hash<auto> opts;
# we use += to maintain the hash<auto> type
opts += getConfigItemValue("db2db-remote-options");
*string column_name = getConfigItemValue("db2db-remote-select-column");
auto value = getConfigItemValue("db2db-remote-select-value-template");
if (exists column_name) {
opts."select"."where"{column_name} = value;
} else if (exists value) {
stepWarning("CONFIG-ERROR", "the \"db2db-remote-select-value-template\" config item value %y has been "
"ignored because config item \"db2db-remote-select-column\" is not set", value);
}
logInfo("opening select stream %s:%s:%s -> mapper %y", remote, datasource, table, mapper_name);
logDebug("select options: %N", opts);
# get remote instance for remote communication
DbRemoteReceive recv(remote,
datasource,
"select",
table,
opts,
);
on_error {
recv.disconnect();
mapper.discard();
mapper.rollback();
}
on_success {
mapper.flush();
mapper.commit();
}
while (*hash<auto> h = recv.getData()) {
log(LL_INFO, "received block: %d rows", h.firstValue().lsize());
mapper.queueData(h);
}
import();
}
string validation() {
# get mapper name
string mapper_name = getConfigItemValue("db2db-mapper-name");
InboundTableMapper mapper = cast<InboundTableMapper>(getMapper(mapper_name));
AbstractTable table = mapper.getTable();
string column_name = getConfigItemValue("db2db-recovery-column");
auto value = getConfigItemValue("db2db-recovery-value-template");
return checkLocal()
? OMQ::StatComplete
: OMQ::StatRetry;
}
hash<auto> where_hash = {
column_name: value,
private hash<auto> getMapperConstantInput() {
return {
"context": UserApi::getUserContextInfo(),
"static": WorkflowApi::getStaticData(),
"dynamic": WorkflowApi::getDynamicData(),
"keys": WorkflowApi::getOrderKeys(),
};
}
*hash<auto> row = table.findSingle(where_hash);
if (row) {
logInfo("found data with %y = %y; step is already COMPLETE", column_name, value);
return OMQ::StatComplete;
}
logInfo("no data found with %y = %y; step will be retried", column_name, value);
return OMQ::StatRetry;
private doWarning(string err, string fmt) {
stepWarning(err, vsprintf(fmt, argv));
}
private *hash<string, hash<ConfigItemInfo>> getConfigItemsImpl() {
return {
# main configuration items
"db2db-mapper-name": <ConfigItemInfo>{
"description": "the name of the mapper for the DB translations",
"config_group": "Remote DB Import",
},
"db2db-remote-instance": <ConfigItemInfo>{
"description": "the name of the remote qorus instance hosting the remote table",
"config_group": "Remote DB Import",
},
"db2db-remote-datasource": <ConfigItemInfo>{
"description": "the name of the datasource in the remote instance",
"config_group": "Remote DB Import",
},
"db2db-remote-table": <ConfigItemInfo>{
"description": "the name of the table in the remote datasource",
"config_group": "Remote DB Import",
},
"db2db-remote-options": <ConfigItemInfo>{
"type": "hash",
"default_value": {},
"description": "options for the DbRemoteReceive object",
"config_group": "Remote DB Import",
},
"db2db-remote-select-column": <ConfigItemInfo>{
"type": "*string",
"default_value": NOTHING,
"description": "the name of the column for the select criteria in the remote datasource",
"config_group": "Remote DB Import",
},
"db2db-remote-select-value-template": <ConfigItemInfo>{
"type": "*string",
"default_value": NOTHING,
"description": "the column value for the select criteria in the remote table",
"config_group": "Remote DB Import",
},
# recovery items
"db2db-recovery-column": <ConfigItemInfo>{
"description": "the name of the column for recovery",
"config_group": "Remote DB Import Recovery",
},
"db2db-recovery-value-template": <ConfigItemInfo>{
"description": "value for recovery",
"config_group": "Remote DB Import Recovery",
},
};
return BB_RemoteDb2LocalDbImportWithRecovery::getConfigItems();
}
}
# END
# -*- mode: qore; indent-tabs-mode: nil -*-
# test workflow definition
#
# Qorus Integration Engine
%new-style
our string format_version = "2.6";
our hash<auto> groups = {
"test": {"desc": "test interfaces"},
"bb-test": {"desc": "building block test interfaces"},
};
our list<auto> steps = (
{
"name": "bb-test-db-remote-2-local-step:1.0",
"desc": "BB test step",
"classname": "BB_TestDbRemote2LocalStep:1.0",
},
);
our hash<auto> workflows."BB-TEST-DB-WORKFLOW"."1.0" = {
"desc": "BB test: remote 2 local DB step",
"author": "Qore Technologies, s.r.o.",
"steps": steps,
"groups": ("test", "bb-test",),
"mappers": (
"bb-test-db-step:1.0",
),
"sla_threshold": 20,
"autostart": 0,
"remote": False,
};
# -*- mode: qore; indent-tabs-mode: nil -*-
# @file BB_TestDbRemote2Local.qsm Qorus Integration System Salesforce.com account provisioning demo user schema module
/* BB_TestDbRemote2Local.qsm Copyright 2016 - 2019 Qore Technologies, s.r.o.
Permission is hereby granted, free of charge, to any person obtaining a
copy of this software and associated documentation files (the "Software"),
to deal in the Software without restriction, including without limitation
the rights to use, copy, modify, merge, publish, distribute, sublicense,
and/or sell copies of the Software, and to permit persons to whom the
Software is furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in
all copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
DEALINGS IN THE SOFTWARE.
*/
%requires qore >= 0.9.3
module BB_TestDbRemote2Local {
version = "1.0";
desc = "BB test schema module";
author = "Qore Technologies, s.r.o. <info@qoretechnologies.com>";
url = "http://www.qoretechnologies.com";
}
# here we add fallback paths to the QORE_MODULE_DIR search path,
# in case QORE_MODULE_DIR is not set properly for Qorus
%append-module-path /var/opt/qorus/qlib:$OMQ_DIR/qlib:/opt/qorus/qlib
%requires Schema
%requires SqlUtil
%new-style
%strict-args
%require-types
%strict-args
%enable-all-warnings
# private namespace for private schema declarations
namespace Private {
const GenericOptions = {
"replace": True,
};
const IndexOptions = {
"driver": {
"oracle": {
"compute_statistics": True,
},
},
};
const ColumnOptions = {
"driver": {
"oracle": {"character_semantics": True,},
},
};
const T_BbLocalTest = {
"columns": {
"remote_id": c_varchar(20, True, "PK ID field"),
"remote_batch_id": c_varchar(20, True, "batch ID field"),
"qorus_wfiid": c_int(True),
},
"primary_key": {"name": "pk_bb_local", "columns": ("remote_id")},
"indexes": {
"sk_bb_local_q_wfiid": ("columns": ("qorus_wfiid")),
},
};
const T_BbRemoteTest = {
"columns": {
"id": c_varchar(20, True, "PK ID field"),
"batch_id": c_varchar(20, True, "batch ID field"),
},
"primary_key": {"name": "pk_bb_remote", "columns": ("id")},
"indexes": {
"sk_bb_remote_batch_id": ("columns": ("batch_id")),
},
};
const Tables = {
"bb_local": T_BbLocalTest,
"bb_remote": T_BbRemoteTest,
};
}
public namespace BB_TestDbRemote2Local {
public string sub get_datasource_name() {
return "omquser";
}
public BB_TestDbRemote2Local sub get_user_schema(AbstractDatasource ds, *string dts, *string its) {
return new BB_TestDbRemote2Local(ds, dts, its);
}
public class BB_TestDbRemote2Local inherits AbstractSchema {
public {
const SchemaName = "BB_TestDbRemote2Local";
const SchemaVersion = "1.0";
}
constructor(AbstractDatasource ds, *string dts, *string its) : AbstractSchema(ds, dts, its) {
}
private string getNameImpl() {
return SchemaName;
}
private string getVersionImpl() {
return SchemaVersion;
}
private *hash getTablesImpl() {
return Tables;
}
private *hash getSequencesImpl() {
return;
}
private *hash getIndexOptionsImpl() {
return IndexOptions;
}
private *hash getGenericOptionsImpl() {
return GenericOptions;
}
private *hash getColumnOptionsImpl() {
return ColumnOptions;
}
}
}
# name: BB_TestDbRemote2LocalStep
# version: 1.0
# desc: test step logic
# author: Qore Technologies, s.r.o.
# requires: BB_RemoteDb2LocalDbImportStep
%new-style
%require-types
%strict-args
%enable-all-warnings
#! this step creates an account in the billing system
class BB_TestDbRemote2LocalStep inherits BB_RemoteDb2LocalDbImportStep {
private {
#! default values for config items in this step
const DefaultConfigItemValues = {
"remote2local-db-mapper-name": "bb-test-db-step",
"remote2local-db-remote-datasource": "omquser",
"remote2local-db-remote-table": "bb_remote",
"remote2local-db-remote-select-column": "batch_id",
"remote2local-db-remote-select-value-template": "$static:batch_id",
# recovery items
"remote2local-db-recovery-column": "qorus_wfiid",
"remote2local-db-recovery-value-template": "$local:workflow_instanceid",
};
}
#! config items for this step
/** - rest-connection-name: the name of the billing system REST connection
*/
private *hash<string, hash<ConfigItemInfo>> getConfigItemsImpl() {
hash<string, hash<ConfigItemInfo>> rv = BB_RemoteDb2LocalDbImportStep::getConfigItemsImpl();
map rv{$1.key}.default_value = $1.value, DefaultConfigItemValues.pairIterator();
return rv;
}
}
# END
# name: bb-test-db-step
# version: 1.0
# desc: BB test mapper
# type: InboundTableMapper
# author: Qore Technologies, s.r.o.
# parse-options: PO_NEW_STYLE
# define-group: test: test interfaces
# define-group: bb-test: building block test interfaces
# groups: test, bb-test
OPTION: datasource: "omquser"
OPTION: table: "bb_local"
OPTION: input: (
"static": "static data",
"dynamic": "dynamic data",
"keys": "workflow order keys",
"context": "current execution context info",
"id": "input ID",
"batch_id": "input batch ID",
"qorus_wfiid": "source Qorus interface instance ID",
)
FIELD:remote_id: {"name": "id"}
FIELD:remote_batch_id: {"name": "batch_id"}
FIELD:qorus_wfiid: {"name": "context.workflow_instanceid"}
# END
#! /usr/bin/env qore
%new-style
%strict-args
%require-types
%enable-all-warnings
%requires QorusInterfaceTest
%requires Util
%requires BulkSqlUtil
%exec-class Test
class Test inherits QorusWorkflowTest {
private {
const ConnectionName = get_random_string();
const WorkflowName = "BB-TEST-DB-WORKFLOW";
const BatchId = get_random_string();
const Id1 = get_random_string();
const Id2 = get_random_string();
const RemoteData = (
{"id": Id1, "batch_id": BatchId},
{"id": Id2, "batch_id": BatchId},
);
const StepConfig = {
"remote2local-db-remote-instance": ConnectionName,
};
}
constructor() : QorusWorkflowTest("BB-TEST-DB-WORKFLOW", "1.0", \ARGV) {
addTestCase("test", \mainTest());
set_return_value(main());
}
globalSetUp() {
# create remote connection to local instance
string url = qorus_get_local_url();
qrest.post("remote/qorus", {
"name": ConnectionName,