def _grant_s3_table_bucket_lf_permissions()

in migration/bring-your-own-s3-tables/bring_your_own_s3_table_bucket.py [0:0]

• 65 lines of code
• 17 McCabe index (conditional complexity)

def _grant_s3_table_bucket_lf_permissions(lf_client, s3tables_client, project_role_arn, table_bucket_arn, 
                                          table_bucket_namespace, table_name, execute_flag):
    if not table_bucket_namespace and table_name:
        raise Exception(f"Error: Please provide namespace name along with the table name '{table_name}', or remove table name from input.")
    # Import all tables under provide S3 Table Bucket into SMUS Project
    elif not table_bucket_namespace and not table_name:
        next_token = None
        while True:
            if next_token:
                response = s3tables_client.list_tables(
                    tableBucketARN=table_bucket_arn,
                    nextToken=next_token
                )
            else:
                response = s3tables_client.list_tables(
                    tableBucketARN=table_bucket_arn
                )
            tables_list = response["tables"]
            for table in tables_list:
                for namespace in table["namespace"]:
                    _grant_table_lf_permissions(lf_client,
                                                s3tables_client,
                                                table_bucket_arn,
                                                namespace,
                                                table["name"],
                                                project_role_arn,
                                                execute_flag)
                
            if 'continuationToken' in response:
                next_token = response['continuationToken']
            else:
                break
    # Import all tables under provide S3 Table Bucket and namespace into SMUS Project
    elif table_bucket_namespace and not table_name:
        next_token = None
        while True:
            if next_token:
                response = s3tables_client.list_tables(
                    tableBucketARN=table_bucket_arn,
                    namespace=table_bucket_namespace,
                    nextToken=next_token
                )
            else:
                response = s3tables_client.list_tables(
                    tableBucketARN=table_bucket_arn,
                    namespace=table_bucket_namespace
                )
            tables_list = response["tables"]
            for table in tables_list:
                _grant_table_lf_permissions(lf_client,
                                            s3tables_client,
                                            table_bucket_arn,
                                            table_bucket_namespace,
                                            table["name"],
                                            project_role_arn,
                                            execute_flag)
                
            if 'continuationToken' in response:
                next_token = response['continuationToken']
            else:
                break
    # Import specific table into SMUS Project
    else:
        _grant_table_lf_permissions(lf_client,
                                    s3tables_client,
                                    table_bucket_arn,
                                    table_bucket_namespace,
                                    table_name,
                                    project_role_arn,
                                    execute_flag)