@app.post("/api/triggerloader/{loader_name}")
async def trigger_Loader(loader_name,loader_data:Loader_Data):
url = "https://{host}:443/restpp/ddl/testing_Loader?tag={loader_name}&filename=logic_file"
access_token = token
headers = {"Authorization": f"Bearer {access_token}", "Content-Type": "application/json"}
# Convert the loader data in the request body to the desired format
json_data = loader_data.loader
formatted_data = []
for i in range(len(json_data['E_ID'])):
row = {
"E_ID": json_data['E_ID'][i],
"E_Name": json_data['E_Name'][i],
"C_ID": json_data['C_ID'][i],
"C_Nam": json_data['C_Nam'][i]
}
formatted_data.append(row)
# Create a temporary file and write the formatted data to it
with tempfile.NamedTemporaryFile(mode='w', delete=False,dir='.') as temp_file:
for row in formatted_data:
temp_file.write(json.dumps(row) + '\n')
temp_file.flush()
# Set the filename to the temporary file
filename = temp_file.name
# Set the JSON data to be sent in the request body
data = {"filename": filename}
# Convert the data to JSON format
json_data = json.dumps(data)
print(json_data)
response = requests.post(url, headers=headers, data=json_data)
if response.status_code == 200:
print("Loader job triggered successfully.")
else:
print(f"Error triggering loader job. Response code: {response.status_code}")
print(response.json())
# Remove the temporary file
# os.remove(filename)
My loader is not not triggering the loader job i created even though there is no error in code.