Forum Discussion
Using REST APIs to create Items in Fabric data factory pipeline
Hello Everyone, I am currently working on automating the process of building a Data Factory pipeline object by utilizing an image and OpenAI. The goal is to extract details from the image and progra...
Thank you for your reply.
For example, the code that I am using is:
activities_list = [ { "name": "Lookup1", "type": "Lookup" }, { "name": "IfCondition2", "type": "IfCondition", "dependsOn": [ { "activity": "Lookup1", "dependencyConditions": [ "Succeeded" ] } ], "typeProperties": { "ifFalseActivities": [], "ifTrueActivities": [ { "name": "Copy1ifT", "type": "Copy", "dependsOn": [], "typeProperties": { "source": { "datasetSettings": {} }, "sink": { "datasetSettings": {} } } } ] } }, { "name": "SqlServerStoredProcedure3", "type": "SqlServerStoredProcedure", "dependsOn": [ { "activity": "IfCondition2", "dependencyConditions": [ "Succeeded" ] } ] } ]
# Now that the activities_list is created, assign it to the activities tag in properties
data['properties'] = { "activities": activities_list }
# Convert data from dict to string, then Byte Literal, before doing a Base-64 encoding
data_str = str(data).replace("'", '"')
createPipeline_json = data_str.encode(encoding="utf-8")
createPipeline_Json64 = base64.b64encode(createPipeline_json)
# Create a new data pipeline in Fabric
timestr = time.strftime("%Y%m%d-%H%M%S")
pipelineName = f"Pipeline from image with AI1-{timestr}"
payload = {
"displayName": pipelineName,
"type": "DataPipeline",
"definition": {
"parts": [
{
"path": "pipeline-content.json",
"payload": createPipeline_Json64,
"payloadType": "InlineBase64"
}
]
}
}
print(f"Creating pipeline: {pipelineName}")
# Call the Fabric REST API to generate the pipeline
client = fabric.FabricRestClient()
workspaceId = fabric.get_workspace_id()
try:
response = client.post(f"/v1/workspaces/{workspaceId}/items", json=payload)
if response.status_code != 201:
raise FabricHTTPException(response)
except WorkspaceNotFoundException as e:
print("Workspace is not available or cannot be found.")
except FabricHTTPException as e:
print(e)
print("Fabric HTTP Exception. Check that you have the correct Fabric API endpoints.")
response = client.get(f"/v1/workspaces/{workspaceId}/Datapipelines")
df_items = pd.json_normalize(response.json()['value'])
print("List of pipelines in the workspace:")
print(df_items)
Using the above piece of code is giving me the error which I mentioned in my original question. The same error is observed when I use the IF condition activity, for each activity or the web activity.
But if I use the same activities_list (reformatted as a proper JSON object) directly in the JSON code of the pipeline (in VIEW>EDIT JSON CODE, and paste the activities list there), then the required pipeline is produced, which means that the activities_list is correct.
Any help in helping me discern what's going wrong here will be highly appreciated.