marker/shuffle-cracked
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

#204: Worked on multi-files. Found coroutine async issues

Browse Source
This commit is contained in:
frikky
2020-12-01 16:49:54 +01:00
parent 099e73916e
commit 93491f5ffb
4 changed files with 64 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app_sdk/app_base.py
+52 -15
View File
@@ -56,8 +56,11 @@ class AppBase:
# - How can you download / stream a file?
# - Can you decide if you want a stream or the files directly?
def get_file(self, value):
print("INSIDE GET_FILE")
full_execution = self.full_execution
org_id = full_execution["workflow"]["execution_org"]["id"]
print("SHOULD GET FILES BASED ON ORG %s, workflow %s and value(s) %s" % (org_id, full_execution["workflow"]["id"], value))
get_path = "/api/v1/files/%s?execution_id=%s" % (value, full_execution["execution_id"])
@@ -317,7 +320,7 @@ class AppBase:
try:
return int(data)
except ValueError:
print("ValueError while casting %s" % data)
print("ValueError while casting %s to int" % data)
return data
if "lower" in thistype:
return data.lower()
@@ -329,15 +332,19 @@ class AppBase:
return data.strip()
if "split" in thistype:
return data.split()
if "len" in thistype or "length" in thistype:
if "len" in thistype or "length" in thistype or "lenght" in thistype:
tmp = ""
try:
tmp = json.loads(data)
tmpdata = data.replace("\'", "\"")
tmp = json.loads(tmpdata)
except:
print("Passing bug")
pass
if isinstance(tmp, list):
return str(len(tmp))
return len(tmp)
elif isinstance(tmp, object):
return len(tmp)
return str(len(data))
if "parse" in thistype:
@@ -383,7 +390,7 @@ class AppBase:
#print("Running %s" % data)
# Look for the INNER wrapper first, then move out
wrappers = ["int", "number", "lower", "upper", "trim", "strip", "split", "parse", "len", "length"]
wrappers = ["int", "number", "lower", "upper", "trim", "strip", "split", "parse", "len", "length", "lenght"]
found = False
for wrapper in wrappers:
if wrapper not in data.lower():
@@ -413,6 +420,7 @@ class AppBase:
continue
parsed_value = parse_type(innervalue[0], thistype.lower())
print("Parsed value from %s: %s" % (thistype, parsed_value))
return parsed_value
print("DATA: %s\n" % data)
@@ -1057,13 +1065,40 @@ class AppBase:
minlength = len(json_replacement)
tmpitem = tmpitem.replace(actualitem[0][0], replacement, 1)
params[parameter["name"]] = tmpitem
multi_execution_lists.append(json_replacement)
multi_parameters[parameter["name"]] = json_replacement
#print("LENGTH OF ARR: %d" % len(resultarray))
#print("RESULTARRAY: %s" % resultarray)
print("MULTI finished: %s" % replacement)
# This code handles files.
print("(1) ------------ PARAM: %s" % parameter["schema"]["type"])
resultarray = []
isfile = False
try:
if parameter["schema"]["type"] == "file" and len(value) > 0:
print("(1) SHOULD HANDLE FILE IN MULTI. Get based on value %s" % tmpitem)
# This is silly :)
for tmp_file_split in json.loads(tmpitem):
print("PRE GET FILE %s" % tmp_file_split)
file_value = self.get_file(tmp_file_split)
print("PRE AWAIT %s" % file_value)
await file_value
print("POST AWAIT %s" % file_value)
resultarray.append(file_value)
print("(1) FILE VALUE FOR VAL %s: %s" % (tmp_file_split, file_value))
isfile = True
except KeyError as e:
print("(1) SCHEMA ERROR IN FILE HANDLING: %s" % e)
except json.decoder.JSONDecodeError as e:
print("(1) JSON ERROR IN FILE HANDLING: %s" % e)
if not isfile:
params[parameter["name"]] = tmpitem
multi_parameters[parameter["name"]] = json_replacement
else:
print("Resultarray: %s" % resultarray)
params[parameter["name"]] = resultarray
multi_parameters[parameter["name"]] = resultarray
multi_execution_lists.append(json_replacement)
print("MULTI finished: %s" % json_replacement)
else:
# This is here to handle for loops within variables.. kindof
# 1. Find the length of the longest array
@@ -1109,14 +1144,15 @@ class AppBase:
# This code handles files.
print("------------ PARAM: %s" % parameter["schema"]["type"])
isfile = False
try:
if parameter["schema"]["type"] == "file" and len(value) > 0:
print("SHOULD HANDLE FILE IN MULTI. Get based on value %s" % parameter["value"])
file_value = self.get_file(tmpitem)
file_value = await self.get_file(tmpitem)
print("FILE VALUE FOR VAL %s: %s" % (tmpitem, file_value))
resultarray.append(file_value)
isfile = True
except KeyError as e:
print("SCHEMA ERROR IN FILE HANDLING: %s" % e)
@@ -1143,7 +1179,7 @@ class AppBase:
try:
if parameter["schema"]["type"] == "file" and len(value) > 0:
print("\n SHOULD HANDLE FILE. Get based on value %s. <--- is this a valid ID?" % parameter["value"])
file_value = self.get_file(value)
file_value = await self.get_file(value)
print("FILE VALUE: %s \n" % file_value)
params[parameter["name"]] = file_value
@@ -1282,7 +1318,8 @@ class AppBase:
print("KeyError: %s" % e)
baseparams[key] = "KeyError: %s" % e
print("Running with params %s (1)" % baseparams)
print("Running with params (1): %s" % baseparams)
ret = await func(**baseparams)
print("Return from execution: %s" % ret)
if isinstance(ret, dict) or isinstance(ret, list):
backend/app_sdk/build.sh
+1 -1
View File
@@ -1,6 +1,6 @@
#!/bin/bash
NAME=shuffle-app_sdk
VERSION=0.8.11
VERSION=0.8.2
docker rmi docker.pkg.github.com/frikky/shuffle/$NAME:$VERSION --force
docker build . -t frikky/shuffle:app_sdk -t frikky/$NAME:$VERSION -t docker.pkg.github.com/frikky/shuffle/$NAME:$VERSION -t ghcr.io/frikky/$NAME:$VERSION
backend/go-app/walkoff.go
+10 -6
View File
@@ -954,12 +954,11 @@ func handleWorkflowQueue(resp http.ResponseWriter, request *http.Request) {
return
}
runWorkflowExecutionTransaction(0, workflowExecution.ExecutionId, actionResult, resp)
runWorkflowExecutionTransaction(ctx, 0, workflowExecution.ExecutionId, actionResult, resp)
}
// Will make sure transactions are always ran for an execution. This is recursive if it fails. Allowed to fail up to 5 times
func runWorkflowExecutionTransaction(attempts int64, workflowExecutionId string, actionResult ActionResult, resp http.ResponseWriter) {
ctx := context.Background()
func runWorkflowExecutionTransaction(ctx context.Context, attempts int64, workflowExecutionId string, actionResult ActionResult, resp http.ResponseWriter) {
// Should start a tx for the execution here
tx, err := dbclient.NewTransaction(ctx)
if err != nil {
@@ -1266,16 +1265,21 @@ func runWorkflowExecutionTransaction(attempts int64, workflowExecutionId string,
}
if _, err = tx.Commit(); err != nil {
log.Printf("[ERROR] tx.Commit: %v", err)
if attempts >= 5 {
log.Printf("[ERROR] QUITTING: tx.Commit %d: %v", attempts, err)
tx.Rollback()
workflowExecution.Status = "ABORTED"
setWorkflowExecution(ctx, *workflowExecution)
if attempts >= 0 {
resp.WriteHeader(401)
resp.Write([]byte(`{"success": false}`))
return
}
log.Printf("[WARNING] tx.Commit %d: %v", attempts, err)
attempts += 1
runWorkflowExecutionTransaction(attempts, workflowExecutionId, actionResult, resp)
runWorkflowExecutionTransaction(ctx, attempts, workflowExecutionId, actionResult, resp)
return
}
frontend/src/views/AngularWorkflow.jsx
+1 -1
View File
@@ -5902,7 +5902,7 @@ const AngularWorkflow = (props) => {
try {
const tmp = String(JSON.parse(showResult))
if (!showResult.includes("{") && !showResult.includes("[")) {
console.log("IN HERE: ", tmp)
//console.log("IN HERE: ", tmp)
jsonvalid = false
}
} catch (e) {