\n","Connections changes should be expressed as an array of tuples [{from_1:to_1},{from_N:to_N}]"],"metadata":{"nteract":{"transient":{"deleting":false}},"microsoft":{"language":"python","language_group":"synapse_pyspark"}},"id":"4ae65012-350c-40c0-a68a-4069c567a85f"},{"cell_type":"code","source":["from typing import Optional\n","from sempy_labs._helper_functions import (\n"," resolve_workspace_name_and_id,\n"," lro,\n"," _decode_b64,\n",")\n","import sempy_labs._icons as icons\n","\n","import base64\n","from typing import Optional, Tuple, List\n","from uuid import UUID\n","\n","\n","def update_data_pipeline_definition(\n"," name: str, pipeline_content: dict, workspace: Optional[str] = None\n","):\n"," \"\"\"\n"," Updates an existing data pipeline with a new definition.\n","\n"," Parameters\n"," ----------\n"," name : str\n"," The name of the data pipeline.\n"," pipeline_content : dict\n"," The data pipeline content (not in Base64 format).\n"," workspace : str, default=None\n"," The name of the workspace.\n"," Defaults to None which resolves to the workspace of the attached lakehouse\n"," or if no lakehouse attached, resolves to the workspace of the notebook.\n"," \"\"\"\n","\n"," (workspace, workspace_id) = resolve_workspace_name_and_id(workspace)\n"," client = fabric.FabricRestClient()\n"," pipeline_payload = base64.b64encode(json.dumps(pipeline_content).encode('utf-8')).decode('utf-8')\n"," pipeline_id = fabric.resolve_item_id(\n"," item_name=name, type=\"DataPipeline\", workspace=workspace\n"," )\n","\n"," request_body = {\n"," \"definition\": {\n"," \"parts\": [\n"," {\n"," \"path\": \"pipeline-content.json\",\n"," \"payload\": pipeline_payload,\n"," \"payloadType\": \"InlineBase64\"\n"," }\n"," ]\n"," }\n"," }\n","\n","\n"," response = client.post(\n"," f\"v1/workspaces/{workspace_id}/items/{pipeline_id}/updateDefinition\",\n"," json=request_body,\n"," )\n","\n"," lro(client, response, return_status_code=True)\n","\n"," print(\n"," f\"{icons.green_dot} The '{name}' pipeline was updated within the '{workspace}' workspace.\"\n"," )\n","\n","def _is_valid_uuid(\n"," guid: str,\n","):\n"," \"\"\"\n"," Validates if a string is a valid GUID in version 4\n","\n"," Parameters\n"," ----------\n"," guid : str\n"," GUID to be validated.\n","\n"," Returns\n"," -------\n"," bool\n"," Boolean that indicates if the string is a GUID or not.\n"," \"\"\"\n","\n"," try:\n"," UUID(str(guid), version=4)\n"," return True\n"," except ValueError:\n"," return False"],"outputs":[],"execution_count":null,"metadata":{"jupyter":{"source_hidden":false,"outputs_hidden":false},"nteract":{"transient":{"deleting":false}},"microsoft":{"language":"python","language_group":"synapse_pyspark"}},"id":"bdd46d4a-ef58-4f9a-b2e8-a428361a17c1"},{"cell_type":"code","source":["import json\n","from jsonpath_ng import jsonpath, parse\n","from typing import Optional, Tuple, List\n","from uuid import UUID\n","\n","source_ws = ''\n","target_ws = ''\n","\n","\n","# Swaps the connection properties of an activity belonging to the specified item type(s)\n","def swap_pipeline_connection(pl_json: dict, p_source_ws: str,p_target_ws: str, \n"," p_item_type: List =['DataWarehouse','Lakehouse','Notebook'], \n"," p_conn_id_from_to: Optional[List[Tuple[str,str]]]=[]):\n"," \n"," source_ws_id = fabric.resolve_workspace_id(source_ws)\n","\n"," target_ws_id = fabric.resolve_workspace_id(target_ws)\n","\n"," if 'Warehouse' in p_item_type or 'Lakehouse' in p_item_type:\n"," ls_expr = parse('$..linkedService')\n"," for endpoint_match in ls_expr.find(pl_json):\n"," if endpoint_match.value['properties']['type'] == 'DataWarehouse' \\\n"," and endpoint_match.value['properties']['typeProperties']['workspaceId'] == source_ws_id \\\n"," and 'Warehouse' in p_item_type:\n"," # only update the warehouse if it was located in the source workspace i.e. we will update the properties to the target workspace if the warehouse resided in the same workspace as the pipeline\n"," #print(endpoint_match.value)\n"," warehouse_id = endpoint_match.value['properties']['typeProperties']['artifactId']\n"," #print(warehouse_id)\n"," warehouse_endpoint = endpoint_match.value['properties']['typeProperties']['endpoint']\n"," #print(warehouse_endpoint)\n"," \n"," source_wh_name = fabric.resolve_item_name(item_id = warehouse_id,workspace=source_ws_id)\n"," #print(remote_wh_name)\n"," # find the warehouse id of the warehouse with the same name in the target workspace\n"," target_wh_id = fabric.resolve_item_id(item_name = source_wh_name,type='Warehouse',workspace=target_ws_id)\n"," # look up the connection string for the warehouse in the target workspace\n"," whurl = f\"v1/workspaces/{target_ws_id}/warehouses/{target_wh_id}\"\n"," whresponse = client.get(whurl)\n"," lhconnStr = whresponse.json()['properties']['connectionString']\n"," endpoint_match.value['properties']['typeProperties']['artifactId'] = target_wh_id\n"," endpoint_match.value['properties']['typeProperties']['workspaceId'] = target_ws_id\n"," endpoint_match.value['properties']['typeProperties']['endpoint'] = lhconnStr\n"," #print(endpoint_match.value)\n"," ls_expr.update(endpoint_match,endpoint_match.value)\n"," if endpoint_match.value['properties']['type'] == 'Lakehouse' \\\n"," and endpoint_match.value['properties']['typeProperties']['workspaceId'] == source_ws_id \\\n"," and 'Lakehouse' in p_item_type:\n"," #print(endpoint_match.value)\n"," lakehouse_id = endpoint_match.value['properties']['typeProperties']['artifactId']\n"," remote_lh_name = fabric.resolve_item_name(item_id = lakehouse_id,workspace=source_ws_id)\n"," # find the lakehouse id of the lakehouse with the same name in the target workspace\n"," target_lh_id = fabric.resolve_item_id(item_name = remote_lh_name,type='Lakehouse',workspace=target_ws_id)\n"," endpoint_match.value['properties']['typeProperties']['artifactId'] = target_lh_id\n"," endpoint_match.value['properties']['typeProperties']['workspaceId'] = target_ws_id\n"," ls_expr.update(endpoint_match,endpoint_match.value)\n"," # print(endpoint_match.value)\n","\n","\n"," if 'Notebook' in p_item_type: \n"," ls_expr = parse('$..activities')\n","\n"," for endpoint_match in ls_expr.find(pl_json):\n"," for activity in endpoint_match.value:\n"," #print(activity['type'])\n"," if activity['type']=='TridentNotebook' and 'Notebook' in p_item_type: #only update if the notebook was in the same workspace as the pipeline\n"," print('change from '+activity['typeProperties']['workspaceId'])\n"," source_nb_id = activity['typeProperties']['notebookId']\n"," source_nb_name = fabric.resolve_item_name(item_id = source_nb_id,workspace=source_ws_id)\n"," target_nb_id = fabric.resolve_item_id(item_name = source_nb_name,type='Notebook',workspace=target_ws_id)\n"," activity['typeProperties']['notebookId']=target_nb_id\n"," activity['typeProperties']['workspaceId']=target_ws_id\n"," print('to notebook '+ target_nb_id)\n"," #ls_expr.update(endpoint_match,endpoint_match.value)\n","\n"," if p_conn_from_to:\n"," for ti_conn_from_to in p_conn_from_to:\n"," if not _is_valid_uuid(ti_conn_from_to[0]):\n"," print('Connection from is string '+ str(ti_conn_from_to[0]))\n"," dfC_filt = df_conns[df_conns[\"Connection Name\"] == ti_conn_from_to[0]] \n"," connId_from = dfC_filt['Connection Id'].iloc[0] \n"," else:\n"," connId_from = ti_conn_from_to[0]\n","\n"," if not _is_valid_uuid(ti_conn_from_to[1]):\n"," print('Connection from is string '+ str(ti_conn_from_to[1]))\n"," dfC_filt = df_conns[df_conns[\"Connection Name\"] == ti_conn_from_to[1]] \n"," connId_to = dfC_filt['Connection Id'].iloc[0] \n"," else:\n"," connId_to = ti_conn_from_to[1]\n","\n"," ls_expr = parse('$..externalReferences')\n"," for externalRef in ls_expr.find(pl_json):\n"," if externalRef.value['connection']==connId_from:\n"," print('Changing connection from '+str(connId_from))\n"," externalRef.value['connection']=connId_to\n"," ls_expr.update(externalRef,externalRef.value)\n"," print('to '+str(connId_to))\n","\n"," return pl_json\n","\n","\n","\n","# loading a dataframe of connections to perform an ID lookup if required \n","df_conns = labs.list_connections()\n","\n","df_pipeline = labs.list_data_pipelines(target_ws)\n","for index, row in df_pipeline.iterrows():\n"," #print(labs.get_data_pipeline_definition(row['Data Pipeline Name'],target_ws))\n"," if row['Data Pipeline Name']=='plRecover_WH6_Prod2_Warehouse2_fixed':\n"," pipeline_json = json.loads(labs.get_data_pipeline_definition(row['Data Pipeline Name'],source_ws))\n","\n"," p_new_json = swap_pipeline_connection(pipeline_json, source_ws,target_ws,\n"," ['DataWarehouse','Lakehouse','Notebook'],\n"," [p_connections_from_to]) \n"," #print(json.dumps(pipeline_json, indent=4))\n"," \n"," update_data_pipeline_definition(name=row['Data Pipeline Name'],pipeline_content=pipeline_json, workspace=target_ws)\n"],"outputs":[],"execution_count":null,"metadata":{"jupyter":{"source_hidden":false,"outputs_hidden":false},"nteract":{"transient":{"deleting":false}},"microsoft":{"language":"python","language_group":"synapse_pyspark"}},"id":"079958e8-2880-484a-a994-41caf47e747e"},{"cell_type":"markdown","source":["##### Commit changes made above to Git"],"metadata":{"nteract":{"transient":{"deleting":false}},"microsoft":{"language":"python","language_group":"synapse_pyspark"}},"id":"44174276-b983-4e80-9451-0afb9589cf1f"},{"cell_type":"code","source":["labs.commit_to_git(comment='Initial', workspace=target_ws)"],"outputs":[],"execution_count":null,"metadata":{"jupyter":{"source_hidden":false,"outputs_hidden":false},"nteract":{"transient":{"deleting":false}},"microsoft":{"language":"python","language_group":"synapse_pyspark"}},"id":"9a5c3d84-f71d-4348-b419-c4953ac9e1d0"}],"metadata":{"kernel_info":{"name":"synapse_pyspark"},"kernelspec":{"name":"synapse_pyspark","language":"Python","display_name":"Synapse PySpark"},"language_info":{"name":"python"},"microsoft":{"language":"python","language_group":"synapse_pyspark","ms_spell_check":{"ms_spell_check_language":"en"}},"widgets":{},"nteract":{"version":"nteract-front-end@1.0.0"},"synapse_widget":{"version":"0.1","state":{}},"spark_compute":{"compute_id":"/trident/default","session_options":{"conf":{"spark.synapse.nbs.session.timeout":"1200000"}}},"dependencies":{"lakehouse":{}}},"nbformat":4,"nbformat_minor":5} \ No newline at end of file +{"cells":[{"cell_type":"markdown","source":["##### Branch out to new workspace notebook - post activity\n","\n","After cloning a workspace, this notebook will reconfigure any references to the old workspace by rebinding them to the new workspace. \n","\n","For example a pipeline referencing a warehouse or a default lakehouse of a notebook.\n","\n","This notebook runs post activity tasks can be run after [branch out to new workspace functionality](https://blog.fabric.microsoft.com/en-us/blog/introducing-new-branching-capabilities-in-fabric-git-integration) or the [custom AzDO script](https://github.com/microsoft/fabric-toolbox/blob/main/accelerators/CICD/Branch-out-to-new-workspace/AzDO/scripts/BranchOut-Feature-Workspace-Automation.py).\n","\n","Summary of post activities in order:\n","
- \n","
- Default lakehouses and warehouse are updated to local lakehouse/warehouses \n","
- Either creates shortcuts in local lakehouse back to tables in the source lakehouse, or copies the data from source lakehouse. Set via parameter below. \n","
- Copy warehouse data. Set via parameter below \n","
- Changes directlake semantic model connections for semantic models to \"local\" lakehouse/warehouse \n","
- Rebinds reports to \"local\" semantic models \n","
- Changes pipeline lakehouse/warehouse references to local item \n","
- Ability to swap connections in pipelines from old to new \n","
- Commit changes to git \n","
- \n","
- Requires Semantic Link Labs installed by pip install below or added to environment library. \n","
- Requires JmesPath library for data pipeline JSON manipulation i.e. connection swaps. \n","
- \n","
- Does not recreate item shares or external shortcuts \n","
- Does not re-apply lakehouse SQL Endpoint or Warehouse object/row/column level security \n","
- Does not recreate data access roles in Lakehouse \n","
- Untested with Lakehouses where with schema support enabled \n","
In this case the lakehouses will be pre-created. Shortcuts created or data copied done in subsequent cell"],"metadata":{"nteract":{"transient":{"deleting":false}},"microsoft":{"language":"python","language_group":"synapse_pyspark"}},"id":"2d515be2-cc9f-4a09-9c10-f2ad12485c23"},{"cell_type":"code","source":["if has_wh_views_on_lh:\n"," lhs = fabric.list_items(type='Lakehouse',workspace=source_ws)\n"," for index,row in fabric.list_items(type='Lakehouse',workspace=source_ws).iterrows():\n"," #try:\n"," print(f\"Creating {row['Display Name']} lakehouse...\",end=\"\")\n"," fabric.create_lakehouse(display_name = row['Display Name'],description='Created programatically via branch out script',max_attempts=3, workspace=target_ws)\n"," print('done')"],"outputs":[],"execution_count":null,"metadata":{"microsoft":{"language":"python","language_group":"synapse_pyspark"}},"id":"53dbce3a-a318-400c-acca-e3a8e9fd19ae"},{"cell_type":"markdown","source":["##### Either create shortcuts from source to target lakehouse(s) or copy data\n","\n","Loops through lakehouse(s) in the target workspace and either populates them with shortcuts or data\n","\n"],"metadata":{"nteract":{"transient":{"deleting":false}},"microsoft":{"language":"python","language_group":"synapse_pyspark"}},"id":"a15065f3-670d-4bc9-b337-51709f6cdb1f"},{"cell_type":"code","source":["# ensure target workspace ID is set based on name\n","if target_ws_id == '':\n"," target_ws_id = fabric.resolve_workspace_id(target_ws)\n","\n","# populate the lakehouses with shortcuts or data\n","create_shortcuts_or_copy_data(copy_lakehouse_data,create_lakehouse_shortcuts)\n","time.sleep(120) # wait for sql endpoint to update\n"],"outputs":[],"execution_count":null,"metadata":{"jupyter":{"source_hidden":false,"outputs_hidden":false},"nteract":{"transient":{"deleting":false}},"microsoft":{"language":"python","language_group":"synapse_pyspark"},"collapsed":false},"id":"8bec3fbc-4a75-4ba3-86d5-0620ec504a8f"},{"cell_type":"markdown","source":["##### If not done in ADO, connect workspace to Git, initialize and update\n","This cell will only run if there are warehouse views which depend on lakehouse tables based on parameter has_wh_views_on_lh"],"metadata":{"nteract":{"transient":{"deleting":false}},"microsoft":{"language":"python","language_group":"synapse_pyspark"}},"id":"c7dd8a46-f6bd-4a84-90f0-e176357ced17"},{"cell_type":"code","source":["if has_wh_views_on_lh:\n","\n"," # branch name should be the same name as the target workspace, if not modify parameters to pass in the branch name from ADO\n"," branch_name = target_ws \n"," # fetch existing Git connection details from source workspace\n"," gitconnx = labs.get_git_connection(source_ws)\n"," # connect target workspace to the same git provider and new branch\n"," if gitconnx.loc[0, 'Git Provider Type'] == 'AzureDevOps':\n"," result = labs.connect_workspace_to_azure_dev_ops(\n"," gitconnx.loc[0, 'Organization Name'],\n"," gitconnx.loc[0, 'Project Name'],\n"," gitconnx.loc[0, 'Repository Name'],\n"," branch_name,\n"," gitconnx.loc[0, 'Directory Name'],\n"," target_ws)\n"," elif gitconnx.loc[0, 'Git Provider Type'] == 'GitHub': #### TODO ####\n"," #TODO connect workspace with Github details\n"," labs.connect_workspace_to_github(\n"," '', #owner_name: str,\n"," gitconnx.loc[0, 'Repository Name'], #repository_name: str,\n"," branch_name,\n"," gitconnx.loc[0, 'Directory Name'], #directory_name: str,\n"," '', #connection_id: UUID,\n"," '', #source: str = \"ConfiguredConnection\",\n"," target_ws)\n"," else:\n"," raise ValueError(f\"Unsupported Git provider type {gitconnx.loc[0, 'Git Provider Type']}.\")"],"outputs":[],"execution_count":null,"metadata":{"microsoft":{"language":"python","language_group":"synapse_pyspark"},"collapsed":false},"id":"93fe3d56-d870-433f-bdea-1e50ca58440b"},{"cell_type":"code","source":["if has_wh_views_on_lh:\n"," # inialize the git connection\n"," commit_hash = initialize_git_connection(target_ws,'PreferRemote')\n","\n","\n"," # update from git \n"," labs.update_from_git(remote_commit_hash=commit_hash,\n"," conflict_resolution_policy='PreferRemote', \n"," allow_override=True,\n"," workspace=target_ws)\n","\n"],"outputs":[],"execution_count":null,"metadata":{"microsoft":{"language":"python","language_group":"synapse_pyspark"}},"id":"b0061d77-6213-4eeb-a55f-5a870077628c"},{"cell_type":"code","source":["labs.commit_to_git(comment='Initial', workspace=target_ws)"],"outputs":[],"execution_count":null,"metadata":{"microsoft":{"language":"python","language_group":"synapse_pyspark"}},"id":"89d2fa99-1073-44a9-94cc-74e74abf68b4"},{"cell_type":"markdown","source":["##### Update default and attached lakehouses/warehouses for notebooks\n","\n","Update notebook dependencies based on but now supports T-SQL notebooks:\n","https://github.com/PowerBiDevCamp/FabConWorkshopSweden/blob/main/DemoFiles/GitUpdateWorkspace/updateWorkspaceDependencies_v1.ipynb\n"],"metadata":{"nteract":{"transient":{"deleting":false}},"microsoft":{"language":"python","language_group":"synapse_pyspark"}},"id":"aaae8a08-588d-4dd8-9d2c-2200b7a88d30"},{"cell_type":"code","source":["for notebook in notebookutils.notebook.list(workspaceId=target_ws_id):\n"," updates = False\n"," if notebook.displayName == 'ETL':#True: #notebook.displayName == 'T-SQL_Notebook': #notebook.displayName != 'Create Feature Branch':\n","\n"," # Get the current notebook definition\n"," json_payload = json.loads(notebookutils.notebook.getDefinition(notebook.displayName,workspaceId=source_ws_id))\n"," #print(json.dumps(json_payload, indent=4))\n"," # Check for any attached lakehouses\n"," if 'dependencies' in json_payload['metadata'] \\\n"," and 'lakehouse' in json_payload['metadata']['dependencies'] \\\n"," and json_payload['metadata'][\"dependencies\"][\"lakehouse\"] is not None:\n"," # Extract attached and default lakehouses\n"," current_lakehouse = json_payload['metadata']['dependencies']['lakehouse']\n"," # if default lakehouse setting exists\n"," if 'default_lakehouse_name' in current_lakehouse:\n"," print(f\"Updating notebook {notebook.displayName} with new default lakehouse: {current_lakehouse['default_lakehouse_name']} in workspace {target_ws}\")\n"," source_lh_name = fabric.resolve_item_name(item_id = current_lakehouse['default_lakehouse'],type='Lakehouse',workspace=source_ws_id)\n"," current_lakehouse['default_lakehouse'] = fabric.resolve_item_id(item_name = source_lh_name,type='Lakehouse',workspace=target_ws_id)\n"," current_lakehouse['default_lakehouse_workspace_id'] = target_ws_id\n"," updates = True\n"," # loop through all attached lakehouess\n"," for lakehouse in json_payload['metadata']['dependencies']['lakehouse']['known_lakehouses']:\n"," source_lh_id = lakehouse['id']\n"," # find source lakehouse name\n"," source_lh_name = fabric.resolve_item_name(item_id = lakehouse['id'],type='Lakehouse',workspace=source_ws_id)\n"," # find target lakehouse id based on name\n"," target_lh_id = fabric.resolve_item_id(item_name = source_lh_name,type='Lakehouse',workspace=target_ws_id)\n"," lakehouse['id'] = target_lh_id\n"," print(f'Updating attached lakehouse {source_lh_name} from {source_lh_id} to target ID {target_lh_id}')\n"," updates = True\n","\n"," if 'dependencies' in json_payload['metadata'] and 'warehouse' in json_payload['metadata']['dependencies']:\n"," # Fetch existing details\n"," current_warehouse = json_payload['metadata']['dependencies']['warehouse']\n"," current_warehouse_id = current_warehouse['default_warehouse']\n"," source_wh_name = fabric.resolve_item_name(item_id = current_warehouse_id,workspace=source_ws_id)\n"," #print('Source warehouse name is ' + source_wh_name)\n"," target_wh_id = fabric.resolve_item_id(item_name = source_wh_name,type='Warehouse',workspace=target_ws_id)\n","\n"," if 'default_warehouse' in current_warehouse:\n"," #json_payload['metadata']['dependencies']['warehouse'] = {}\n"," print(f\"Attempting to update notebook {notebook.displayName} with new default warehouse: {target_wh_id} in {target_ws}\")\n"," \n"," json_payload['metadata']['dependencies']['warehouse']['default_warehouse'] = target_wh_id\n"," for warehouse in json_payload['metadata']['dependencies']['warehouse']['known_warehouses']:\n"," if warehouse['id'] == current_warehouse_id:\n"," warehouse['id'] = target_wh_id\n"," updates = True\n","\n"," if updates:\n"," notebookutils.notebook.updateDefinition(\n"," name = notebook.displayName,\n"," content = json.dumps(json_payload),\n"," workspaceId = target_ws_id\n"," )\n"," \n"," print(f\"Updated notebook {notebook.displayName} in {target_ws}\")\n","\n"," else:\n"," print(f'No default lakehouse set for notebook {notebook.displayName}, ignoring.')"],"outputs":[],"execution_count":null,"metadata":{"jupyter":{"source_hidden":false,"outputs_hidden":false},"nteract":{"transient":{"deleting":false}},"microsoft":{"language":"python","language_group":"synapse_pyspark"}},"id":"5c60b5d2-f83c-46f8-9870-9fd609166b67"},{"cell_type":"markdown","source":["##### Copy warehouse data via parameterised pipeline\n","\n","Loop through all warehouses and copy the data"],"metadata":{"nteract":{"transient":{"deleting":false}},"microsoft":{"language":"python","language_group":"synapse_pyspark"}},"id":"f198b816-77e9-4f04-9139-d78237bedc72"},{"cell_type":"code","source":["if copy_warehouse_data:\n"," p_logging_verbose = False\n"," df_warehouses = (labs.list_warehouses(target_ws))\n"," #display(df_warehouses)\n"," for index, row in df_warehouses.iterrows():\n"," source_wh_id = labs.resolve_warehouse_id(row['Warehouse Name'],source_ws_id)\n"," target_wh_id = labs.resolve_warehouse_id(row['Warehouse Name'],target_ws_id)\n"," \n"," src_path = f'abfss://'+source_ws_id+'@onelake.dfs.fabric.microsoft.com/'+source_wh_id\n"," tgt_path = f'abfss://'+target_ws_id+'@onelake.dfs.fabric.microsoft.com/'+target_wh_id\n","\n"," # extract the list of schemas per data \n"," schema_list = get_lh_object_list(src_path,['Tables'])\n"," # extract a list of warehouse objects per schema and store in a list\n"," table_list = get_wh_object_list(schema_list['name'],src_path)\n"," \n"," # create a temporary staging lakehouse per warehouse to create shortcuts into, \n"," # which point back to original warehouse data currently in the DR storage account\n"," lhname = 'temp_rlh_' + source_ws+'_'+row['Warehouse Name']\n"," # check if it exists before attempting create\n"," if p_logging_verbose:\n"," print('Checking whether the temporary lakehouse \"'+ lhname +'\" exists in workspace '+target_ws+'...')\n"," temp_lh_id = getItemId(target_ws_id,lhname,'Lakehouse')\n"," if temp_lh_id == 'NotExists':\n"," lhname = lhname[:256] # lakehouse name should not exceed 256 characters\n"," payload = payload = '{\"displayName\": \"' + lhname + '\",' \\\n"," + '\"description\": \"Interim staging lakehouse for primary warehouse recovery: ' \\\n"," + source_ws+'_'+row['Warehouse Name'] + 'into workspace '+ target_ws + '(' + target_ws +')\"}'\n"," try:\n"," lhurl = \"v1/workspaces/\" + target_ws_id + \"/lakehouses\"\n"," lhresponse = client.post(lhurl,json= json.loads(payload))\n"," temp_lh_id = lhresponse.json()['id']\n"," if p_logging_verbose:\n"," print('Temporary lakehouse \"'+ lhname +'\" created with Id ' + temp_lh_id + ': ' + str(lhresponse.status_code) + ' ' + str(lhresponse.text))\n"," except Exception as error:\n"," print(error.errorCode)\n"," else:\n"," if p_logging_verbose:\n"," print('Temporary lakehouse '+lhname+' (' + temp_lh_id + ') already exists.')\n"," \n"," time.sleep(60) # waiting for temporary lakehouse to provision completely \n","\n"," # Create shortcuts for every table in the format of schema_table under the tables folder\n"," for index,itable in table_list.iterrows():\n"," shortcutExists=False\n"," # Check if shortcut exists\n"," try:\n"," url = \"v1/workspaces/\" + target_ws_id + \"/items/\" + temp_lh_id + \"/shortcuts/Tables/\"+itable['schema']+'_'+itable['name']\n"," tlhresponse = client.get(url)\n"," shortcutExists = True\n"," if p_logging_verbose:\n"," print('Shortcut '+itable['schema']+'_'+itable['name'] +' already exists')\n"," except Exception as error:\n"," shortcutExists = False \n","\n"," if not shortcutExists: \n"," # Create shortcuts - one per table per schema\n"," url = \"v1/workspaces/\" + target_ws_id + \"/items/\" + temp_lh_id + \"/shortcuts\"\n"," scpayload = '{' \\\n"," '\"path\": \"Tables/\",' \\\n"," '\"name\": \"'+itable['schema']+'_'+itable['name']+'\",' \\\n"," '\"target\": {' \\\n"," '\"oneLake\": {' \\\n"," '\"workspaceId\": \"' + source_ws_id + '\",' \\\n"," '\"itemId\": \"'+ source_wh_id +'\",' \\\n"," '\"path\": \"/Tables/' + itable['schema']+'/'+itable['name'] + '\"' \\\n"," '}}}' \n"," try:\n"," #print(scpayload) \n"," shctresponse = client.post(url,json= json.loads(scpayload))\n"," if p_logging_verbose:\n"," print('Shortcut '+itable['schema']+'_'+itable['name'] + ' created.' )\n","\n"," except Exception as error:\n"," print('Error creating shortcut '+itable['schema']+'_'+itable['name']+' due to '+str(error) + ':' + shctresponse.text)\n"," \n"," recovery_pipeline_prefix= 'plRecover_WH' \n"," # recovery pipeline name should not exceed 256 characters\n"," recovery_pipeline = recovery_pipeline_prefix+'_'+source_ws + '_'+row['Warehouse Name'][:256]\n"," if p_logging_verbose:\n"," print('Attempting to deploy a copy pipeline in the target workspace to load the target warehouse tables from the shortcuts created above... ')\n"," # Create the pipeline in the target workspace that loads the target warehouse from shortcuts created above \n"," plid = getItemId( target_ws_id,recovery_pipeline,'DataPipeline')\n"," #print(plid)\n"," if plid == 'NotExists':\n"," plid = createDWrecoverypl(target_ws_id,recovery_pipeline_prefix+'_'+source_ws + '_'+row['Warehouse Name'])\n"," if p_logging_verbose:\n"," print('Recovery pipeline ' + recovery_pipeline + ' created with Id '+plid)\n"," else:\n"," if p_logging_verbose:\n"," print('Datawarehouse recovery pipeline \"' + recovery_pipeline + '\" ('+plid+') already exist in workspace \"'+target_ws + '\" ('+target_ws_id+')') \n"," print('\\n')\n","\n"," tablesToCopyParam = table_list[['schema','name']].to_json( orient='records')\n"," # ensure the temporary lakehouse exists\n","\n"," # obtain the connection string for the lakehouse to pass to the copy pipeline\n"," whurl = \"v1/workspaces/\" + target_ws_id + \"/lakehouses/\" + temp_lh_id\n"," whresponse = client.get(whurl)\n"," lhconnStr = whresponse.json()['properties']['sqlEndpointProperties']['connectionString']\n","\n"," # get the SQLEndpoint ID of the lakehouse to pass to the copy pipeline\n"," items = fabric.list_items(workspace=target_ws_id)\n"," #print(items)\n"," temp_lh_sqle_id = items[(items['Type'] == 'SQLEndpoint') & (items['Display Name']==lhname)]['Id'].values[0]\n","\n","\n"," # obtain the connection string for the warehouse to pass to the copy pipeline \n"," whurl = \"v1/workspaces/\" + target_ws_id + \"/warehouses/\" + target_wh_id\n"," whresponse = client.get(whurl)\n"," whconnStr = whresponse.json()['properties']['connectionInfo']\n","\n"," # obtain the pipeline id created to recover this warehouse\n"," plid = getItemId( target_ws_id,recovery_pipeline,'DataPipeline')\n"," if plid == 'NotExists':\n"," print('Error: Could not execute pipeline '+recovery_pipeline+ ' as the ID could not be obtained ')\n"," else:\n"," # pipeline url including pipeline Id unique to each warehouse\n"," plurl = 'v1/workspaces/'+target_ws_id+'/items/'+plid+'/jobs/instances?jobType=Pipeline'\n"," #print(plurl)\n","\n"," payload_data = '{' \\\n"," '\"executionData\": {' \\\n"," '\"parameters\": {' \\\n"," '\"lakehouseId\": \"' + temp_lh_sqle_id + '\",' \\\n"," '\"tablesToCopy\": ' + tablesToCopyParam + ',' \\\n"," '\"workspaceId\": \"' + target_ws_id +'\",' \\\n"," '\"warehouseId\": \"' + target_wh_id + '\",' \\\n"," '\"lakehouseConnStr\": \"' + lhconnStr + '\",' \\\n"," '\"warehouseConnStr\": \"' + whconnStr + '\"' \\\n"," '}}}'\n"," #print(payload_data)\n"," plresponse = client.post(plurl, json=json.loads(payload_data))\n"," if p_logging_verbose:\n"," print(str(plresponse.status_code)) \n"," print('Done')\n"],"outputs":[],"execution_count":null,"metadata":{"jupyter":{"source_hidden":true,"outputs_hidden":false},"nteract":{"transient":{"deleting":false}},"microsoft":{"language":"python","language_group":"synapse_pyspark"},"collapsed":false},"id":"57dafef7-17a2-475f-9e62-eecc6660440c"},{"cell_type":"markdown","source":["##### Update directlake model lakehouse/warehouse connection\n","\n","https://semantic-link-labs.readthedocs.io/en/stable/sempy_labs.directlake.html#sempy_labs.directlake.update_direct_lake_model_connection "],"metadata":{"nteract":{"transient":{"deleting":false}},"microsoft":{"language":"python","language_group":"synapse_pyspark"}},"id":"cc97be77-116e-4cde-bdc6-2971ab98a083"},{"cell_type":"code","source":["\n","df_datasets = fabric.list_datasets(target_ws)\n","\n","# Iterate over each dataset in the dataframe\n","for index, row in df_datasets.iterrows():\n"," try:\n"," # Check if the dataset is not the default semantic model\n"," if not labs.is_default_semantic_model(row['Dataset Name'], fabric.resolve_workspace_id(target_ws)):\n"," #print('Updating semantic model connection ' + row['Dataset Name'] + ' in workspace '+ target_ws)\n"," labs.directlake.update_direct_lake_model_connection(dataset=row['Dataset Name'], \n"," workspace= target_ws,\n"," source=labs.directlake.get_direct_lake_source(row['Dataset Name'], workspace= target_ws)[1], \n"," source_type=labs.directlake.get_direct_lake_source(row['Dataset Name'], workspace= target_ws)[0], \n"," source_workspace=target_ws)\n"," labs.refresh_semantic_model(dataset=row['Dataset Name'], workspace= target_ws)\n"," except Exception as error:\n"," errmsg = f\"Failed to update and refresh semantic model {row['Dataset Name']} due to: {str(error)}\"\n"," print(errmsg)\n"," #raise ValueError(errmsg)\n"],"outputs":[],"execution_count":null,"metadata":{"jupyter":{"source_hidden":false,"outputs_hidden":false},"nteract":{"transient":{"deleting":false}},"microsoft":{"language":"python","language_group":"synapse_pyspark"}},"id":"9deccda6-5c3d-4b88-8ed8-68855ca0949a"},{"cell_type":"markdown","source":["##### Rebind reports to local datasets\n","\n","https://semantic-link-labs.readthedocs.io/en/latest/sempy_labs.report.html#sempy_labs.report.report_rebind"],"metadata":{"nteract":{"transient":{"deleting":false}},"microsoft":{"language":"python","language_group":"synapse_pyspark"}},"id":"36783f3b-4904-4d74-842d-dbd026a3184a"},{"cell_type":"code","source":["df_reports = fabric.list_reports(workspace=target_ws)\n","for index, row in df_reports.iterrows():\n"," #print(row['Name'] + '-' + row['Dataset Id'])\n"," df_datasets = fabric.list_datasets(workspace=target_ws)\n"," dataset_name = df_datasets[df_datasets['Dataset ID'] == row['Dataset Id']]['Dataset Name'].values[0]\n"," print(f'Rebinding report to {dataset_name} in {target_ws}')\n"," labs.report.report_rebind(report=row['Name'],dataset=dataset_name, report_workspace=target_ws, dataset_workspace=target_ws)\n"],"outputs":[],"execution_count":null,"metadata":{"jupyter":{"source_hidden":false,"outputs_hidden":false},"nteract":{"transient":{"deleting":false}},"microsoft":{"language":"python","language_group":"synapse_pyspark"},"collapsed":false},"id":"06268ede-b795-493e-9a8d-772654ce7e20"},{"cell_type":"markdown","source":["##### Update data pipeline source & sink connections\n","\n","Support changes lakehouses, warehouses, notebooks and connections from source to target.
\n","Connections changes should be expressed as an array of tuples [{from_1:to_1},{from_N:to_N}]"],"metadata":{"nteract":{"transient":{"deleting":false}},"microsoft":{"language":"python","language_group":"synapse_pyspark"}},"id":"4ae65012-350c-40c0-a68a-4069c567a85f"},{"cell_type":"code","source":["if len(connections_from_to)>0: \n"," # convert from a string to a proper type i.e. list of tuples \n"," #connections_from_to = ast.literal_eval(connections_from_to)\n"," # loading a dataframe of connections to perform an ID lookup if required \n"," df_conns = labs.list_connections()\n","\n"," df_pipeline = labs.list_data_pipelines(source_ws)\n"," for index, row in df_pipeline.iterrows():\n"," pipeline_json = json.loads(labs.get_data_pipeline_definition(row['Data Pipeline Name'],source_ws))\n","\n"," p_new_json = swap_pipeline_connection(pipeline_json, source_ws,target_ws,\n"," ['DataWarehouse','Lakehouse','Notebook'],\n"," [connections_from_to]) \n"," #print(json.dumps(pipeline_json, indent=4))\n"," \n"," update_data_pipeline_definition(name=row['Data Pipeline Name'],pipeline_content=pipeline_json, workspace=target_ws)\n"],"outputs":[],"execution_count":null,"metadata":{"jupyter":{"source_hidden":false,"outputs_hidden":false},"nteract":{"transient":{"deleting":false}},"microsoft":{"language":"python","language_group":"synapse_pyspark"}},"id":"079958e8-2880-484a-a994-41caf47e747e"},{"cell_type":"markdown","source":["##### Commit changes made above to Git"],"metadata":{"nteract":{"transient":{"deleting":false}},"microsoft":{"language":"python","language_group":"synapse_pyspark"}},"id":"44174276-b983-4e80-9451-0afb9589cf1f"},{"cell_type":"code","source":["labs.commit_to_git(comment='Initial', workspace=target_ws)"],"outputs":[],"execution_count":null,"metadata":{"jupyter":{"source_hidden":false,"outputs_hidden":false},"nteract":{"transient":{"deleting":false}},"microsoft":{"language":"python","language_group":"synapse_pyspark"}},"id":"9a5c3d84-f71d-4348-b419-c4953ac9e1d0"}],"metadata":{"kernel_info":{"name":"synapse_pyspark"},"kernelspec":{"name":"synapse_pyspark","language":"Python","display_name":"Synapse PySpark"},"language_info":{"name":"python"},"microsoft":{"language":"python","language_group":"synapse_pyspark","ms_spell_check":{"ms_spell_check_language":"en"}},"widgets":{"application/vnd.jupyter.widget-state+json":{"version_major":2,"version_minor":0,"state":{"0bf5b65883d845b8a785f718f257f0fe":{"model_name":"LayoutModel","model_module":"@jupyter-widgets/base","model_module_version":"2.0.0","state":{}},"34636ca1188542389c57e315b907be79":{"model_name":"LayoutModel","model_module":"@jupyter-widgets/base","model_module_version":"2.0.0","state":{}},"729255d539d44c0da98b3f6e1c79f301":{"model_name":"LabelModel","model_module":"@jupyter-widgets/controls","model_module_version":"2.0.0","state":{"value":"Initial text","layout":"IPY_MODEL_34b4c41de9ad476088e9e30f193fd2d3","style":"IPY_MODEL_57e49b7587e9459f9f53fc6e48e1ed70"}},"c70d5bf879c243a984ade0d098ab4ff5":{"model_name":"LabelStyleModel","model_module":"@jupyter-widgets/controls","model_module_version":"2.0.0","state":{"description_width":"","font_size":null,"text_color":null,"font_family":null,"font_style":null,"font_variant":null,"font_weight":null,"text_decoration":null}},"4cd8bbf291024e7bb03f1fa73d110710":{"model_name":"LayoutModel","model_module":"@jupyter-widgets/base","model_module_version":"2.0.0","state":{}},"41f8d04c976245918a4be0806750173c":{"model_name":"LayoutModel","model_module":"@jupyter-widgets/base","model_module_version":"2.0.0","state":{}},"2529422048414fcabaaa68abefbd8d68":{"model_name":"LabelStyleModel","model_module":"@jupyter-widgets/controls","model_module_version":"2.0.0","state":{"description_width":"","font_size":null,"text_color":null,"font_family":null,"font_style":null,"font_variant":null,"font_weight":null,"text_decoration":null}},"bbd0570346bf4891a6740d5c472f557d":{"model_name":"LabelModel","model_module":"@jupyter-widgets/controls","model_module_version":"2.0.0","state":{"value":"⌛ Loading","layout":"IPY_MODEL_41f8d04c976245918a4be0806750173c","style":"IPY_MODEL_d464a36f82f14b8d92cd20b8b77b2d4c"}},"1130fc45eb4f47b1bf8a05201004a829":{"model_name":"LabelModel","model_module":"@jupyter-widgets/controls","model_module_version":"2.0.0","state":{"value":"Initial text","layout":"IPY_MODEL_0bf5b65883d845b8a785f718f257f0fe","style":"IPY_MODEL_477e0a21d0ee429d847afc4f29265fa7"}},"1a9e3ec3c78245a58d7da90f00663d0f":{"model_name":"LabelModel","model_module":"@jupyter-widgets/controls","model_module_version":"2.0.0","state":{"value":"Initial text","layout":"IPY_MODEL_34636ca1188542389c57e315b907be79","style":"IPY_MODEL_d13c5bb179b04d789caf440915ad4ab4"}},"638b47eed64847fba74e81e558f85aef":{"model_name":"LabelModel","model_module":"@jupyter-widgets/controls","model_module_version":"2.0.0","state":{"value":"⌛ Loading","layout":"IPY_MODEL_4cd8bbf291024e7bb03f1fa73d110710","style":"IPY_MODEL_c70d5bf879c243a984ade0d098ab4ff5"}},"748fa550c8c34bf68caf6fb5480c2bcc":{"model_name":"LabelModel","model_module":"@jupyter-widgets/controls","model_module_version":"2.0.0","state":{"value":"⌛ Loading","layout":"IPY_MODEL_31877256e23b46fb9fb9619833d4a58a","style":"IPY_MODEL_2529422048414fcabaaa68abefbd8d68"}},"d13c5bb179b04d789caf440915ad4ab4":{"model_name":"LabelStyleModel","model_module":"@jupyter-widgets/controls","model_module_version":"2.0.0","state":{"description_width":"","font_size":null,"text_color":null,"font_family":null,"font_style":null,"font_variant":null,"font_weight":null,"text_decoration":null}},"57e49b7587e9459f9f53fc6e48e1ed70":{"model_name":"LabelStyleModel","model_module":"@jupyter-widgets/controls","model_module_version":"2.0.0","state":{"description_width":"","font_size":null,"text_color":null,"font_family":null,"font_style":null,"font_variant":null,"font_weight":null,"text_decoration":null}},"34b4c41de9ad476088e9e30f193fd2d3":{"model_name":"LayoutModel","model_module":"@jupyter-widgets/base","model_module_version":"2.0.0","state":{}},"477e0a21d0ee429d847afc4f29265fa7":{"model_name":"LabelStyleModel","model_module":"@jupyter-widgets/controls","model_module_version":"2.0.0","state":{"description_width":"","font_size":null,"text_color":null,"font_family":null,"font_style":null,"font_variant":null,"font_weight":null,"text_decoration":null}},"d464a36f82f14b8d92cd20b8b77b2d4c":{"model_name":"LabelStyleModel","model_module":"@jupyter-widgets/controls","model_module_version":"2.0.0","state":{"description_width":"","font_size":null,"text_color":null,"font_family":null,"font_style":null,"font_variant":null,"font_weight":null,"text_decoration":null}},"31877256e23b46fb9fb9619833d4a58a":{"model_name":"LayoutModel","model_module":"@jupyter-widgets/base","model_module_version":"2.0.0","state":{}}}}},"nteract":{"version":"nteract-front-end@1.0.0"},"synapse_widget":{"version":"0.1","state":{}},"spark_compute":{"compute_id":"/trident/default","session_options":{"conf":{"spark.synapse.nbs.session.timeout":"1200000"}}},"dependencies":{"lakehouse":{}}},"nbformat":4,"nbformat_minor":5} \ No newline at end of file diff --git a/accelerators/CICD/Branch-out-to-new-workspace/README.md b/accelerators/CICD/Branch-out-to-new-workspace/README.md index f3d44ae..064824e 100644 --- a/accelerators/CICD/Branch-out-to-new-workspace/README.md +++ b/accelerators/CICD/Branch-out-to-new-workspace/README.md @@ -157,7 +157,7 @@ authentication method has been chosen: 11. This will display run-time parameters which can be modified as necessary: -
+
a. Source workspace: Name of the dev workspace
b. Target workspace: Name of the new workspace to be created which will also serve as the new branch name @@ -167,8 +167,9 @@ a. Source workspace: Name of the dev workspace
f. Swap connections in pipelines: Specify connections to be replaced using format (from 1 ,to1),(from2,to2),...(fromN,toN) using either connection ID or name.
g. Enter Developer Email: Add the email address of the developer to be granted admin role on the new workspace
h. Enter Capacity ID: Enter the capacity ID of the new workspace if different from the default GUID. -
i. Enter the branch name: Enter the source branch name which the new branch will be created from. -
j. Click Run and monitor the release pipeline in Azure Devops and also the progress of the post activity notebook in the Fabric monitoring hub. +
i. Enter the source branch name: Enter the source branch name which the new branch will be created from. +
j. Enter the Git folder where Fabric content is stored. Leave as / if content is stored in root. +
k. Click Run and monitor the release pipeline in Azure Devops and also the progress of the post activity notebook in the Fabric monitoring hub.
12. To debug and monitor the running YAML pipeline click on the “BranchOut” job diff --git a/accelerators/media/pipeline_params.png b/accelerators/media/pipeline_params.png index 1d52d41..c111b87 100644 Binary files a/accelerators/media/pipeline_params.png and b/accelerators/media/pipeline_params.png differ