def pause_schedule()

in python/pipelines/pipeline_ops.py [0:0]

• 23 lines of code
• 5 McCabe index (conditional complexity)

def pause_schedule(
        project_id: str,
        region: str,
        pipeline_name: str) -> list:
    """
    This function pauses all schedules associated with a given pipeline name in a specific project and region.

    Args:
        project_id: The ID of the project that contains the pipeline.
        region: The location of the pipeline.
        pipeline_name: The name of the pipeline to pause schedules for.

    Returns:
        A list of the names of the paused schedules. If no schedules are found, returns None.
    
    Raises:
        Exception: If an error occurs while pausing the schedules.
    """

    # Get the list of schedules for the given pipeline name
    schedules = get_schedules(project_id, region, pipeline_name)
    if schedules is None:
        logging.info(f"No schedules found with display_name {pipeline_name}")
        return None

    # Creating the request header
    headers = requests.structures.CaseInsensitiveDict()
    headers["Content-Type"] = "application/json"
    headers["Authorization"] = "Bearer {}".format(get_gcp_bearer_token())

    # Pause the schedules where the display_name matches
    paused_schedules = []
    for s in schedules:
        url = f"https://{region}-aiplatform.googleapis.com/v1beta1/{s['name']}:pause"
        resp = requests.post(url=url, headers=headers)

        data = resp.json()  # Check the JSON Response Content
        print(resp.status_code == 200)
        if resp.status_code != 200:
            raise Exception(
                f"Unable to pause resourse {s['name']}. request returned with status code {resp.status_code}")
        logging.info(f"scheduled resourse {s['name']} paused")
        paused_schedules.append(s['name'])

    return paused_schedules