def main()

in actions/devel/upload/__main__.py [0:0]

• 29 lines of code
• 10 McCabe index (conditional complexity)

def main(args):
    """
    Action implementing a generic upload wrapper for the nuv devel plugin. The invoker must provide a x-impersonate-auth header containing the Openwhisk BASIC authentication of the wsku/user the action should impersonate 
    when calling this action. The upload action it is supposed to receive a path param similar to /<bucket>/<path>
    and will store the given path under the given MINIO <bucket>. The bucket must exists and the impersonated user must have write permission on it.
    """
    headers = args['__ow_headers']
    method = args['__ow_method']

    if(method.lower() not in ['put','delete'] ):
        return build_error(f"invalid request, HTTP verb {method} is not supported")
    
    if('x-impersonate-auth' not in headers):
        return build_error("invalid request, missing mandatory header: x-impersonate-auth")
    
    if(len(args['__ow_body']) == 0):
        return build_error("invalid request, no file content has been received")
    
    try:        
        upload_data = process_path_param(args['__ow_path'])

        if 'bucket' not in upload_data and 'filename' not in upload_data:
            return build_error("invalid request, bucket and/or filename path error")

        user_data = Authorize(args['couchdb_host'],args['couchdb_user'],args['couchdb_password']).login(headers['x-impersonate-auth'])
        mo_client = mutil.build_mo_client(ut.get_env_value(user_data,"S3_HOST"), ut.get_env_value(user_data,"S3_PORT"),ut.get_env_value(user_data,"S3_ACCESS_KEY")  , ut.get_env_value(user_data,"S3_SECRET_KEY"))        

        if method.lower() in 'put':
            print(f"processing request to upload file {upload_data['filename']}")
            content_as_b64 = args['__ow_body']            
            tmp_file = mutil.prepare_file_upload(user_data['login'],upload_data['filename'],content_as_b64)

            if tmp_file:
                    upload_result, upload_message = mutil.upload_file(mo_client,tmp_file,upload_data['bucket'],upload_data['filename'])
                    return build_response(user_data['login'],upload_data['filename'],upload_data['bucket'],upload_result, upload_message)
        
        if method.lower() in 'delete':
            print(f"processing request to delete file {upload_data['filename']}")
            delete_result = mutil.rm_file(mo_client, upload_data['bucket'], upload_data['filename'])
            return build_delete_response(user_data['login'],upload_data['filename'],upload_data['bucket'],delete_result,None)
        
        return build_error("Unexptected error upload action. Check activation log")
    except Exception as e:        
        return build_error(f"failed to execute nuv devel command. Reason: {e}")