Repository: allenai/OLMo
File: olmo/util.py (Python)

Code before:
        )
        return response["Body"].read()


def flatten_dict(dictionary, parent_key="", separator=".", include_lists=False, root=""):
    d: Dict[str, Any] = {}
    for key, value in dictionary.items():
        new_key = parent_key + separator + key if parent_key else key
        if isinstance(value, MutableMapping):
            d.update(
                **flatten_dict(
                    value,
                    new_key,
                    separator=separator,
                    include_lists=include_lists,
                    root=root,
                )
            )
        elif isinstance(value, list) and include_lists:
            new_list = []
            for v in value:
                if isinstance(v, MutableMapping):
                    new_list.append(
                        flatten_dict(
                            v,
                            parent_key=root,
                            separator=separator,
                            include_lists=include_lists,
                            root=root,
                        )
                    )
                else:
                    new_list.append(v)
            d[new_key] = new_list
        else:
            d[new_key] = value
    return d


Review comment (suggestion):
The normal way of doing this is actually to treat a `List` as a `Mapping[int, Any]`. So the key becomes something like `"foo.bar.0"` and `"foo.bar.1"`, etc. Then you don't need the extra `root` parameter either.

Code after:
        from urllib.parse import urlparse

        parsed = urlparse(url)
        if not parsed.netloc or not parsed.path:
            raise ValueError("bad {} path {}".format(provider, url))
        bucket_name = parsed.netloc
        provider_path = parsed.path
        # Remove '/' at beginning of path.
        if provider_path.startswith("/"):
            provider_path = provider_path[1:]
        return bucket_name, provider_path

    def _ensure_object_info(self):
        if self.object_info is None:
            try:
                self.object_info = self.s3.head_object(Bucket=self.bucket_name, Key=self.path)
            except boto_exceptions.ClientError as e:
                if e.response["ResponseMetadata"]["HTTPStatusCode"] == 404:
                    raise FileNotFoundError(f"weka://{self.bucket_name}/{self.path}") from e
                raise e

    def get_etag(self) -> Optional[str]:
        self._ensure_object_info()
        assert self.object_info is not None
        return self.object_info.get("ETag")

    def get_size(self) -> Optional[int]:
        self._ensure_object_info()
        assert self.object_info is not None
        return self.object_info.get("ContentLength")

    def get_resource(self, temp_file: io.BufferedWriter) -> None:
        self.s3.download_fileobj(Fileobj=temp_file, Bucket=self.bucket_name, Key=self.path)

    def get_bytes_range(self, index: int, length: int) -> bytes:
        response = self.s3.get_object(
            Bucket=self.bucket_name, Key=self.path, Range=f"bytes={index}-{index+length-1}"
        )
        return response["Body"].read()


def flatten_dict(dictionary, parent_key="", separator=".", include_lists=False):
    """
    Flatten a nested dictionary into a single-level dictionary.

    Args:
        dictionary (dict): The nested dictionary to be flattened.
        parent_key (str, optional): The parent key to be prepended to the keys of the flattened dictionary. Defaults to "".
        separator (str, optional): The separator to be used between the parent key and the keys of the flattened dictionary. Defaults to ".".
        include_lists (bool, optional): Whether to convert lists to dictionaries with integer keys. Defaults to False.