def __init__(self, tmproot='.', bucket=None, prefix=None, run=None, s3root=None):
    "\n        Initialize a new context for S3 operations. This object is based used as\n        a context manager for a with statement.\n\n        There are two ways to initialize this object depending whether you want\n        to bind paths to a Metaflow run or not.\n\n        1. With a run object:\n\n            run: (required) Either a FlowSpec object (typically 'self') or a\n                 Run object corresponding to an existing Metaflow run. These\n                 are used to add a version suffix in the S3 path.\n            bucket: (optional) S3 bucket.\n            prefix: (optional) S3 prefix.\n\n        2. Without a run object:\n\n            s3root: (optional) An S3 root URL for all operations. If this is\n                    not specified, all operations require a full S3 URL.\n\n        These options are supported in both the modes:\n\n            tmproot: (optional) Root path for temporary files (default: '.')\n        "
    if run:
        parsed = urlparse(DATATOOLS_S3ROOT)
        if (not bucket):
            bucket = parsed.netloc
        if (not prefix):
            prefix = parsed.path
        if isinstance(run, FlowSpec):
            if current.is_running_flow:
                prefix = os.path.join(prefix, current.flow_name, current.run_id)
            else:
                raise MetaflowS3URLException('Initializing S3 with a FlowSpec outside of a running flow is not supported.')
        else:
            prefix = os.path.join(prefix, run.parent.id, run.id)
        self._s3root = (u's3://%s' % os.path.join(bucket, prefix.strip('/')))
    elif s3root:
        parsed = urlparse(to_unicode(s3root))
        if (parsed.scheme != 's3'):
            raise MetaflowS3URLException('s3root needs to be an S3 URL prefxied with s3://.')
        self._s3root = s3root.rstrip('/')
    else:
        self._s3root = None
    self._tmpdir = mkdtemp(dir=tmproot, prefix='metaflow.s3.')
Initialize a new context for S3 operations. This object is based used as
a context manager for a with statement.

There are two ways to initialize this object depending whether you want
to bind paths to a Metaflow run or not.

1. With a run object:

    run: (required) Either a FlowSpec object (typically 'self') or a
         Run object corresponding to an existing Metaflow run. These
         are used to add a version suffix in the S3 path.
    bucket: (optional) S3 bucket.
    prefix: (optional) S3 prefix.

2. Without a run object:

    s3root: (optional) An S3 root URL for all operations. If this is
            not specified, all operations require a full S3 URL.

These options are supported in both the modes:

    tmproot: (optional) Root path for temporary files (default: '.')
def __init__(self, tmproot='.', bucket=None, prefix=None, run=None, s3root=None):
    "\n        Initialize a new context for S3 operations. This object is based used as\n        a context manager for a with statement.\n\n        There are two ways to initialize this object depending whether you want\n        to bind paths to a Metaflow run or not.\n\n        1. With a run object:\n\n            run: (required) Either a FlowSpec object (typically 'self') or a\n                 Run object corresponding to an existing Metaflow run. These\n                 are used to add a version suffix in the S3 path.\n            bucket: (optional) S3 bucket.\n            prefix: (optional) S3 prefix.\n\n        2. Without a run object:\n\n            s3root: (optional) An S3 root URL for all operations. If this is\n                    not specified, all operations require a full S3 URL.\n\n        These options are supported in both the modes:\n\n            tmproot: (optional) Root path for temporary files (default: '.')\n        "
    if run:
        parsed = urlparse(DATATOOLS_S3ROOT)
        if (not bucket):
            bucket = parsed.netloc
        if (not prefix):
            prefix = parsed.path
        if isinstance(run, FlowSpec):
            if current.is_running_flow:
                prefix = os.path.join(prefix, current.flow_name, current.run_id)
            else:
                raise MetaflowS3URLException('Initializing S3 with a FlowSpec outside of a running flow is not supported.')
        else:
            prefix = os.path.join(prefix, run.parent.id, run.id)
        self._s3root = (u's3://%s' % os.path.join(bucket, prefix.strip('/')))
    elif s3root:
        parsed = urlparse(to_unicode(s3root))
        if (parsed.scheme != 's3'):
            raise MetaflowS3URLException('s3root needs to be an S3 URL prefxied with s3://.')
        self._s3root = s3root.rstrip('/')
    else:
        self._s3root = None
    self._tmpdir = mkdtemp(dir=tmproot, prefix='metaflow.s3.')