Forum Discussion

jarvis2619's avatar
jarvis2619
Regular Visitor
1 year ago

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 programmatically construct a Data Factory pipeline object using Fabric's REST API.

The process works seamlessly when the pipeline contains relatively simple activities like Lookup, Copy, or Outlook Email. The API successfully processes the request, and the corresponding pipeline object is created in the workspace.

 

However, the issue arises when the pipeline includes more complex activities, such as IfCondition, ForEach, or WebActivity. In these cases, the REST API rejects the request and returns the following error message:

{ "requestId": "<someid>", "errorCode": "RequestValidationFailed", "message": "'Value' cannot be null." }

 

{
'Cache-Control': 'no-store, must-revalidate, no-cache',
'Pragma': 'no-cache',
'Transfer-Encoding': 'chunked',
'Content-Type': 'application/json; charset=utf-8',
'x-ms-public-api-error-code': 'RequestValidationFailed',
'Strict-Transport-Security': 'max-age=31536000; includeSubDomains',
'X-Frame-Options': 'deny',
'X-Content-Type-Options': 'nosniff',
'RequestId': '',
'Access-Control-Expose-Headers': 'RequestId',
'request-redirected': 'true',
'home-cluster-uri': '/',
'Date': 'Sat, 16 Nov 2024 07:52:45 GMT'
}

 

Can anyone advice how I should proceed?

General Question
Need Help
Pipeline

3 Replies

  • Expiscornovus's avatar
    Expiscornovus
    Most Valuable Professional
    1 year ago

    Hi jarvis2619,

     

    Are you able to share a bit more details about your setup. This might help us reproduce your issue and find a root cause.

     

    Can you for example share a screenshot of the payload you are using in the example where you are using the complex activities?

  • jarvis2619's avatar
    jarvis2619
    Regular Visitor
    1 year ago

    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.