def set_access_tier()

in parallel_change_blob_access_tier.py [0:0]

• 50 lines of code
• 12 McCabe index (conditional complexity)

def set_access_tier(container_client, blob_path, access_tier):
    
    assert access_tier in valid_tiers
    
    if verify_existence:
        
        if not blob_exists(container_client,blob_path):
           print('Warning: {} does not exist'.format(blob_path))
           return

    if verify_access_tier:
        
        try:
            
            blob_client = container_client.get_blob_client(blob_path)
            properties = blob_client.get_blob_properties()
            tier = properties['blob_tier']
            
            assert tier is not None, 'Error retrieving blob tier; is this a GPv1 storage account?'
            assert tier in valid_tiers, 'Unrecognized tier {} for {}'.format(
                tier,blob_path)
            
            # If this blob is already at the tier we want it
            if tier == access_tier:
                
                # Force the tier if tier forcing is requested *and* the 
                # tier is inferred, otherwise there's nothing to do here
                force_tier = force_tier_on_inferred_blobs and properties['tier_inferred']
                if not force_tier:
                    print('Skipping {}, already at tier {}'.format(
                        blob_path,access_tier))
                    return
                    
        except Exception as e:
            print('Error verifying access tier for {}: {}'.format(
                blob_path,str(e)))
                
    # ...if we're verifying the access tier
 
    if not execute_changes:
        
        if verbose:
            print('Debug: not setting {} to {}'.format(blob_path,access_tier))
        return
    
    try:

        check_archive_rehydration = (access_tier == 'Archive')
        
        if check_archive_rehydration:
            blob_client = container_client.get_blob_client(blob_path)
            properties = blob_client.get_blob_properties()
            original_tier = properties['blob_tier']
        
        if verbose:
            print('Setting {} to {}'.format(blob_path,access_tier))
            
        container_client.set_standard_blob_tier_blobs(access_tier,blob_path)
        
        # Verify that we've started rehydrating
        if check_archive_rehydration and (original_tier == 'Archive'):
            archive_status = properties['archive_status']
            if 'rehydrate-pending' not in archive_status:
                print('Error: blob {} not re-hydrating'.format(blob_path))
        
    except Exception as e:

        if verbose:
            s = str(e)
            if 'BlobNotFound' in s:
                print('{} does not exist'.format(blob_path))
            else:
                print('Error setting {} to {}: {}'.format(
                    blob_path,access_tier,s))
    
    if sleep_time_after_op > 0:
        time.sleep(sleep_time_after_op)