224 lines
9.4 KiB
VB.net
224 lines
9.4 KiB
VB.net
Option Explicit On
|
|
|
|
Imports System.IO
|
|
Imports DigitalData.Modules.Interfaces
|
|
Imports DigitalData.Modules.Jobs
|
|
Imports DigitalData.Modules.Config
|
|
Imports DigitalData.Modules.Logging
|
|
Imports Newtonsoft.Json.Linq
|
|
Imports System.Collections.Generic
|
|
Imports System.Linq
|
|
Imports System.Text.RegularExpressions
|
|
Imports DigitalData.Modules.Database
|
|
Imports System.Data
|
|
|
|
Public Class GraphQLJob
|
|
Inherits JobBase
|
|
Implements IJob(Of GraphQLArgs)
|
|
|
|
Private _GraphQL As GraphQLInterface = Nothing
|
|
|
|
Private Const PLACEHOLDER_STATIC = "STATIC:"
|
|
|
|
Public Sub New(LogConfig As LogConfig, MSSQL As MSSQLServer)
|
|
MyBase.New(LogConfig, Nothing, MSSQL)
|
|
End Sub
|
|
|
|
Public Sub Start(Args As GraphQLArgs) Implements IJob(Of GraphQLArgs).Start
|
|
Try
|
|
Dim oConfigPath As String = Args.QueryConfigPath
|
|
Dim oConfigManager As New ConfigManager(Of GraphQLConfig)(_LogConfig, oConfigPath)
|
|
|
|
With oConfigManager.Config
|
|
_GraphQL = New GraphQLInterface(_LogConfig, .BaseUrl, .Email, .Password, .CertificateFingerprint)
|
|
End With
|
|
|
|
' Login to get cookie
|
|
_Logger.Debug("Logging in")
|
|
Dim oLoginResponse = _GraphQL.Login()
|
|
|
|
' save cookie for future requests
|
|
_GraphQL.SaveCookies(oLoginResponse.Cookies.Item(0))
|
|
|
|
_Logger.Debug("Loading Queries")
|
|
|
|
' Load query data from TBCUST_JOBRUNNER_QUERY
|
|
Dim oQueryTable As DataTable = _MSSQL.GetDatatable("SELECT * FROM TBCUST_JOBRUNNER_QUERY ORDER BY OPERATION_NAME, CLEAR_BEFORE_FILL ASC")
|
|
Dim oQueryList As New List(Of GraphQL.Query)
|
|
|
|
' Save query data to business objects
|
|
For Each oRow As DataRow In oQueryTable.Rows
|
|
Dim oQuery As New GraphQL.Query With {
|
|
.Id = oRow.Item("GUID"),
|
|
.Name = oRow.Item("TITLE"),
|
|
.ClearBeforeFill = oRow.Item("CLEAR_BEFORE_FILL"),
|
|
.ConnectionId = oRow.Item("CON_ID"), ' TODO: Connection String?
|
|
.DestinationTable = oRow.Item("DESTINATION_TABLE"),
|
|
.OperationName = oRow.Item("OPERATION_NAME"),
|
|
.MappingBasePath = oRow.Item("MAPPING_BASE_PATH"),
|
|
.QueryString = oRow.Item("QUERY_STRING"),
|
|
.QueryConstraint = oRow.Item("QUERY_CONSTRAINT")
|
|
}
|
|
oQueryList.Add(oQuery)
|
|
Next
|
|
|
|
_Logger.Debug("Getting the data from GraphQL")
|
|
|
|
For Each oQuery As GraphQL.Query In oQueryList
|
|
Try
|
|
_Logger.NewBlock($"Query [{oQuery.Name}]")
|
|
|
|
Dim oConnectionId As Integer = oQuery.ConnectionId
|
|
Dim oConnectionString = _MSSQL.Get_ConnectionStringforID(oConnectionId)
|
|
|
|
Dim oDatabase As New MSSQLServer(_LogConfig, oConnectionString)
|
|
|
|
' Reset all records to status = 0
|
|
_Logger.Info("Resetting data with constraint [{1}]", oQuery.Name, oQuery.QueryConstraint)
|
|
|
|
Dim oResetSQL = $"UPDATE {oQuery.DestinationTable} SET STATUS = 0"
|
|
If oQuery.QueryConstraint <> String.Empty Then
|
|
oResetSQL &= $" WHERE {oQuery.QueryConstraint}"
|
|
End If
|
|
_MSSQL.ExecuteNonQuery(oResetSQL)
|
|
|
|
_Logger.Info("Getting data..", oQuery.Name)
|
|
|
|
' get the data from GraphQL
|
|
Dim oDataResponse = _GraphQL.GetData(oQuery.QueryString, oQuery.OperationName)
|
|
Dim oResult As String
|
|
|
|
' write data to string
|
|
Using oStream = oDataResponse.GetResponseStream()
|
|
Using oReader As New StreamReader(oStream)
|
|
oResult = oReader.ReadToEnd()
|
|
End Using
|
|
End Using
|
|
|
|
' Fill the query object with field mapping data from TBCUST_JOBRUNNER_QUERY_MAPPING
|
|
Dim oSQL As String = "SELECT t2.* FROM TBCUST_JOBRUNNER_QUERY_MAPPING t
|
|
JOIN TBCUST_JOBRUNNER_MAPPING t2 ON t.MAPPING_ID = t2.GUID
|
|
WHERE t.QUERY_ID = {0}"
|
|
Dim oMappingTable As DataTable = _MSSQL.GetDatatable(String.Format(oSQL, oQuery.Id))
|
|
|
|
For Each oMapping As DataRow In oMappingTable.Rows
|
|
oQuery.MappingFields.Add(New GraphQL.FieldMapping With {
|
|
.DestinationColumn = oMapping.Item("DestinationColumn"),
|
|
.SourcePath = oMapping.Item("SourcePath")
|
|
})
|
|
Next
|
|
|
|
' Handle the response from GraphQL and insert Data
|
|
Dim oQueryHandleResult = HandleResponse(oResult, oQuery, oDatabase)
|
|
|
|
If IsNothing(oQueryHandleResult) Then
|
|
Continue For
|
|
End If
|
|
|
|
' Finally delete all old records
|
|
Dim oDeleteSQL = $"DELETE FROM {oQuery.DestinationTable} WHERE STATUS = 0"
|
|
If oQuery.QueryConstraint <> String.Empty Then
|
|
oDeleteSQL &= $" AND {oQuery.QueryConstraint}"
|
|
End If
|
|
|
|
_Logger.Info("Success, deleting old records..", oQuery.Name)
|
|
_MSSQL.ExecuteNonQuery(oDeleteSQL)
|
|
|
|
Catch ex As Exception
|
|
_Logger.Warn("Error while getting Data for Name/OperationName [{0}]/[{1}]", oQuery.Name, oQuery.OperationName)
|
|
_Logger.Error(ex)
|
|
|
|
_Logger.Info("Failure, deleting new records..", oQuery.Name)
|
|
|
|
' If a crash happens, delete all records which were inserted in this run,
|
|
' thus going back to the previous state
|
|
Dim oDeleteSQL = $"DELETE FROM {oQuery.DestinationTable} WHERE STATUS = 1"
|
|
If oQuery.QueryConstraint <> String.Empty Then
|
|
oDeleteSQL &= $" AND {oQuery.QueryConstraint}"
|
|
End If
|
|
_MSSQL.ExecuteNonQuery(oDeleteSQL)
|
|
Finally
|
|
_Logger.EndBlock()
|
|
End Try
|
|
Next
|
|
|
|
' logout
|
|
_Logger.Debug("Logging out")
|
|
Dim oLogoutResponse = _GraphQL.Logout()
|
|
Catch ex As Exception
|
|
_Logger.Error(ex)
|
|
Throw ex
|
|
End Try
|
|
End Sub
|
|
|
|
Private Function HandleResponse(JsonString As String, QueryData As GraphQL.Query, DB As Database.MSSQLServer) As GraphQL.Query
|
|
Dim oObj As JObject = JObject.Parse(JsonString)
|
|
Dim oResultList As JToken
|
|
|
|
If _GraphQL.ReadJSONPathFragmented(oObj, QueryData.MappingBasePath) = False Then
|
|
_Logger.Warn("There is an error in the MappingBasePath [{1}] configuration of query [{0}]", QueryData.Name, QueryData.MappingBasePath)
|
|
End If
|
|
|
|
Try
|
|
oResultList = oObj.SelectToken(QueryData.MappingBasePath, errorWhenNoMatch:=True)
|
|
Catch ex As Exception
|
|
_Logger.Warn("HandleResponse: Could not find BasePath: [{0}] for query [{1}]", QueryData.MappingBasePath, QueryData.Name)
|
|
_Logger.Error(ex)
|
|
Return Nothing
|
|
End Try
|
|
|
|
If oResultList Is Nothing Then
|
|
_Logger.Warn("HandleResponse: Could not find BasePath: [{0}] for query [{1}]", QueryData.MappingBasePath, QueryData.Name)
|
|
Return Nothing
|
|
End If
|
|
|
|
_Logger.Info("HandleResponse: Processing Queue [{0}] with [{1}] Items", QueryData.Name, oResultList.Count)
|
|
|
|
For Each oResultItem As JToken In oResultList
|
|
Try
|
|
Dim oValues As New List(Of String)
|
|
Dim oKeys As New List(Of String)
|
|
|
|
For Each oMapping In QueryData.MappingFields
|
|
Dim oValue As String = String.Empty
|
|
|
|
If oMapping.SourcePath.StartsWith(PLACEHOLDER_STATIC) Then
|
|
oValue = oMapping.SourcePath.Replace(PLACEHOLDER_STATIC, String.Empty)
|
|
Else
|
|
Dim oToken = oResultItem.SelectToken(oMapping.SourcePath)
|
|
|
|
If oToken Is Nothing Then
|
|
_Logger.Warn("HandleResponse: Could not find value at SourcePath: {0}", oMapping.SourcePath)
|
|
oValue = String.Empty
|
|
Else
|
|
oValue = oToken.ToString
|
|
End If
|
|
End If
|
|
|
|
oValues.Add(oValue)
|
|
oKeys.Add(oMapping.DestinationColumn)
|
|
Next
|
|
|
|
Dim oColumnValues = oValues.
|
|
Select(Function(Value) Regex.Replace(Value, "'", "''")).
|
|
Select(Function(Value) $"'{Value}'").
|
|
ToList()
|
|
Dim oValueString = String.Join(",", oColumnValues)
|
|
|
|
Dim oColumns = String.Join(",", oKeys.ToArray)
|
|
Dim oSQL As String = $"INSERT INTO {QueryData.DestinationTable} ({oColumns}) VALUES ({oValueString})"
|
|
|
|
DB.ExecuteNonQuery(oSQL)
|
|
Catch ex As Exception
|
|
_Logger.Error(ex)
|
|
End Try
|
|
Next
|
|
|
|
Return QueryData
|
|
End Function
|
|
|
|
Public Function ShouldStart(Arguments As GraphQLArgs) As Boolean Implements IJob(Of GraphQLArgs).ShouldStart
|
|
Return Arguments.Enabled
|
|
End Function
|
|
End Class
|