sasjs/adapter
1
0
Fork 0
mirror of https://github.com/sasjs/adapter.git synced 2026-01-08 13:00:05 +00:00
Code Issues Projects Releases Wiki Activity

gitfe Merge branches 'errorResponse' and 'master' of github.com:sasjs/adapter

Browse Source
This commit is contained in:
Mihajlo Medjedovic
2020-10-27 14:40:41 +01:00
parent 3f796b300d d5920c5885
commit 80707d77d9
34 changed files with 4529 additions and 162 deletions
Download Patch File Download Diff File Expand all files Collapse all files

138
src/SASViyaApiClient.ts
View File

@@ -52,7 +52,9 @@ export class SASViyaApiClient {
public set debug(value: boolean) {
this._debug = value
this.sessionManager.debug = value
if (this.sessionManager) {
this.sessionManager.debug = value
}
}
/**
@@ -145,41 +147,51 @@ export class SASViyaApiClient {
const promises = contextsList.map((context: any) => {
const linesOfCode = ['%put &=sysuserid;']
return this.executeScript(
`test-${context.name}`,
linesOfCode,
context.name,
accessToken,
null,
true
).catch(() => null)
return () =>
this.executeScript(
`test-${context.name}`,
linesOfCode,
context.name,
accessToken,
null,
true,
true
).catch((err) => err)
})
const results = await Promise.all(promises)
let results: any[] = []
for (const promise of promises) results.push(await promise())
results.forEach((result: any, index: number) => {
if (result) {
let sysUserId = ''
if (result && result.body && result.body.details) {
try {
const resultParsed = JSON.parse(result.body.details)
if (result.log) {
const sysUserIdLog = result.log
.split('\n')
.find((line: string) => line.startsWith('SYSUSERID='))
if (resultParsed && resultParsed.body) {
let sysUserId = ''
if (sysUserIdLog) {
sysUserId = sysUserIdLog.replace('SYSUSERID=', '')
const sysUserIdLog = resultParsed.body
.split('\n')
.find((line: string) => line.startsWith('SYSUSERID='))
if (sysUserIdLog) {
sysUserId = sysUserIdLog.replace('SYSUSERID=', '')
executableContexts.push({
createdBy: contextsList[index].createdBy,
id: contextsList[index].id,
name: contextsList[index].name,
version: contextsList[index].version,
attributes: {
sysUserId
}
})
}
}
} catch (error) {
throw error
}
executableContexts.push({
createdBy: contextsList[index].createdBy,
id: contextsList[index].id,
name: contextsList[index].name,
version: contextsList[index].version,
attributes: {
sysUserId
}
})
}
})
@@ -327,7 +339,9 @@ export class SASViyaApiClient {
originalContext = await this.getComputeContextByName(
contextName,
accessToken
).catch((_) => {})
).catch((err) => {
throw err
})
// Try to find context by id, when context name has been changed.
if (!originalContext) {
@@ -423,7 +437,8 @@ export class SASViyaApiClient {
contextName: string,
accessToken?: string,
data = null,
expectWebout = false
expectWebout = false,
waitForResult = true
): Promise<any> {
try {
const headers: any = {
@@ -435,7 +450,12 @@ export class SASViyaApiClient {
}
let executionSessionId: string
const session = await this.sessionManager.getSession(accessToken)
const session = await this.sessionManager
.getSession(accessToken)
.catch((err) => {
throw err
})
executionSessionId = session!.id
const jobArguments: { [key: string]: any } = {
@@ -474,7 +494,9 @@ export class SASViyaApiClient {
if (data) {
if (JSON.stringify(data).includes(';')) {
files = await this.uploadTables(data, accessToken)
files = await this.uploadTables(data, accessToken).catch((err) => {
throw err
})
jobVariables['_webin_file_count'] = files.length
@@ -505,7 +527,13 @@ export class SASViyaApiClient {
const { result: postedJob, etag } = await this.request<Job>(
`${this.serverUrl}/compute/sessions/${executionSessionId}/jobs`,
postJobRequest
)
).catch((err) => {
throw err
})
if (!waitForResult) {
return session
}
if (this.debug) {
console.log(`Job has been submitted for '${fileName}'.`)
@@ -521,7 +549,9 @@ export class SASViyaApiClient {
const { result: currentJob } = await this.request<Job>(
`${this.serverUrl}/compute/sessions/${executionSessionId}/jobs/${postedJob.id}`,
{ headers }
)
).catch((err) => {
throw err
})
let jobResult
let log
@@ -534,9 +564,13 @@ export class SASViyaApiClient {
{
headers
}
).then((res: any) =>
res.result.items.map((i: any) => i.line).join('\n')
)
.then((res: any) =>
res.result.items.map((i: any) => i.line).join('\n')
)
.catch((err) => {
throw err
})
}
if (jobStatus === 'failed' || jobStatus === 'error') {
@@ -547,6 +581,8 @@ export class SASViyaApiClient {
if (expectWebout) {
resultLink = `/compute/sessions/${executionSessionId}/filerefs/_webout/content`
} else {
return currentJob
}
if (resultLink) {
@@ -562,9 +598,13 @@ export class SASViyaApiClient {
{
headers
}
).then((res: any) =>
res.result.items.map((i: any) => i.line).join('\n')
)
.then((res: any) =>
res.result.items.map((i: any) => i.line).join('\n')
)
.catch((err) => {
throw err
})
return Promise.reject(
new ErrorResponse('Job execution failed', {
@@ -580,7 +620,11 @@ export class SASViyaApiClient {
})
}
await this.sessionManager.clearSession(executionSessionId, accessToken)
await this.sessionManager
.clearSession(executionSessionId, accessToken)
.catch((err) => {
throw err
})
return { result: jobResult?.result, log }
} catch (e) {
@@ -590,7 +634,9 @@ export class SASViyaApiClient {
linesOfCode,
contextName,
accessToken,
data
data,
false,
true
)
} else {
throw e
@@ -903,13 +949,16 @@ export class SASViyaApiClient {
* @param debug - sets the _debug flag in the job arguments.
* @param data - any data to be passed in as input to the job.
* @param accessToken - an optional access token for an authorized user.
* @param waitForResult - a boolean indicating if the function should wait for a result.
* @param expectWebout - a boolean indicating whether to expect a _webout response.
*/
public async executeComputeJob(
sasJob: string,
contextName: string,
debug: boolean,
data?: any,
accessToken?: string
accessToken?: string,
waitForResult = true,
expectWebout = false
) {
if (isRelativePath(sasJob) && !this.rootFolderName) {
throw new Error(
@@ -983,6 +1032,8 @@ export class SASViyaApiClient {
jobToExecute.code = code
}
if (!code) code = ''
const linesToExecute = code.replace(/\r\n/g, '\n').split('\n')
return await this.executeScript(
sasJob,
@@ -990,7 +1041,8 @@ export class SASViyaApiClient {
contextName,
accessToken,
data,
true
expectWebout,
waitForResult
)
}

53
src/SASjs.ts
View File

@@ -44,7 +44,7 @@ const defaultConfig: SASjsConfig = {
pathSASViya: '/SASJobExecution',
appLoc: '/Public/seedapp',
serverType: ServerType.SASViya,
debug: true,
debug: false,
contextName: 'SAS Job Execution compute context',
useComputeApi: false
}
@@ -670,6 +670,50 @@ export default class SASjs {
)
}
/**
* Kicks off execution of the given job via the compute API.
* @returns an object representing the compute session created for the given job.
* @param sasJob - the path to the SAS program (ultimately resolves to
* the SAS `_program` parameter to run a Job Definition or SAS 9 Stored
* Process). Is prepended at runtime with the value of `appLoc`.
* @param data - a JSON object containing one or more tables to be sent to
* SAS. Can be `null` if no inputs required.
* @param config - provide any changes to the config here, for instance to
* enable/disable `debug`. Any change provided will override the global config,
* for that particular function call.
* @param accessToken - a valid access token that is authorised to execute compute jobs.
* The access token is not required when the user is authenticated via the browser.
* @param waitForResult - a boolean that indicates whether the function needs to wait for execution to complete.
*/
public async startComputeJob(
sasJob: string,
data: any,
config: any = {},
accessToken?: string,
waitForResult?: boolean
) {
config = {
...this.sasjsConfig,
...config
}
this.isMethodSupported('startComputeJob', ServerType.SASViya)
if (!config.contextName) {
throw new Error(
'Context name is undefined. Please set a `contextName` in your SASjs or override config.'
)
}
return this.sasViyaApiClient?.executeComputeJob(
sasJob,
config.contextName,
data,
accessToken,
!!waitForResult,
false
)
}
private async executeJobViaComputeApi(
sasJob: string,
data: any,
@@ -689,13 +733,16 @@ export default class SASjs {
sasjsWaitingRequest.requestPromise.promise = new Promise(
async (resolve, reject) => {
const waitForResult = true
const expectWebout = true
this.sasViyaApiClient
?.executeComputeJob(
sasJob,
config.contextName,
config.debug,
data,
accessToken
accessToken,
waitForResult,
expectWebout
)
.then((response) => {
if (!config.debug) {

102
src/SessionManager.ts
View File

@@ -2,6 +2,12 @@ import { Session, Context, CsrfToken } from './types'
import { asyncForEach, makeRequest, isUrl } from './utils'
const MAX_SESSION_COUNT = 1
const RETRY_LIMIT: number = 3
let RETRY_COUNT: number = 0
const INTERNAL_SAS_ERROR = {
status: 304,
message: 'Not Modified'
}
export class SessionManager {
constructor(
@@ -27,19 +33,22 @@ export class SessionManager {
async getSession(accessToken?: string) {
await this.createSessions(accessToken)
this.createAndWaitForSession(accessToken)
await this.createAndWaitForSession(accessToken)
const session = this.sessions.pop()
const secondsSinceSessionCreation =
(new Date().getTime() - new Date(session!.creationTimeStamp).getTime()) /
1000
if (
!session!.attributes ||
secondsSinceSessionCreation >= session!.attributes.sessionInactiveTimeout
) {
await this.createSessions(accessToken)
const freshSession = this.sessions.pop()
return freshSession
}
return session
}
@@ -48,22 +57,37 @@ export class SessionManager {
method: 'DELETE',
headers: this.getHeaders(accessToken)
}
return await this.request<Session>(
`${this.serverUrl}/compute/sessions/${id}`,
deleteSessionRequest
).then(() => {
this.sessions = this.sessions.filter((s) => s.id !== id)
})
)
.then(() => {
this.sessions = this.sessions.filter((s) => s.id !== id)
})
.catch((err) => {
throw err
})
}
private async createSessions(accessToken?: string) {
if (!this.sessions.length) {
if (!this.currentContext) {
await this.setCurrentContext(accessToken)
await this.setCurrentContext(accessToken).catch((err) => {
throw err
})
}
await asyncForEach(new Array(MAX_SESSION_COUNT), async () => {
const createdSession = await this.createAndWaitForSession(accessToken)
const createdSession = await this.createAndWaitForSession(
accessToken
).catch((err) => {
throw err
})
this.sessions.push(createdSession)
}).catch((err) => {
throw err
})
}
}
@@ -73,13 +97,18 @@ export class SessionManager {
method: 'POST',
headers: this.getHeaders(accessToken)
}
const { result: createdSession, etag } = await this.request<Session>(
`${this.serverUrl}/compute/contexts/${this.currentContext!.id}/sessions`,
createSessionRequest
)
).catch((err) => {
throw err
})
await this.waitForSession(createdSession, etag, accessToken)
this.sessions.push(createdSession)
return createdSession
}
@@ -89,6 +118,8 @@ export class SessionManager {
items: Context[]
}>(`${this.serverUrl}/compute/contexts?limit=10000`, {
headers: this.getHeaders(accessToken)
}).catch((err) => {
throw err
})
const contextsList =
@@ -107,6 +138,8 @@ export class SessionManager {
}
this.currentContext = currentContext
Promise.resolve()
}
}
@@ -114,6 +147,7 @@ export class SessionManager {
const headers: any = {
'Content-Type': 'application/json'
}
if (accessToken) {
headers.Authorization = `Bearer ${accessToken}`
}
@@ -132,24 +166,41 @@ export class SessionManager {
'If-None-Match': etag
}
const stateLink = session.links.find((l: any) => l.rel === 'state')
return new Promise(async (resolve, _) => {
if (sessionState === 'pending') {
if (stateLink) {
if (this.debug) {
console.log('Polling session status... \n') // ?
}
const { result: state } = await this.request<string>(
const { result: state } = await this.requestSessionStatus<string>(
`${this.serverUrl}${stateLink.href}?wait=30`,
{
headers
},
'text'
)
).catch((err) => {
throw err
})
sessionState = state.trim()
if (this.debug) {
console.log(`Current state is '${sessionState}'\n`)
}
// There is an internal error present in SAS Viya 3.5
// Retry to wait for a session status in such case of SAS internal error
if (
sessionState === INTERNAL_SAS_ERROR.message &&
RETRY_COUNT < RETRY_LIMIT
) {
RETRY_COUNT++
resolve(this.waitForSession(session, etag, accessToken))
}
resolve(sessionState)
}
} else {
@@ -169,6 +220,7 @@ export class SessionManager {
[this.csrfToken.headerName]: this.csrfToken.value
}
}
return await makeRequest<T>(
url,
options,
@@ -177,6 +229,36 @@ export class SessionManager {
this.setCsrfToken(token)
},
contentType
)
).catch((err) => {
throw err
})
}
private async requestSessionStatus<T>(
url: string,
options: RequestInit,
contentType: 'text' | 'json' = 'json'
) {
if (this.csrfToken) {
options.headers = {
...options.headers,
[this.csrfToken.headerName]: this.csrfToken.value
}
}
return await makeRequest<T>(
url,
options,
(token) => {
this.csrfToken = token
this.setCsrfToken(token)
},
contentType
).catch((err) => {
if (err.status === INTERNAL_SAS_ERROR.status)
return { result: INTERNAL_SAS_ERROR.message }
throw err
})
}
}

184
src/utils/makeRequest.ts
View File

@@ -2,7 +2,7 @@ import { CsrfToken } from '../types'
import { needsRetry } from './needsRetry'
let retryCount: number = 0
let retryLimit: number = 5
const retryLimit: number = 5
export async function makeRequest<T>(
url: string,
@@ -18,57 +18,118 @@ export async function makeRequest<T>(
: (res: Response) => res.text()
let etag = null
const result = await fetch(url, request).then(async (response) => {
if (response.redirected && response.url.includes('SASLogon/login')) {
return Promise.reject({ status: 401 })
}
if (!response.ok) {
if (response.status === 403) {
const tokenHeader = response.headers.get('X-CSRF-HEADER')
const result = await fetch(url, request)
.then(async (response) => {
if (response.redirected && response.url.includes('SASLogon/login')) {
return Promise.reject({ status: 401 })
}
if (tokenHeader) {
const token = response.headers.get(tokenHeader)
callback({
headerName: tokenHeader,
value: token || ''
if (!response.ok) {
if (response.status === 403) {
const tokenHeader = response.headers.get('X-CSRF-HEADER')
if (tokenHeader) {
const token = response.headers.get(tokenHeader)
callback({
headerName: tokenHeader,
value: token || ''
})
retryRequest = {
...request,
headers: { ...request.headers, [tokenHeader]: token }
}
return await fetch(url, retryRequest).then((res) => {
etag = res.headers.get('ETag')
return responseTransform(res)
})
} else {
let body: any = await response.text().catch((err) => {
throw err
})
try {
body = JSON.parse(body)
body.message = `Forbidden. Check your permissions and user groups, and also the scopes granted when registering your CLIENT_ID. ${
body.message || ''
}`
body = JSON.stringify(body)
} catch (_) {}
return Promise.reject({ status: response.status, body })
}
} else {
let body: any = await response.text().catch((err) => {
throw err
})
retryRequest = {
...request,
headers: { ...request.headers, [tokenHeader]: token }
if (needsRetry(body)) {
if (retryCount < retryLimit) {
retryCount++
let retryResponse = await makeRequest(
url,
retryRequest || request,
callback,
contentType
).catch((err) => {
throw err
})
retryCount = 0
etag = retryResponse.etag
return retryResponse.result
} else {
retryCount = 0
throw new Error('Request retry limit exceeded')
}
}
return fetch(url, retryRequest).then((res) => {
etag = res.headers.get('ETag')
return responseTransform(res)
})
} else {
let body: any = await response.text()
if (response.status === 401) {
try {
body = JSON.parse(body)
try {
body = JSON.parse(body)
body.message = `Unauthorized request. Check your credentials(client, secret, access token). ${
body.message || ''
}`
body.message = `Forbidden. Check your permissions and user groups, and also the scopes granted when registering your CLIENT_ID. ${
body.message || ''
}`
body = JSON.stringify(body)
} catch (_) {}
body = JSON.stringify(body)
} catch (_) {}
}
return Promise.reject({ status: response.status, body })
}
} else {
let body: any = await response.text()
if (response.status === 204) {
return Promise.resolve()
}
const responseTransformed = await responseTransform(response).catch(
(err) => {
throw err
}
)
let responseText = ''
if (needsRetry(body)) {
if (typeof responseTransformed === 'string') {
responseText = responseTransformed
} else {
responseText = JSON.stringify(responseTransformed)
}
if (needsRetry(responseText)) {
if (retryCount < retryLimit) {
retryCount++
let retryResponse = await makeRequest(
const retryResponse = await makeRequest(
url,
retryRequest || request,
callback,
contentType
)
).catch((err) => {
throw err
})
retryCount = 0
etag = retryResponse.etag
@@ -80,57 +141,14 @@ export async function makeRequest<T>(
}
}
if (response.status === 401) {
try {
body = JSON.parse(body)
etag = response.headers.get('ETag')
body.message = `Unauthorized request. Check your credentials(client, secret, access token). ${
body.message || ''
}`
body = JSON.stringify(body)
} catch (_) {}
}
return Promise.reject({ status: response.status, body })
return responseTransformed
}
} else {
if (response.status === 204) {
return Promise.resolve()
}
const responseTransformed = await responseTransform(response)
let responseText = ''
if (typeof responseTransformed === 'string') {
responseText = responseTransformed
} else {
responseText = JSON.stringify(responseTransformed)
}
if (needsRetry(responseText)) {
if (retryCount < retryLimit) {
retryCount++
const retryResponse = await makeRequest(
url,
retryRequest || request,
callback,
contentType
)
retryCount = 0
etag = retryResponse.etag
return retryResponse.result
} else {
retryCount = 0
throw new Error('Request retry limit exceeded')
}
}
etag = response.headers.get('ETag')
return responseTransformed
}
})
})
.catch((err) => {
throw err
})
return { result, etag }
}