mirror of
https://github.com/sasjs/adapter.git
synced 2025-12-11 01:14:36 +00:00
Merge branch 'master' into sasjs-job
This commit is contained in:
@@ -147,18 +147,21 @@ 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,
|
||||
true
|
||||
).catch((err) => err)
|
||||
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 && result.body && result.body.details) {
|
||||
@@ -336,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) {
|
||||
@@ -445,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 } = {
|
||||
@@ -484,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
|
||||
|
||||
@@ -515,7 +527,9 @@ 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
|
||||
@@ -535,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
|
||||
@@ -548,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') {
|
||||
@@ -576,9 +596,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', {
|
||||
@@ -594,7 +618,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) {
|
||||
|
||||
@@ -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
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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 }
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user