Repository: huggingface/datasets
Pull request: Refac module factory + avoid etag requests for hub datasets
## Refactor the module factory

When trying to extend the `data_files` logic to avoid doing unnecessary ETag requests, I noticed that the module preparation mechanism needed a refactor:
- the function was 600 lines long
- it was not readable
- it contained many different cases that made it complex to maintain
- it was hard to properly test it
- it was hard to extend without breaking anything

The module preparation mechanism is in charge of taking the name of a dataset or a metric given by the user (ex: "squad", "accuracy", "lhoestq/test", "path/to/my/script.py", "path/to/my/data/directory", "json", "csv") and return a module (possibly downloaded from the Hub) that contains the dataset builder or the metric class to use.

### Implementation details

I decided to separate all these use cases into different dataset/metric module factories.

First, the metric module factories:
- **CanonicalMetricModuleFactory**: "accuracy", "rouge", ...
- **LocalMetricModuleFactory**: "path/to/my/metric.py"

Then, the dataset module factories:
- **CanonicalDatasetModuleFactory**: "squad", "glue", ...
- **CommunityDatasetModuleFactoryWithScript**: "lhoestq/test"
- **CommunityDatasetModuleFactoryWithoutScript**: "lhoestq/demo1"
- **PackagedDatasetModuleFactory**: "json", "csv", ...
- **LocalDatasetModuleFactoryWithScript**: "path/to/my/script.py"
- **LocalDatasetModuleFactoryWithoutScript**: "path/to/my/data/directory"

And finally, additional factories when users have no internet:
- **CachedDatasetModuleFactory**
- **CachedMetricModuleFactory**

### Breaking changes

One thing is that I still don't know at what extent we want to keep backward compatibility for `prepare_module`. For now I just kept it (except I removed two parameters) just in case, but it's not used anywhere anymore.

## Avoid etag requests for hub datasets

To do this I added a class `DataFilesDict` that can be hashed to define the cache directory  of the dataset.
It contains the usual data files formatted as `{"train": ["train.txt"]}` for example.
But each list of file is a `DataFilesList` that also has a `origin_metadata` attribute that contains metadata about the origin of each file:
- for URLs: it stores the ETags of the files
- for local files: it stores the last modification data
- for files from a Hugging Face repository on the Hub: it stores the pattern (`*`, `*.csv`, "train.txt", etc.) and the commit sha of the repository (so there're no ETag requests !)

This way if any file changes, the hash of the `DataFilesDict` changes too !

You can instantiate a `DataFilesDict` by using patterns for local/remote files or files in a HF repository:
- for local/remote files: `DataFilesDict.from_local_or_remote(patterns)`
- for files in a HF repository: `DataFilesDict.from_hf_repo(patterns, dataset_info)`

Fix #2859 

## TODO

Fix the latest test:
- [x] fix the call to dataset_info in offline mode (related to https://github.com/huggingface/huggingface_hub/issues/372)

Add some more tests:
- [x] test all the factories
- [x] test the new data files logic

Other:
- [x] docstrings
- [x] comments

Problem statement:
Loading allenai/c4 in streaming mode does too many HEAD requests
This does 60,000+ HEAD requests to get all the ETags of all the data files:
```python
from datasets import load_dataset
load_dataset("allenai/c4", streaming=True)
```
It makes loading the dataset completely impractical.

The ETags are used to compute the config id (it must depend on the data files being used).
Instead of using the ETags, we could simply use the commit hash of the dataset repository on the hub, as well and the glob pattern used to resolve the files (here it's `*` by default, to load all the files of the repository)

Hints from issue discussion:
https://github.com/huggingface/datasets/blob/6c766f9115d686182d76b1b937cb27e099c45d68/src/datasets/builder.py#L179-L186

Commit message: style

Patch to review:
diff --git a/metrics/comet/comet.py b/metrics/comet/comet.py
index 7655eb09675..a2005c4d7e4 100644
--- a/metrics/comet/comet.py
+++ b/metrics/comet/comet.py
@@ -35,7 +35,7 @@
 ```
 """
 
-from comet.models import download_model  # From: unbabel-comet
+import comet.models  # From: unbabel-comet
 
 import datasets
 
@@ -121,7 +121,7 @@ def _info(self):
                     "references": datasets.Value("string", id="sequence"),
                 }
             ),
-            codebase_urls=["hhttps://github.com/Unbabel/COMET"],
+            codebase_urls=["https://github.com/Unbabel/COMET"],
             reference_urls=[
                 "https://github.com/Unbabel/COMET",
                 "https://www.aclweb.org/anthology/2020.emnlp-main.213/",
@@ -131,9 +131,9 @@ def _info(self):
 
     def _download_and_prepare(self, dl_manager):
         if self.config_name == "default":
-            self.scorer = download_model("wmt-large-da-estimator-1719")
+            self.scorer = comet.models.download_model("wmt-large-da-estimator-1719")
         else:
-            self.scorer = download_model(self.config_name)
+            self.scorer = comet.models.download_model(self.config_name)
 
     def _compute(self, sources, predictions, references, cuda=True, show_progress=False):
         data = {"src": sources, "mt": predictions, "ref": references}
diff --git a/setup.py b/setup.py
index c6b31cca167..d8f9a38b4ac 100644
--- a/setup.py
+++ b/setup.py
@@ -102,7 +102,7 @@
     # for data streaming via http
     "aiohttp",
     # To get datasets from the Datasets Hub on huggingface.co
-    "huggingface_hub>=0.0.14,<0.1.0",
+    "huggingface_hub>=0.0.18,<0.1.0",
     # Utilities from PyPA to e.g., compare versions
     "packaging",
 ]
diff --git a/src/datasets/builder.py b/src/datasets/builder.py
index 64b79a011a2..fdf8310093e 100644
--- a/src/datasets/builder.py
+++ b/src/datasets/builder.py
@@ -26,12 +26,10 @@
 import urllib
 from dataclasses import dataclass
 from functools import partial
-from pathlib import PurePath
-from typing import Dict, List, Mapping, Optional, Sequence, Tuple, Union
+from typing import Dict, Mapping, Optional, Tuple, Union
 
 from datasets.features import Features
 from datasets.utils.mock_download_manager import MockDownloadManager
-from datasets.utils.py_utils import map_nested
 
 from . import config, utils
 from .arrow_dataset import Dataset
@@ -43,6 +41,7 @@
     ReadInstruction,
 )
 from .arrow_writer import ArrowWriter, BeamWriter
+from .data_files import DataFilesDict, _sanitize_patterns
 from .dataset_dict import DatasetDict, IterableDatasetDict
 from .fingerprint import Hasher
 from .info import DatasetInfo, DatasetInfosDict, PostProcessedInfo
@@ -51,7 +50,7 @@
 from .splits import Split, SplitDict, SplitGenerator
 from .utils import logging
 from .utils.download_manager import DownloadManager, GenerateMode
-from .utils.file_utils import DownloadConfig, is_relative_path, is_remote_url, request_etags, url_or_path_join
+from .utils.file_utils import DownloadConfig, is_remote_url
 from .utils.filelock import FileLock
 from .utils.info_utils import get_size_checksum_dict, verify_checksums, verify_splits
 from .utils.streaming_download_manager import StreamingDownloadManager
@@ -90,7 +89,7 @@ class BuilderConfig:
     name: str = "default"
     version: Optional[Union[str, utils.Version]] = "0.0.0"
     data_dir: Optional[str] = None
-    data_files: Optional[Union[str, Sequence[str], Mapping[str, Union[str, Sequence[str]]]]] = None
+    data_files: Optional[DataFilesDict] = None
     description: Optional[str] = None
 
     def __post_init__(self):
@@ -99,12 +98,11 @@ def __post_init__(self):
         for invalid_char in invalid_windows_characters:
             if invalid_char in self.name:
                 raise InvalidConfigName(
-                    (
-                        "Bad characters from black list '{}' found in '{}'. "
-                        "They could create issues when creating a directory "
-                        "for this config on Windows filesystem."
-                    ).format(invalid_windows_characters, self.name)
+                    f"Bad characters from black list '{invalid_windows_characters}' found in '{self.name}'. "
+                    f"They could create issues when creating a directory for this config on Windows filesystem."
                 )
+        if self.data_files is not None and not isinstance(self.data_files, DataFilesDict):
+            raise ValueError(f"Expected a DataFilesDict in data_files but got {self.data_files}")
 
     def __eq__(self, o):
         # we need to override the default dataclass __eq__ since it doesn't check for
@@ -117,8 +115,6 @@ def create_config_id(
         self,
         config_kwargs: dict,
         custom_features: Optional[Features] = None,
-        use_auth_token: Optional[Union[bool, str]] = None,
-        base_path: Optional[Union[bool, str]] = None,
     ) -> str:
         """
         The config id is used to build the cache directory.
@@ -136,14 +132,12 @@ def create_config_id(
         # name and version are already used to build the cache directory
         config_kwargs_to_add_to_suffix.pop("name", None)
         config_kwargs_to_add_to_suffix.pop("version", None)
-        # data files are handled differently
-        config_kwargs_to_add_to_suffix.pop("data_files", None)
         # data dir handling (when specified it points to the manually downloaded data):
         # it was previously ignored before the introduction of config id because we didn't want
         # to change the config name. Now it's fine to take it into account for the config id.
         # config_kwargs_to_add_to_suffix.pop("data_dir", None)
         if "data_dir" in config_kwargs_to_add_to_suffix and config_kwargs_to_add_to_suffix["data_dir"] is None:
-            del config_kwargs_to_add_to_suffix["data_dir"]
+            config_kwargs_to_add_to_suffix.pop("data_dir", None)
         if config_kwargs_to_add_to_suffix:
             # we don't care about the order of the kwargs
             config_kwargs_to_add_to_suffix = {
@@ -158,49 +152,6 @@ def create_config_id(
             else:
                 suffix = Hasher.hash(config_kwargs_to_add_to_suffix)
 
-        if self.data_files is not None:
-            m = Hasher()
-            if suffix:
-                m.update(suffix)
-            if isinstance(self.data_files, str):
-                data_files = {"train": [self.data_files]}
-            elif isinstance(self.data_files, (tuple, list)):
-                data_files = {"train": self.data_files}
-            elif isinstance(self.data_files, dict):
-                data_files = {
-                    str(key): files if isinstance(files, (tuple, list)) else [files]
-                    for key, files in self.data_files.items()
-                }
-            else:
-                raise ValueError("Please provide a valid `data_files` in `DatasetBuilder`")
-
-            def abspath(data_file) -> str:
-                data_file = data_file.as_posix() if isinstance(data_file, PurePath) else str(data_file)
-                return url_or_path_join(base_path, data_file) if is_relative_path(data_file) else data_file
-
-            data_files: Dict[str, List[str]] = map_nested(abspath, data_files)
-            remote_urls = [
-                data_file for key in data_files for data_file in data_files[key] if is_remote_url(data_file)
-            ]
-            etags = dict(
-                zip(
-                    remote_urls,
-                    request_etags(
-                        remote_urls, use_auth_token=use_auth_token, tqdm_kwargs={"desc": "Check remote data files"}
-                    ),
-                )
-            )
-            for key in sorted(data_files.keys()):
-                m.update(key)
-                for data_file in data_files[key]:
-                    if is_remote_url(data_file):
-                        m.update(data_file)
-                        m.update(etags[data_file])
-                    else:
-                        m.update(os.path.abspath(data_file))
-                        m.update(str(os.path.getmtime(data_file)))
-            suffix = m.hexdigest()
-
         if custom_features is not None:
             m = Hasher()
             if suffix:
@@ -255,6 +206,8 @@ def __init__(
         features: Optional[Features] = None,
         use_auth_token: Optional[Union[bool, str]] = None,
         namespace: Optional[str] = None,
+        data_files: Optional[Union[str, list, dict, DataFilesDict]] = None,
+        data_dir: Optional[str] = None,
         **config_kwargs,
     ):
         """Constructs a DatasetBuilder.
@@ -274,6 +227,12 @@ def __init__(
                 It can be used to changed the :obj:`datasets.Features` description of a dataset for example.
             use_auth_token (:obj:`str` or :obj:`bool`, optional): Optional string or boolean to use as Bearer token
                 for remote files on the Datasets Hub. If True, will get token from ``"~/.huggingface"``.
+            namespace: `str`, used to separate builders with the same name but not coming from the same namespace.
+                For example to separate "squad" from "lhoestq/squad" (the builder name would be "lhoestq___squad").
+            data_files: for builders like "csv" or "json" that need the user to specify data files. They can be either
+                local or remote files. For convenience you can use a DataFilesDict.
+            data_files: `str`, for builders that require manual download. It must be the path to the local directory containing
+                the manually downloaded data.
             config_kwargs: will override the defaults kwargs in config
 
         """
@@ -284,14 +243,18 @@ def __init__(
         self.use_auth_token = use_auth_token
         self.namespace = namespace
 
+        if data_files is not None and not isinstance(data_files, DataFilesDict):
+            data_files = DataFilesDict.from_local_or_remote(
+                _sanitize_patterns(data_files), base_path=base_path, use_auth_token=use_auth_token
+            )
+
         # Prepare config: DatasetConfig contains name, version and description but can be extended by each dataset
         if "features" in inspect.signature(self.BUILDER_CONFIG_CLASS.__init__).parameters and features is not None:
             config_kwargs["features"] = features
-        # Discard default config parameters
-        if "data_files" in config_kwargs and config_kwargs["data_files"] is None:
-            del config_kwargs["data_files"]
-        if "data_dir" in config_kwargs and config_kwargs["data_dir"] is None:
-            del config_kwargs["data_dir"]
+        if data_files is not None:
+            config_kwargs["data_files"] = data_files
+        if data_dir is not None:
+            config_kwargs["data_dir"] = data_dir
         self.config, self.config_id = self._create_builder_config(
             name,
             custom_features=features,
@@ -412,8 +375,6 @@ def _create_builder_config(self, name=None, custom_features=None, **config_kwarg
         config_id = builder_config.create_config_id(
             config_kwargs,
             custom_features=custom_features,
-            use_auth_token=self.use_auth_token,
-            base_path=self.base_path if self.base_path is not None else "",
         )
         is_custom = config_id not in self.builder_configs
         if is_custom:
@@ -1193,7 +1154,7 @@ def _prepare_split(self, split_generator):
         generator = self._generate_tables(**split_generator.gen_kwargs)
         with ArrowWriter(features=self.info.features, path=fpath) as writer:
             for key, table in utils.tqdm(
-                generator, unit=" tables", leave=False, disable=bool(logging.get_verbosity() == logging.NOTSET)
+                generator, unit=" tables", leave=False, disable=True  # bool(logging.get_verbosity() == logging.NOTSET)
             ):
                 writer.write_table(table)
             num_examples, num_bytes = writer.finalize()
diff --git a/src/datasets/config.py b/src/datasets/config.py
index 200b1b436c9..0eec2f5f60e 100644
--- a/src/datasets/config.py
+++ b/src/datasets/config.py
@@ -110,7 +110,7 @@
 USE_BEAM = os.environ.get("USE_BEAM", "AUTO").upper()
 BEAM_VERSION = "N/A"
 BEAM_AVAILABLE = False
-if USE_BEAM in ("1", "ON", "YES", "AUTO"):
+if USE_BEAM in ENV_VARS_TRUE_AND_AUTO_VALUES:
     try:
         BEAM_VERSION = version.parse(importlib_metadata.version("apache_beam"))
         BEAM_AVAILABLE = True
@@ -150,6 +150,11 @@
 DEFAULT_EXTRACTED_DATASETS_PATH = os.path.join(DEFAULT_DOWNLOADED_DATASETS_PATH, EXTRACTED_DATASETS_DIR)
 EXTRACTED_DATASETS_PATH = Path(os.getenv("HF_DATASETS_EXTRACTED_DATASETS_PATH", DEFAULT_EXTRACTED_DATASETS_PATH))
 
+# Download count for the website
+HF_UPDATE_DOWNLOAD_COUNTS = (
+    os.environ.get("HF_UPDATE_DOWNLOAD_COUNTS", "AUTO").upper() in ENV_VARS_TRUE_AND_AUTO_VALUES
+)
+
 # Batch size constants. For more info, see:
 # https://github.com/apache/arrow/blob/master/docs/source/cpp/arrays.rst#size-limitations-and-recommendations)
 DEFAULT_MAX_BATCH_SIZE = 10_000
@@ -159,11 +164,7 @@
 MAX_TABLE_NBYTES_FOR_PICKLING = 4 << 30
 
 # Offline mode
-HF_DATASETS_OFFLINE = os.environ.get("HF_DATASETS_OFFLINE", "AUTO").upper()
-if HF_DATASETS_OFFLINE in ("1", "ON", "YES"):
-    HF_DATASETS_OFFLINE = True
-else:
-    HF_DATASETS_OFFLINE = False
+HF_DATASETS_OFFLINE = os.environ.get("HF_DATASETS_OFFLINE", "AUTO").upper() in ENV_VARS_TRUE_VALUES
 
 # In-memory
 DEFAULT_IN_MEMORY_MAX_SIZE = 0  # Disabled
diff --git a/src/datasets/data_files.py b/src/datasets/data_files.py
new file mode 100644
index 00000000000..fd451cde41b
--- /dev/null
+++ b/src/datasets/data_files.py
@@ -0,0 +1,328 @@
+import glob
+import os
+from functools import partial
+from pathlib import Path, PurePath
+from typing import Dict, List, Optional, Tuple, Union
+
+import huggingface_hub
+from tqdm.contrib.concurrent import thread_map
+
+from .splits import Split
+from .utils import logging
+from .utils.file_utils import is_relative_path, is_remote_url, request_etag
+from .utils.tqdm_utils import tqdm
+
+
+DEFAULT_SPLIT = str(Split.TRAIN)
+
+
+logger = logging.get_logger(__name__)
+
+
+class Url(str):
+    pass
+
+
+def _sanitize_patterns(patterns: Union[Dict, List, str, None]) -> Dict[str, Union[List[str], "DataFilesList"]]:
+    """
+    Take the data_files patterns from the user, and format them into a dictionary.
+    Each key is the name of the split, and each value is a list of data files patterns (paths or urls).
+    The default split is "train".
+
+    Returns:
+        patterns: dictionary of split_name -> list_of _atterns
+    """
+    if patterns is None:
+        return {DEFAULT_SPLIT: ["*"]}
+    if isinstance(patterns, dict):
+        return {str(key): value if isinstance(value, list) else [value] for key, value in patterns.items()}
+    elif isinstance(patterns, str):
+        return {DEFAULT_SPLIT: [patterns]}
+    else:
+        return {DEFAULT_SPLIT: list(patterns)}
+
+
+def _resolve_single_pattern_locally(
+    base_path: str, pattern: str, allowed_extensions: Optional[List[str]] = None
+) -> List[Path]:
+    """
+    Return the absolute paths to all the files that match the given patterns.
+    It also supports absolute paths in patterns.
+    If an URL is passed, it is returned as is.
+    """
+    data_files_ignore = ["README.md", "config.json", "dataset_infos.json", "dummy_data.zip"]
+    if is_relative_path(pattern):
+        glob_iter = list(Path(base_path).rglob(pattern))
+    else:
+        glob_iter = [Path(filepath) for filepath in glob.glob(pattern)]
+
+    matched_paths = [
+        filepath.resolve()
+        for filepath in glob_iter
+        if filepath.name not in data_files_ignore and not filepath.name.startswith(".") and filepath.is_file()
+    ]
+    if allowed_extensions is not None:
+        out = [
+            filepath
+            for filepath in matched_paths
+            if any(suffix[1:] in allowed_extensions for suffix in filepath.suffixes)
+        ]
+        if len(out) < len(matched_paths):
+            invalid_matched_files = list(set(matched_paths) - set(out))
+            logger.info(
+                f"Some files matched the pattern '{pattern}' at {Path(base_path).resolve()} but don't have valid data file extensions: {invalid_matched_files}"
+            )
+    else:
+        out = matched_paths
+    if not out:
+        error_msg = f"Unable to resolve any data file that matches '{pattern}' at {Path(base_path).resolve()}"
+        if allowed_extensions is not None:
+            error_msg += f" with any supported extension {list(allowed_extensions)}"
+        raise FileNotFoundError(error_msg)
+    return out
+
+
+def resolve_patterns_locally_or_by_urls(
+    base_path: str, patterns: List[str], allowed_extensions: Optional[List[str]] = None
+) -> List[Union[Path, Url]]:
+    """
+    Resolve the paths and URLs of the data files from the patterns passed by the user.
+    URLs are just returned as is.
+
+    Examples:
+
+        >>> import huggingface_hub
+        >>> from datasets.data_files import resolve_patterns_in_dataset_repository
+        >>> base_path = /Users/username/Desktop/hf/datasets
+        >>> resolve_patterns_locally_or_by_urls(base_path, ["src/**/*.yaml"])
+        [PosixPath('/Users/quentinlhoest/Desktop/hf/shirte/datasets/src/datasets/utils/resources/readme_structure.yaml')]
+
+    Args:
+        base_path (str): Base path to use when resolving relative paths.
+        patterns (List[str]): Unix patterns or paths or URLs of the data files to resolve.
+            The paths can be absolute or relative to base_path.
+        allowed_extensions (Optional[list], optional): White-list of file extensions to use. Defaults to None (all extensions).
+            For example: allowed_extensions=["csv", "json", "txt", "parquet"]
+
+    Returns:
+        List[Union[Path, Url]]: List of paths or URLs to the local or remote files that match the patterns.
+    """
+    data_files = []
+    for pattern in patterns:
+        if is_remote_url(pattern):
+            data_files.append(Url(pattern))
+        else:
+            data_files.extend(_resolve_single_pattern_locally(base_path, pattern, allowed_extensions))
+    return data_files
+
+
+def _exec_patterns_in_dataset_repository(
+    dataset_info: huggingface_hub.hf_api.DatasetInfo,
+    patterns: List[str],
+    allowed_extensions: Optional[list] = None,
+) -> List[PurePath]:
+    data_files_ignore = ["README.md", "config.json", "dataset_infos.json", "dummy_data.zip"]
+    all_data_files = [
+        PurePath("/" + dataset_file.rfilename) for dataset_file in dataset_info.siblings
+    ]  # add a / at the beginning to make the pattern **/* match files at the root
+    filenames = []
+    for pattern in patterns:
+        matched_paths = [
+            filepath.relative_to("/")
+            for filepath in all_data_files
+            if filepath.name not in data_files_ignore and not filepath.name.startswith(".") and filepath.match(pattern)
+        ]
+        if allowed_extensions is not None:
+            out = [
+                filepath
+                for filepath in matched_paths
+                if any(suffix[1:] in allowed_extensions for suffix in filepath.suffixes)
+            ]
+            if len(out) < len(matched_paths):
+                invalid_matched_files = list(set(matched_paths) - set(out))
+                logger.info(
+                    f"Some files matched the pattern {pattern} in dataset repository {dataset_info.id} but don't have valid data file extensions: {invalid_matched_files}"
+                )
+        else:
+            out = matched_paths
+        if not out:
+            error_msg = f"Unable to resolve data_file {pattern} in dataset repository {dataset_info.id}"
+            if allowed_extensions is not None:
+                error_msg += f" with any supported extension {list(allowed_extensions)}"
+            raise FileNotFoundError(error_msg)
+        filenames.extend(out)
+    return filenames
+
+
+def resolve_patterns_in_dataset_repository(
+    dataset_info: huggingface_hub.hf_api.DatasetInfo,
+    patterns: List[str],
+    allowed_extensions: Optional[list] = None,
+) -> List[Url]:
+    """
+    Resolve the URLs of the data files from the patterns passed by the user.
+
+    Examples:
+
+        >>> import huggingface_hub
+        >>> from datasets.data_files import resolve_patterns_in_dataset_repository
+        >>> dataset_info = huggingface_hub.HfApi().dataset_info("lhoestq/demo1")
+        >>> resolve_patterns_in_dataset_repository(dataset_info, ["*.csv"])
+        ['https://huggingface.co/datasets/lhoestq/demo1/resolve/0ca0d9f35b390ad11516095aeb27fd30cfe72578/data/test.csv',
+        'https://huggingface.co/datasets/lhoestq/demo1/resolve/0ca0d9f35b390ad11516095aeb27fd30cfe72578/data/train.csv']
+
+    Args:
+        dataset_info (huggingface_hub.hf_api.DatasetInfo): dataset info obtained using the hugginggace_hub.HfApi
+        patterns (List[str]): Unix patterns or paths of the files in the dataset repository.
+            The paths should be relative to the root of the repository.
+        allowed_extensions (Optional[list], optional): White-list of file extensions to use. Defaults to None (all extensions).
+            For example: allowed_extensions=["csv", "json", "txt", "parquet"]
+
+    Returns:
+        List[Url]: List of URLs to the files in the dataset repository that match the patterns.
+    """
+    data_files_urls: List[Url] = []
+    for filename in _exec_patterns_in_dataset_repository(dataset_info, patterns, allowed_extensions):
+        data_files_urls.append(
+            Url(
+                huggingface_hub.hf_hub_url(
+                    dataset_info.id, filename.as_posix(), revision=dataset_info.sha, repo_type="dataset"
+                )
+            )
+        )
+    return data_files_urls
+
+
+def _get_single_origin_metadata_locally_or_by_urls(
+    data_file: Union[Path, Url], use_auth_token: Optional[Union[bool, str]] = None
+) -> Tuple[str]:
+    if isinstance(data_file, Url):
+        data_file = str(data_file)
+        return (request_etag(data_file, use_auth_token=use_auth_token),)
+    else:
+        data_file = str(data_file.resolve())
+        return (str(os.path.getmtime(data_file)),)
+
+
+def _get_origin_metadata_locally_or_by_urls(
+    data_files: List[Union[Path, Url]], max_workers=64, use_auth_token: Optional[Union[bool, str]] = None
+) -> Tuple[str]:
+    return thread_map(
+        partial(_get_single_origin_metadata_locally_or_by_urls, use_auth_token=use_auth_token),
+        data_files,
+        max_workers=max_workers,
+        tqdm_class=tqdm,
+        desc="Resolving data files",
+        disable=len(data_files) <= 16 or logging.get_verbosity() == logging.NOTSET,
+    )
+
+
+class DataFilesList(List[Union[Path, Url]]):
+    """
+    List of data files (absolute local paths or URLs).
+    It has two construction methods given the user's data files patterns :
+    - ``from_hf_repo``: resolve patterns inside a dataset repository
+    - ``from_local_or_remote``: resolve patterns from a local path
+
+    Moreover DataFilesList has an additional attribute ``origin_metadata``.
+    It can store:
+    - the last modified time of local files
+    - ETag of remote files
+    - commit sha of a dataset repository
+
+    Thanks to this additional attribute, it is possible to hash the list
+    and get a different hash if and only if at least one file changed.
+    This is useful for caching Dataset objects that are obtained from a list of data files.
+    """
+
+    def __init__(self, data_files: List[Union[Path, Url]], origin_metadata: List[Tuple[str]]):
+        super().__init__(data_files)
+        self.origin_metadata = origin_metadata
+
+    @staticmethod
+    def from_hf_repo(
+        patterns: List[str],
+        dataset_info: huggingface_hub.hf_api.DatasetInfo,
+        allowed_extensions: Optional[List[str]] = None,
+    ) -> "DataFilesList":
+        data_files = resolve_patterns_in_dataset_repository(dataset_info, patterns, allowed_extensions)
+        origin_metadata = [(dataset_info.id, dataset_info.sha) for _ in patterns]
+        return DataFilesList(data_files, origin_metadata)
+
+    @staticmethod
+    def from_local_or_remote(
+        patterns: List[str],
+        base_path: Optional[str] = None,
+        allowed_extensions: Optional[List[str]] = None,
+        use_auth_token: Optional[Union[bool, str]] = None,
+    ) -> "DataFilesList":
+        base_path = base_path if base_path is not None else str(Path().resolve())
+        data_files = resolve_patterns_locally_or_by_urls(base_path, patterns, allowed_extensions)
+        origin_metadata = _get_origin_metadata_locally_or_by_urls(data_files, use_auth_token=use_auth_token)
+        return DataFilesList(data_files, origin_metadata)
+
+
+class DataFilesDict(Dict[str, DataFilesList]):
+    """
+    Dict of split_name -> list of data files (absolute local paths or URLs).
+    It has two construction methods given the user's data files patterns :
+    - ``from_hf_repo``: resolve patterns inside a dataset repository
+    - ``from_local_or_remote``: resolve patterns from a local path
+
+    Moreover each list is a DataFilesList. It is possible to hash the dictionary
+    and get a different hash if and only if at least one file changed.
+    For more info, see ``DataFilesList``.
+
+    This is useful for caching Dataset objects that are obtained from a list of data files.
+
+    Changing the order of the keys of this dictionary also doesn't change its hash.
+    """
+
+    @staticmethod
+    def from_local_or_remote(
+        patterns: Dict[str, Union[List[str], DataFilesList]],
+        base_path: Optional[str] = None,
+        allowed_extensions: Optional[List[str]] = None,
+        use_auth_token: Optional[Union[bool, str]] = None,
+    ) -> "DataFilesDict":
+        out = DataFilesDict()
+        for key, patterns_for_key in patterns.items():
+            out[key] = (
+                DataFilesList.from_local_or_remote(
+                    patterns_for_key,
+                    base_path=base_path,
+                    allowed_extensions=allowed_extensions,
+                    use_auth_token=use_auth_token,
+                )
+                if not isinstance(patterns_for_key, DataFilesList)
+                else patterns_for_key
+            )
+        return out
+
+    @staticmethod
+    def from_hf_repo(
+        patterns: Dict[str, Union[List[str], DataFilesList]],
+        dataset_info: huggingface_hub.hf_api.DatasetInfo,
+        allowed_extensions: Optional[List[str]] = None,
+    ) -> "DataFilesDict":
+        out = DataFilesDict()
+        for key, patterns_for_key in patterns.items():
+            out[key] = (
+                DataFilesList.from_hf_repo(
+                    patterns_for_key, dataset_info=dataset_info, allowed_extensions=allowed_extensions
+                )
+                if not isinstance(patterns_for_key, DataFilesList)
+                else patterns_for_key
+            )
+        return out
+
+    def __reduce__(self):
+        """
+        To make sure the order of the keys doesn't matter when pickling and hashing:
+
+        >>> from datasets.data_files import DataFilesDict
+        >>> from datasets.fingerprint import Hasher
+        >>> assert Hasher.hash(DataFilesDict(a=[], b=[])) == Hasher.hash(DataFilesDict(b=[], a=[]))
+
+        """
+        return DataFilesDict, (dict(sorted(self.items())),)
diff --git a/src/datasets/load.py b/src/datasets/load.py
index b652900eeb6..1ea7e8dba1f 100644
--- a/src/datasets/load.py
+++ b/src/datasets/load.py
@@ -16,7 +16,6 @@
 # Lint as: python3
 """Access datasets."""
 import filecmp
-import glob
 import importlib
 import inspect
 import json
@@ -26,23 +25,24 @@
 import time
 import warnings
 from collections import Counter
-from pathlib import Path, PurePath
+from dataclasses import dataclass
+from pathlib import Path
 from typing import Dict, List, Mapping, Optional, Sequence, Tuple, Type, Union
 from urllib.parse import urlparse
 
 import fsspec
-import huggingface_hub
+import requests
 from huggingface_hub import HfApi
 
 from . import config
 from .arrow_dataset import Dataset
 from .builder import DatasetBuilder
+from .data_files import DataFilesDict, DataFilesList, _sanitize_patterns
 from .dataset_dict import DatasetDict, IterableDatasetDict
 from .features import Features
 from .filesystems import extract_path_from_uri, is_remote_filesystem
 from .iterable_dataset import IterableDataset
 from .metric import Metric
-from .naming import camelcase_to_snakecase
 from .packaged_modules import _EXTENSION_TO_MODULE, _PACKAGED_DATASETS_MODULES, hash_python_lines
 from .splits import Split
 from .streaming import extend_module_for_streaming
@@ -50,26 +50,27 @@
 from .utils.download_manager import GenerateMode
 from .utils.file_utils import (
     DownloadConfig,
+    OfflineModeIsEnabled,
+    _raise_if_offline_mode_is_enabled,
     cached_path,
     head_hf_s3,
     hf_github_url,
     hf_hub_url,
     init_hf_modules,
     is_relative_path,
-    is_remote_url,
     relative_to_absolute_path,
     url_or_path_join,
-    url_or_path_parent,
 )
 from .utils.filelock import FileLock
 from .utils.info_utils import is_small_dataset
 from .utils.logging import get_logger
-from .utils.py_utils import NestedDataStructure
 from .utils.version import Version
 
 
 logger = get_logger(__name__)
 
+DEFAULT_SPLIT = str(Split.TRAIN)
+
 
 def init_dynamic_modules(
     name: str = config.MODULE_NAME_FOR_DYNAMIC_MODULES, hf_modules_cache: Optional[Union[Path, str]] = None
@@ -153,7 +154,16 @@ def convert_github_url(url_path: str) -> Tuple[str, Optional[str]]:
     return url_path, sub_directory
 
 
-def get_imports(file_path: str):
+def increase_load_count(name: str, resource_type: str):
+    """Update the download count of a dataset or metric."""
+    if not config.HF_DATASETS_OFFLINE and config.HF_UPDATE_DOWNLOAD_COUNTS:
+        try:
+            head_hf_s3(name, filename=name + ".py", dataset=(resource_type == "dataset"))
+        except Exception:
+            pass
+
+
+def get_imports(file_path: str) -> Tuple[str, str, str, str]:
     r"""Find whether we should import or clone additional files for a given processing script.
         And list the import.
 
@@ -226,161 +236,762 @@ def get_imports(file_path: str):
     return imports
 
 
-def _resolve_data_files_locally_or_by_urls(
-    base_path: str, patterns: Union[str, List[str], Dict], allowed_extensions: Optional[list] = None
-) -> Union[List[Path], Dict]:
+def _download_additional_modules(
+    name: str, base_path: str, imports: Tuple[str, str, str, str], download_config: Optional[DownloadConfig]
+) -> List[Tuple[str, str]]:
     """
-    Return the absolute paths to all the files that match the given patterns.
-    It also supports absolute paths in patterns.
-    If an URL is passed, it is returned as is."""
-    data_files_ignore = ["README.md", "config.json", "dataset_infos.json", "dummy_data.zip"]
-    if isinstance(patterns, str):
-        if is_remote_url(patterns):
-            return [patterns]
-        if is_relative_path(patterns):
-            glob_iter = list(Path(base_path).rglob(patterns))
+    Download additional module for a module <name>.py at URL (or local path) <base_path>/<name>.py
+    The imports must have been parsed first using ``get_imports``.
+
+    If some modules need to be installed with pip, an error is raised showing how to install them.
+    This function return the list of downloaded modules as tuples (import_name, module_file_path).
+
+    The downloaded modules can then be moved into an importable directory with ``_copy_script_and_other_resouces_in_importable_dir``.
+    """
+    local_imports = []
+    library_imports = []
+    for import_type, import_name, import_path, sub_directory in imports:
+        if import_type == "library":
+            library_imports.append((import_name, import_path))  # Import from a library
+            continue
+
+        if import_name == name:
+            raise ValueError(
+                f"Error in the {name} script, importing relative {import_name} module "
+                f"but {import_name} is the name of the script. "
+                f"Please change relative import {import_name} to another name and add a '# From: URL_OR_PATH' "
+                f"comment pointing to the original relative import file path."
+            )
+        if import_type == "internal":
+            url_or_filename = url_or_path_join(base_path, import_path + ".py")
+        elif import_type == "external":
+            url_or_filename = import_path
         else:
-            glob_iter = [Path(filepath) for filepath in glob.glob(patterns)]
+            raise ValueError("Wrong import_type")
 
-        matched_paths = [
-            filepath.resolve()
-            for filepath in glob_iter
-            if filepath.name not in data_files_ignore and not filepath.name.startswith(".") and filepath.is_file()
-        ]
-        if allowed_extensions is not None:
-            out = [
-                filepath
-                for filepath in matched_paths
-                if any(suffix[1:] in allowed_extensions for suffix in filepath.suffixes)
-            ]
-            if len(out) < len(matched_paths):
-                invalid_matched_files = list(set(matched_paths) - set(out))
-                logger.info(
-                    f"Some files matched the pattern '{patterns}' at {Path(base_path).resolve()} but don't have valid data file extensions: {invalid_matched_files}"
+        local_import_path = cached_path(
+            url_or_filename,
+            download_config=download_config,
+        )
+        if sub_directory is not None:
+            local_import_path = os.path.join(local_import_path, sub_directory)
+        local_imports.append((import_name, local_import_path))
+
+    # Check library imports
+    needs_to_be_installed = []
+    for library_import_name, library_import_path in library_imports:
+        try:
+            lib = importlib.import_module(library_import_name)  # noqa F841
+        except ImportError:
+            needs_to_be_installed.append((library_import_name, library_import_path))
+    if needs_to_be_installed:
+        raise ImportError(
+            f"To be able to use {name}, you need to install the following dependencies"
+            f"{[lib_name for lib_name, lib_path in needs_to_be_installed]} using 'pip install "
+            f"{' '.join([lib_path for lib_name, lib_path in needs_to_be_installed])}' for instance'"
+        )
+    return local_imports
+
+
+def _copy_script_and_other_resouces_in_importable_dir(
+    name: str,
+    importable_directory_path: str,
+    subdirrectory_name: str,
+    original_local_path: str,
+    local_imports: List[Tuple[str, str]],
+    additional_files: List[Tuple[str, str]],
+    download_mode: Optional[GenerateMode],
+) -> str:
+    """Copy a script and its required imports to an importable directory
+
+    Args:
+        name (str): name of the resource to load
+        importable_directory_path (str): path to the loadable folder in the dynamic modules directory
+        subdirrectory_name (str): name of the subdirectory in importable_directory_path in which to place the script
+        original_local_path (str): local path to the resource script
+        local_imports (List[Tuple[str, str]]): list of (destination_filename, import_file_to_copy)
+        additional_files (List[Tuple[str, str]]): list of (destination_filename, additional_file_to_copy)
+        download_mode (Optional[GenerateMode]): download mode
+
+    Return:
+        importable_local_file: path to an importable module with importlib.import_module
+    """
+
+    # Define a directory with a unique name in our dataset or metric folder
+    # path is: ./datasets|metrics/dataset|metric_name/hash_from_code/script.py
+    # we use a hash as subdirrectory_name to be able to have multiple versions of a dataset/metric processing file together
+    importable_subdirectory = os.path.join(importable_directory_path, subdirrectory_name)
+    importable_local_file = os.path.join(importable_subdirectory, name + ".py")
+
+    # Prevent parallel disk operations
+    lock_path = importable_directory_path + ".lock"
+    with FileLock(lock_path):
+        # Create main dataset/metrics folder if needed
+        if download_mode == GenerateMode.FORCE_REDOWNLOAD and os.path.exists(importable_directory_path):
+            shutil.rmtree(importable_directory_path)
+        os.makedirs(importable_directory_path, exist_ok=True)
+
+        # add an __init__ file to the main dataset folder if needed
+        init_file_path = os.path.join(importable_directory_path, "__init__.py")
+        if not os.path.exists(init_file_path):
+            with open(init_file_path, "w"):
+                pass
+
+        # Create hash dataset folder if needed
+        os.makedirs(importable_subdirectory, exist_ok=True)
+        # add an __init__ file to the hash dataset folder if needed
+        init_file_path = os.path.join(importable_subdirectory, "__init__.py")
+        if not os.path.exists(init_file_path):
+            with open(init_file_path, "w"):
+                pass
+
+        # Copy dataset.py file in hash folder if needed
+        if not os.path.exists(importable_local_file):
+            shutil.copyfile(original_local_path, importable_local_file)
+
+        # Record metadata associating original dataset path with local unique folder
+        meta_path = importable_local_file.split(".py")[0] + ".json"
+        if not os.path.exists(meta_path):
+            meta = {"original file path": original_local_path, "local file path": importable_local_file}
+            # the filename is *.py in our case, so better rename to filenam.json instead of filename.py.json
+            with open(meta_path, "w", encoding="utf-8") as meta_file:
+                json.dump(meta, meta_file)
+
+        # Copy all the additional imports
+        for import_name, import_path in local_imports:
+            if os.path.isfile(import_path):
+                full_path_local_import = os.path.join(importable_subdirectory, import_name + ".py")
+                if not os.path.exists(full_path_local_import):
+                    shutil.copyfile(import_path, full_path_local_import)
+            elif os.path.isdir(import_path):
+                full_path_local_import = os.path.join(importable_subdirectory, import_name)
+                if not os.path.exists(full_path_local_import):
+                    shutil.copytree(import_path, full_path_local_import)
+            else:
+                raise OSError(f"Error with local import at {import_path}")
+
+        # Copy aditional files like dataset infos file if needed
+        for file_name, original_path in additional_files:
+            destination_additional_path = os.path.join(importable_subdirectory, file_name)
+            if not os.path.exists(destination_additional_path) or not filecmp.cmp(
+                original_path, destination_additional_path
+            ):
+                shutil.copyfile(original_path, destination_additional_path)
+        return importable_local_file
+
+
+def infer_module_for_data_files(data_files_list: DataFilesList) -> Optional[str]:
+    extensions_counter = Counter(suffix[1:] for filepath in data_files_list for suffix in Path(filepath).suffixes)
+    if extensions_counter:
+        return _EXTENSION_TO_MODULE[extensions_counter.most_common(1)[0][0]]
+
+
+@dataclass
+class DatasetModuleFactoryResult:
+    module_path: str
+    hash: str
+    builder_kwargs: dict
+
+
+@dataclass
+class MetricModuleFactoryResult:
+    module_path: str
+    hash: str
+
+
+class _DatasetModuleFactory:
+    def get_module(self) -> DatasetModuleFactoryResult:
+        raise NotImplementedError
+
+
+class _MetricModuleFactory:
+    def get_module(self) -> MetricModuleFactoryResult:
+        raise NotImplementedError
+
+
+class CanonicalDatasetModuleFactory(_DatasetModuleFactory):
+    """Get the module of a canonical dataset. The dataset script is downloaded from GitHub."""
+
+    def __init__(
+        self,
+        name: str,
+        revision: Optional[Union[str, Version]] = None,
+        download_config: Optional[DownloadConfig] = None,
+        download_mode: Optional[GenerateMode] = None,
+        dynamic_modules_path: Optional[str] = None,
+    ):
+        self.name = name
+        self.revision = revision
+        self.download_config = download_config
+        self.download_mode = download_mode
+        self.dynamic_modules_path = dynamic_modules_path
+        assert self.name.count("/") == 0
+        increase_load_count(name, resource_type="dataset")
+
+    def download_dataset_script(self) -> str:
+        file_path = hf_github_url(path=self.name, name=self.name + ".py", revision=self.revision)
+        return cached_path(file_path, download_config=self.download_config)
+
+    def download_dataset_script_from_master(self) -> str:
+        file_path = hf_github_url(path=self.name, name=self.name + ".py", revision="master")
+        local_path = cached_path(file_path, download_config=self.download_config)
+        logger.warning(
+            f"Couldn't find a directory or a dataset named '{self.name}'. "
+            f"It was picked from the master branch on github instead at {file_path}"
+        )
+        return local_path
+
+    def download_dataset_infos_file(self, revision: Optional[str]) -> str:
+        dataset_infos = hf_github_url(path=self.name, name=config.DATASETDICT_INFOS_FILENAME, revision=revision)
+        # Download the dataset infos file if available
+        try:
+            return cached_path(
+                dataset_infos,
+                download_config=self.download_config,
+            )
+        except (FileNotFoundError, ConnectionError):
+            return None
+
+    def get_module(self) -> DatasetModuleFactoryResult:
+        # get script and other files
+        try:
+            local_path = self.download_dataset_script()
+            revision = self.revision
+        except FileNotFoundError:
+            if self.revision is not None:
+                raise
+            else:
+                local_path = self.download_dataset_script_from_master()
+                logger.warning(
+                    f"Dataset '{self.name}' not available for this version, using the one from the 'master' branch instead."
                 )
-        else:
-            out = matched_paths
-        if not out:
-            error_msg = f"Unable to resolve any data file that matches '{patterns}' at {Path(base_path).resolve()}"
-            if allowed_extensions is not None:
-                error_msg += f" with any supported extension {list(allowed_extensions)}"
-            raise FileNotFoundError(error_msg)
-        return out
-    elif isinstance(patterns, dict):
-        return {
-            k: _resolve_data_files_locally_or_by_urls(base_path, v, allowed_extensions=allowed_extensions)
-            for k, v in patterns.items()
+                revision = "master"
+        dataset_infos_path = self.download_dataset_infos_file(revision)
+        imports = get_imports(local_path)
+        local_imports = _download_additional_modules(
+            name=self.name,
+            base_path=hf_github_url(path=self.name, name="", revision=revision),
+            imports=imports,
+            download_config=self.download_config,
+        )
+        additional_files = [(config.DATASETDICT_INFOS_FILENAME, dataset_infos_path)] if dataset_infos_path else []
+        # the copy the script and the files in an importable directory
+        dynamic_modules_path = self.dynamic_modules_path if self.dynamic_modules_path else init_dynamic_modules()
+        importable_directory_path = os.path.join(dynamic_modules_path, "datasets", self.name)
+        Path(importable_directory_path).mkdir(parents=True, exist_ok=True)
+        (Path(importable_directory_path).parent / "__init__.py").touch(exist_ok=True)
+        hash = files_to_hash([local_path] + [loc[1] for loc in local_imports])
+        importable_local_file = _copy_script_and_other_resouces_in_importable_dir(
+            name=self.name,
+            importable_directory_path=importable_directory_path,
+            subdirrectory_name=hash,
+            original_local_path=local_path,
+            local_imports=local_imports,
+            additional_files=additional_files,
+            download_mode=self.download_mode,
+        )
+        logger.debug(f"Created importable dataset file at {importable_local_file}")
+        # make the new module to be noticed by the import system
+        importlib.invalidate_caches()
+        module_path = ".".join([os.path.basename(dynamic_modules_path), "datasets", self.name, hash, self.name])
+        builder_kwargs = {"hash": hash, "base_path": hf_github_url(self.name, "", revision=revision)}
+        return DatasetModuleFactoryResult(module_path, hash, builder_kwargs)
+
+
+class CanonicalMetricModuleFactory(_MetricModuleFactory):
+    """Get the module of a metric. The metric script is downloaded from GitHub."""
+
+    def __init__(
+        self,
+        name: str,
+        revision: Optional[Union[str, Version]] = None,
+        download_config: Optional[DownloadConfig] = None,
+        download_mode: Optional[GenerateMode] = None,
+        dynamic_modules_path: Optional[str] = None,
+    ):
+        self.name = name
+        self.revision = revision
+        self.download_config = download_config
+        self.download_mode = download_mode
+        self.dynamic_modules_path = dynamic_modules_path
+        assert self.name.count("/") == 0
+        increase_load_count(name, resource_type="metric")
+
+    def download_metric_script(self) -> str:
+        file_path = hf_github_url(path=self.name, name=self.name + ".py", revision=self.revision, dataset=False)
+        return cached_path(file_path, download_config=self.download_config)
+
+    def download_metric_script_from_master(self) -> str:
+        file_path = hf_github_url(path=self.name, name=self.name + ".py", revision="master", dataset=False)
+        local_path = cached_path(file_path, download_config=self.download_config)
+        logger.warning(
+            f"Couldn't find a directory or a metric named '{self.name}'. "
+            f"It was picked from the master branch on github instead at {file_path}"
+        )
+        return local_path
+
+    def get_module(self) -> MetricModuleFactoryResult:
+        # get script and other files
+        try:
+            local_path = self.download_metric_script()
+            revision = self.revision
+        except FileNotFoundError:
+            if self.revision is not None:
+                raise
+            else:
+                local_path = self.download_metric_script_from_master()
+                revision = "master"
+        imports = get_imports(local_path)
+        local_imports = _download_additional_modules(
+            name=self.name,
+            base_path=hf_github_url(path=self.name, name="", revision=revision),
+            imports=imports,
+            download_config=self.download_config,
+        )
+        # the copy the script and the files in an importable directory
+        dynamic_modules_path = self.dynamic_modules_path if self.dynamic_modules_path else init_dynamic_modules()
+        importable_directory_path = os.path.join(dynamic_modules_path, "metrics", self.name)
+        Path(importable_directory_path).mkdir(parents=True, exist_ok=True)
+        (Path(importable_directory_path).parent / "__init__.py").touch(exist_ok=True)
+        hash = files_to_hash([local_path] + [loc[1] for loc in local_imports])
+        importable_local_file = _copy_script_and_other_resouces_in_importable_dir(
+            name=self.name,
+            importable_directory_path=importable_directory_path,
+            subdirrectory_name=hash,
+            original_local_path=local_path,
+            local_imports=local_imports,
+            additional_files=[],
+            download_mode=self.download_mode,
+        )
+        logger.debug(f"Created importable metric file at {importable_local_file}")
+        # make the new module to be noticed by the import system
+        importlib.invalidate_caches()
+        module_path = ".".join([os.path.basename(dynamic_modules_path), "metrics", self.name, hash, self.name])
+        return MetricModuleFactoryResult(module_path, hash)
+
+
+class LocalMetricModuleFactory(_MetricModuleFactory):
+    """Get the module of a local metric. The metric script is loaded from a local script."""
+
+    def __init__(
+        self,
+        path: str,
+        download_config: Optional[DownloadConfig] = None,
+        download_mode: Optional[GenerateMode] = None,
+        dynamic_modules_path: Optional[str] = None,
+    ):
+        self.path = path
+        self.name = Path(path).stem
+        self.download_config = download_config
+        self.download_mode = download_mode
+        self.dynamic_modules_path = dynamic_modules_path
+
+    def get_module(self) -> MetricModuleFactoryResult:
+        # get script and other files
+        imports = get_imports(self.path)
+        local_imports = _download_additional_modules(
+            name=self.name,
+            base_path=str(Path(self.path).parent),
+            imports=imports,
+            download_config=self.download_config,
+        )
+        # the copy the script and the files in an importable directory
+        dynamic_modules_path = self.dynamic_modules_path if self.dynamic_modules_path else init_dynamic_modules()
+        importable_directory_path = os.path.join(dynamic_modules_path, "metrics", self.name)
+        Path(importable_directory_path).mkdir(parents=True, exist_ok=True)
+        (Path(importable_directory_path).parent / "__init__.py").touch(exist_ok=True)
+        hash = files_to_hash([self.path] + [loc[1] for loc in local_imports])
+        importable_local_file = _copy_script_and_other_resouces_in_importable_dir(
+            name=self.name,
+            importable_directory_path=importable_directory_path,
+            subdirrectory_name=hash,
+            original_local_path=self.path,
+            local_imports=local_imports,
+            additional_files=[],
+            download_mode=self.download_mode,
+        )
+        logger.debug(f"Created importable dataset file at {importable_local_file}")
+        # make the new module to be noticed by the import system
+        importlib.invalidate_caches()
+        module_path = ".".join([os.path.basename(dynamic_modules_path), "metrics", self.name, hash, self.name])
+        return MetricModuleFactoryResult(module_path, hash)
+
+
+class LocalDatasetModuleFactoryWithScript(_DatasetModuleFactory):
+    """Get the module of a local dataset. The dataset script is loaded from a local script."""
+
+    def __init__(
+        self,
+        path: str,
+        download_config: Optional[DownloadConfig] = None,
+        download_mode: Optional[GenerateMode] = None,
+        dynamic_modules_path: Optional[str] = None,
+    ):
+        self.path = path
+        self.name = Path(path).stem
+        self.download_config = download_config
+        self.download_mode = download_mode
+        self.dynamic_modules_path = dynamic_modules_path
+
+    def get_module(self) -> DatasetModuleFactoryResult:
+        # get script and other files
+        dataset_infos_path = Path(self.path).parent / config.DATASETDICT_INFOS_FILENAME
+        imports = get_imports(self.path)
+        local_imports = _download_additional_modules(
+            name=self.name,
+            base_path=str(Path(self.path).parent),
+            imports=imports,
+            download_config=self.download_config,
+        )
+        additional_files = (
+            [(config.DATASETDICT_INFOS_FILENAME, str(dataset_infos_path))] if dataset_infos_path.is_file() else []
+        )
+        # the copy the script and the files in an importable directory
+        dynamic_modules_path = self.dynamic_modules_path if self.dynamic_modules_path else init_dynamic_modules()
+        importable_directory_path = os.path.join(dynamic_modules_path, "datasets", self.name)
+        Path(importable_directory_path).mkdir(parents=True, exist_ok=True)
+        (Path(importable_directory_path).parent / "__init__.py").touch(exist_ok=True)
+        hash = files_to_hash([self.path] + [loc[1] for loc in local_imports])
+        importable_local_file = _copy_script_and_other_resouces_in_importable_dir(
+            name=self.name,
+            importable_directory_path=importable_directory_path,
+            subdirrectory_name=hash,
+            original_local_path=self.path,
+            local_imports=local_imports,
+            additional_files=additional_files,
+            download_mode=self.download_mode,
+        )
+        logger.debug(f"Created importable dataset file at {importable_local_file}")
+        # make the new module to be noticed by the import system
+        importlib.invalidate_caches()
+        module_path = ".".join([os.path.basename(dynamic_modules_path), "datasets", self.name, hash, self.name])
+        builder_kwargs = {"hash": hash, "base_path": str(Path(self.path).parent)}
+        return DatasetModuleFactoryResult(module_path, hash, builder_kwargs)
+
+
+class LocalDatasetModuleFactoryWithoutScript(_DatasetModuleFactory):
+    """Get the module of a dataset loaded from the user's data files. The dataset builder module to use is infered from the data files extensions."""
+
+    def __init__(
+        self,
+        path: str,
+        data_files: Optional[Union[str, List, Dict]] = None,
+        download_mode: Optional[GenerateMode] = None,
+    ):
+        self.path = path
+        self.name = Path(path).stem
+        self.data_files = data_files
+        self.download_mode = download_mode
+
+    def get_module(self) -> DatasetModuleFactoryResult:
+        data_files = DataFilesDict.from_local_or_remote(
+            _sanitize_patterns(self.data_files), base_path=self.path, allowed_extensions=_EXTENSION_TO_MODULE.keys()
+        )
+        infered_module_names = {
+            key: infer_module_for_data_files(data_files_list) for key, data_files_list in data_files.items()
         }
-    else:
-        return sum(
-            [
-                _resolve_data_files_locally_or_by_urls(base_path, pattern, allowed_extensions=allowed_extensions)
-                for pattern in patterns
-            ],
-            [],
+        if len(set(list(infered_module_names.values()))) > 1:
+            raise ValueError(f"Couldn't infer the same data file format for all splits. Got {infered_module_names}")
+        infered_module_name = next(iter(infered_module_names.values()))
+        if not infered_module_name:
+            raise FileNotFoundError(f"No data files or dataset script found in {self.path}")
+        module_path, hash = _PACKAGED_DATASETS_MODULES[infered_module_name]
+        builder_kwargs = {
+            "hash": hash,
+            "data_files": data_files,
+            "name": os.path.basename(self.path),
+            "base_path": self.path,
+        }
+        return DatasetModuleFactoryResult(module_path, hash, builder_kwargs)
+
+
+class PackagedDatasetModuleFactory(_DatasetModuleFactory):
+    """Get the dataset builder module from the ones that are packaged with the library: csv, json, etc."""
+
+    def __init__(
+        self,
+        name: str,
+        data_files: Optional[Union[str, List, Dict]] = None,
+        download_config: Optional[DownloadConfig] = None,
+        download_mode: Optional[GenerateMode] = None,
+    ):
+        self.name = name
+        self.data_files = data_files
+        self.downnload_config = download_config
+        self.download_mode = download_mode
+        increase_load_count(name, resource_type="dataset")
+
+    def get_module(self) -> DatasetModuleFactoryResult:
+        data_files = DataFilesDict.from_local_or_remote(
+            _sanitize_patterns(self.data_files), use_auth_token=self.downnload_config.use_auth_token
         )
+        module_path, hash = _PACKAGED_DATASETS_MODULES[self.name]
+        builder_kwargs = {"hash": hash, "data_files": data_files}
+        return DatasetModuleFactoryResult(module_path, hash, builder_kwargs)
 
 
-def _resolve_data_files_in_dataset_repository(
-    dataset_info: huggingface_hub.hf_api.DatasetInfo,
-    patterns: Union[str, List[str], Dict],
-    allowed_extensions: Optional[list] = None,
-) -> Union[List[PurePath], Dict]:
-    data_files_ignore = ["README.md", "config.json", "dataset_infos.json", "dummy_data.zip"]
-    if isinstance(patterns, str):
-        all_data_files = [
-            PurePath("/" + dataset_file.rfilename) for dataset_file in dataset_info.siblings
-        ]  # add a / at the beginning to make the pattern **/* match files at the root
-        matched_paths = [
-            filepath.relative_to("/")
-            for filepath in all_data_files
-            if filepath.name not in data_files_ignore
-            and not filepath.name.startswith(".")
-            and filepath.match(patterns)
-        ]
-        if allowed_extensions is not None:
-            out = [
-                filepath
-                for filepath in matched_paths
-                if any(suffix[1:] in allowed_extensions for suffix in filepath.suffixes)
+class CommunityDatasetModuleFactoryWithoutScript(_DatasetModuleFactory):
+    """
+    Get the module of a dataset loaded from data files of a dataset repository.
+    The dataset builder module to use is infered from the data files extensions.
+    """
+
+    def __init__(
+        self,
+        name: str,
+        revision: Optional[Union[str, Version]] = None,
+        data_files: Optional[Union[str, List, Dict]] = None,
+        download_config: Optional[DownloadConfig] = None,
+        download_mode: Optional[GenerateMode] = None,
+    ):
+        self.name = name
+        self.revision = revision
+        self.data_files = data_files
+        self.download_config = download_config
+        self.download_mode = download_mode
+        assert self.name.count("/") == 1
+        increase_load_count(name, resource_type="dataset")
+
+    def get_module(self) -> DatasetModuleFactoryResult:
+        dataset_info = HfApi(config.HF_ENDPOINT).dataset_info(
+            self.name,
+            revision=self.revision,
+            token=self.download_config.use_auth_token,
+            timeout=100.0,
+        )
+        data_files = DataFilesDict.from_hf_repo(
+            _sanitize_patterns(self.data_files),
+            dataset_info=dataset_info,
+            allowed_extensions=_EXTENSION_TO_MODULE.keys(),
+        )
+        infered_module_names = {
+            key: infer_module_for_data_files(data_files_list) for key, data_files_list in data_files.items()
+        }
+        if len(set(list(infered_module_names.values()))) > 1:
+            raise ValueError(f"Couldn't infer the same data file format for all splits. Got {infered_module_names}")
+        infered_module_name = next(iter(infered_module_names.values()))
+        if not infered_module_name:
+            raise FileNotFoundError(f"No data files or dataset script found in {self.path}")
+        module_path, hash = _PACKAGED_DATASETS_MODULES[infered_module_name]
+        builder_kwargs = {
+            "hash": hash,
+            "data_files": data_files,
+            "name": self.name.replace("/", "___"),
+            "base_path": hf_hub_url(self.name, "", revision=self.revision),
+        }
+        return DatasetModuleFactoryResult(module_path, hash, builder_kwargs)
+
+
+class CommunityDatasetModuleFactoryWithScript(_DatasetModuleFactory):
+    """Get the module of a dataset from a dataset repository. The dataset script comes from the script inside the dataset repository."""
+
+    def __init__(
+        self,
+        name: str,
+        revision: Optional[Union[str, Version]] = None,
+        download_config: Optional[DownloadConfig] = None,
+        download_mode: Optional[GenerateMode] = None,
+        dynamic_modules_path: Optional[str] = None,
+    ):
+        self.name = name
+        self.revision = revision
+        self.download_config = download_config
+        self.download_mode = download_mode
+        self.dynamic_modules_path = dynamic_modules_path
+        assert self.name.count("/") == 1
+        increase_load_count(name, resource_type="dataset")
+
+    def download_dataset_script(self) -> str:
+        file_path = hf_hub_url(path=self.name, name=self.name.split("/")[1] + ".py", revision=self.revision)
+        return cached_path(file_path, download_config=self.download_config)
+
+    def download_dataset_infos_file(self) -> str:
+        dataset_infos = hf_hub_url(path=self.name, name=config.DATASETDICT_INFOS_FILENAME, revision=self.revision)
+        # Download the dataset infos file if available
+        try:
+            return cached_path(
+                dataset_infos,
+                download_config=self.download_config,
+            )
+        except (FileNotFoundError, ConnectionError):
+            return None
+
+    def get_module(self) -> DatasetModuleFactoryResult:
+        # get script and other files
+        local_path = self.download_dataset_script()
+        dataset_infos_path = self.download_dataset_infos_file()
+        imports = get_imports(local_path)
+        local_imports = _download_additional_modules(
+            name=self.name,
+            base_path=hf_hub_url(path=self.name, name="", revision=self.revision),
+            imports=imports,
+            download_config=self.download_config,
+        )
+        additional_files = [(config.DATASETDICT_INFOS_FILENAME, dataset_infos_path)] if dataset_infos_path else []
+        # the copy the script and the files in an importable directory
+        dynamic_modules_path = self.dynamic_modules_path if self.dynamic_modules_path else init_dynamic_modules()
+        importable_directory_path = os.path.join(dynamic_modules_path, "datasets", self.name.replace("/", "___"))
+        Path(importable_directory_path).mkdir(parents=True, exist_ok=True)
+        (Path(importable_directory_path).parent / "__init__.py").touch(exist_ok=True)
+        (Path(importable_directory_path) / "__init__.py").touch(exist_ok=True)
+        hash = files_to_hash([local_path] + [loc[1] for loc in local_imports])
+        importable_local_file = _copy_script_and_other_resouces_in_importable_dir(
+            name=self.name.split("/")[1],
+            importable_directory_path=importable_directory_path,
+            subdirrectory_name=hash,
+            original_local_path=local_path,
+            local_imports=local_imports,
+            additional_files=additional_files,
+            download_mode=self.download_mode,
+        )
+        logger.debug(f"Created importable dataset file at {importable_local_file}")
+        # make the new module to be noticed by the import system
+        importlib.invalidate_caches()
+        module_path = ".".join(
+            [
+                os.path.basename(dynamic_modules_path),
+                "datasets",
+                self.name.replace("/", "___"),
+                hash,
+                self.name.split("/")[1],
             ]
-            if len(out) < len(matched_paths):
-                invalid_matched_files = list(set(matched_paths) - set(out))
-                logger.info(
-                    f"Some files matched the pattern {patterns} in dataset repository {dataset_info.id} but don't have valid data file extensions: {invalid_matched_files}"
-                )
-        else:
-            out = matched_paths
-        if not out:
-            error_msg = f"Unable to resolve data_file {patterns} in dataset repository {dataset_info.id}"
-            if allowed_extensions is not None:
-                error_msg += f" with any supported extension {list(allowed_extensions)}"
-            raise FileNotFoundError(error_msg)
-        return out
-    elif isinstance(patterns, dict):
-        return {
-            k: _resolve_data_files_in_dataset_repository(dataset_info, v, allowed_extensions=allowed_extensions)
-            for k, v in patterns.items()
+        )
+        builder_kwargs = {
+            "hash": hash,
+            "base_path": hf_hub_url(self.name, "", revision=self.revision),
+            "namespace": self.name.split("/")[0],
         }
-    else:
-        return sum(
+        return DatasetModuleFactoryResult(module_path, hash, builder_kwargs)
+
+
+class CachedDatasetModuleFactory(_DatasetModuleFactory):
+    """
+    Get the module of a dataset that has been loaded once already and cached.
+    The script that is loaded from the cache is the most recent one with a matching name.
+    """
+
+    def __init__(
+        self,
+        name: str,
+        dynamic_modules_path: Optional[str] = None,
+    ):
+        self.name = name
+        self.dynamic_modules_path = dynamic_modules_path
+        assert self.name.count("/") <= 1
+
+    def get_module(self) -> DatasetModuleFactoryResult:
+        dynamic_modules_path = self.dynamic_modules_path if self.dynamic_modules_path else init_dynamic_modules()
+        importable_directory_path = os.path.join(dynamic_modules_path, "datasets", self.name.replace("/", "___"))
+        hashes = (
+            [h for h in os.listdir(importable_directory_path) if len(h) == 64]
+            if os.path.isdir(importable_directory_path)
+            else None
+        )
+        if not hashes:
+            raise FileNotFoundError(f"Dataset {self.name} is not cached in {dynamic_modules_path}")
+        # get most recent
+
+        def _get_modification_time(module_hash):
+            return (Path(importable_directory_path) / module_hash / (self.name.split("/")[-1] + ".py")).stat().st_mtime
+
+        hash = sorted(hashes, key=_get_modification_time)[-1]
+        logger.warning(
+            f"Using the latest cached version of the module from {os.path.join(importable_directory_path, hash)} "
+            f"(last modified on {time.ctime(_get_modification_time(hash))}) since it "
+            f"couldn't be found locally at {self.name}, or remotely on the Hugging Face Hub."
+        )
+        # make the new module to be noticed by the import system
+        module_path = ".".join(
             [
-                _resolve_data_files_in_dataset_repository(dataset_info, pattern, allowed_extensions=allowed_extensions)
-                for pattern in patterns
-            ],
-            [],
+                os.path.basename(dynamic_modules_path),
+                "datasets",
+                self.name.replace("/", "___"),
+                hash,
+                self.name.split("/")[-1],
+            ]
         )
+        importlib.invalidate_caches()
+        builder_kwargs = {"hash": hash, "namespace": self.name.split("/")[0]}
+        return DatasetModuleFactoryResult(module_path, hash, builder_kwargs)
 
 
-def _infer_module_for_data_files(data_files: Union[PurePath, List[PurePath], Dict]) -> Optional[str]:
-    extensions_counter = Counter(
-        suffix[1:] for filepath in NestedDataStructure(data_files).flatten() for suffix in filepath.suffixes
-    )
-    if extensions_counter:
-        return _EXTENSION_TO_MODULE[extensions_counter.most_common(1)[0][0]]
+class CachedMetricModuleFactory(_MetricModuleFactory):
+    """
+    Get the module of a metric that has been loaded once already and cached.
+    The script that is loaded from the cache is the most recent one with a matching name.
+    """
 
+    def __init__(
+        self,
+        name: str,
+        dynamic_modules_path: Optional[str] = None,
+    ):
+        self.name = name
+        self.dynamic_modules_path = dynamic_modules_path
+        assert self.name.count("/") == 0
+
+    def get_module(self) -> MetricModuleFactoryResult:
+        dynamic_modules_path = self.dynamic_modules_path if self.dynamic_modules_path else init_dynamic_modules()
+        importable_directory_path = os.path.join(dynamic_modules_path, "metrics", self.name)
+        hashes = (
+            [h for h in os.listdir(importable_directory_path) if len(h) == 64]
+            if os.path.isdir(importable_directory_path)
+            else None
+        )
+        if not hashes:
+            raise FileNotFoundError(f"Metric {self.name} is not cached in {dynamic_modules_path}")
+        # get most recent
+
+        def _get_modification_time(module_hash):
+            return (Path(importable_directory_path) / module_hash / (self.name + ".py")).stat().st_mtime
+
+        hash = sorted(hashes, key=_get_modification_time)[-1]
+        logger.warning(
+            f"Using the latest cached version of the module from {os.path.join(importable_directory_path, hash)} "
+            f"(last modified on {time.ctime(_get_modification_time(hash))}) since it "
+            f"couldn't be found locally at {self.name}, or remotely on the Hugging Face Hub."
+        )
+        # make the new module to be noticed by the import system
+        module_path = ".".join([os.path.basename(dynamic_modules_path), "metrics", self.name, hash, self.name])
+        importlib.invalidate_caches()
+        return MetricModuleFactoryResult(module_path, hash)
 
-def prepare_module(
+
+def dataset_module_factory(
     path: str,
     revision: Optional[Union[str, Version]] = None,
     download_config: Optional[DownloadConfig] = None,
     download_mode: Optional[GenerateMode] = None,
-    dataset: bool = True,
     force_local_path: Optional[str] = None,
     dynamic_modules_path: Optional[str] = None,
-    return_resolved_file_path: bool = False,
-    return_associated_base_path: bool = False,
-    return_namespace: bool = False,
-    data_files: Optional[Union[Dict, List, str]] = None,
-    script_version="deprecated",
+    data_files: Optional[Union[Dict, List, str, DataFilesDict]] = None,
     **download_kwargs,
-) -> Union[Tuple[str, str], Tuple[str, str, Optional[str]]]:
+) -> DatasetModuleFactoryResult:
     r"""
-        Download/extract/cache a dataset (if dataset==True) or a metric (if dataset==False)
+    Download/extract/cache a dataset module.
 
-    Dataset and metrics codes are cached inside the the dynamic modules cache to allow easy import (avoid ugly sys.path tweaks)
-    and using cloudpickle (among other things).
+    Dataset codes are cached inside the the dynamic modules cache to allow easy import (avoid ugly sys.path tweaks).
 
     Args:
 
-        path (str): Path or name of the dataset, or path to a metric script.
-            Depending on ``path``, the module that is returned id either generic moduler (csv, json, text etc.) or a module defined defined a dataset or metric script (a python file).
+        path (str): Path or name of the dataset.
+            Depending on ``path``, the module that is returned is either generic module (csv, json, text etc.) or a module defined by dataset cript (a python file).
 
             For local datasets:
 
-            - if ``path`` is a local directory (but doesn't contain a dataset script)
+            - if ``path`` is a local directory containing data files (but doesn't contain a dataset script)
               -> load a generic module (csv, json, text etc.) based on the content of the directory
               e.g. ``'./path/to/directory/with/my/csv/data'``.
-            - if ``path`` is a local dataset or metric script or a directory containing a local dataset or metric script (if the script has the same name as the directory):
-              -> load the module from the dataset or metric script
+            - if ``path`` is a local dataset script or a directory containing a local dataset script (if the script has the same name as the directory):
+              -> load the module from the dataset script
               e.g. ``'./dataset/squad'`` or ``'./dataset/squad/squad.py'``.
 
             For datasets on the Hugging Face Hub (list all available datasets and ids with ``datasets.list_datasets()``)
 
-            - if ``path`` is a canonical dataset or metric on the HF Hub (ex: `glue`, `squad`)
-              -> load the module from the dataset or metric script in the github repository at huggingface/datasets
-              e.g. ``'squad'`` or ``'glue'`` or ``accuracy``.
-            - if ``path`` is a dataset repository on the HF hub (without a dataset script)
+            - if ``path`` is a canonical dataset on the HF Hub (ex: `glue`, `squad`)
+              -> load the module from the dataset script in the github repository at huggingface/datasets
+              e.g. ``'squad'`` or ``'glue'``.
+            - if ``path`` is a dataset repository on the HF hub containing data files (without a dataset script)
               -> load a generic module (csv, text etc.) based on the content of the repository
               e.g. ``'username/dataset_name'``, a dataset repository on the HF hub containing your data files.
             - if ``path`` is a dataset repository on the HF hub with a dataset script (if the script has the same name as the directory)
@@ -388,24 +999,21 @@ def prepare_module(
               e.g. ``'username/dataset_name'``, a dataset repository on the HF hub containing a dataset script `'dataset_name.py'`.
 
         revision (Optional ``Union[str, datasets.Version]``):
-            If specified, the module will be loaded from the datasets repository at this version.
-            By default:
+            Which revision of a dataset repository to use, or which revision of a canonical dataset to use.
+            If specified, the module will be loaded at this version.
+            By default for dataset repositories:
+            - for dataset repositories, it is set to the "main" branch
+            By default for canonical datasets:
             - it is set to the local version of the lib.
             - it will also try to load it from the master branch if it's not available at the local version of the lib.
             Specifying a version that is different from your local version of the lib might cause compatibility issues.
         download_config (:class:`DownloadConfig`, optional): Specific download configuration parameters.
         download_mode (:class:`GenerateMode`, default ``REUSE_DATASET_IF_EXISTS``): Download/generate mode.
-        dataset (bool): True if the script to load is a dataset, False if the script is a metric.
         force_local_path (Optional str): Optional path to a local path to download and prepare the script to.
             Used to inspect or modify the script folder.
         dynamic_modules_path (Optional str, defaults to HF_MODULES_CACHE / "datasets_modules", i.e. ~/.cache/huggingface/modules/datasets_modules):
             Optional path to the directory in which the dynamic modules are saved. It must have been initialized with :obj:`init_dynamic_modules`.
             By default the datasets and metrics are stored inside the `datasets_modules` module.
-        return_resolved_file_path (Optional bool, defaults to False):
-            If True, the url or path to the resolved dataset or metric script is returned with the other ouputs
-        return_associated_base_path (Optional bool, defaults to False):
-            If True, the base path associated to the dataset is returned with the other ouputs.
-            It corresponds to the directory or base url where the dataset script/dataset repo is at.
         data_files (:obj:`Union[Dict, List, str]`, optional): Defining the data_files of the dataset configuration.
         script_version:
             .. deprecated:: 1.13
@@ -413,417 +1021,257 @@ def prepare_module(
         download_kwargs: optional attributes for DownloadConfig() which will override the attributes in download_config if supplied.
 
     Returns:
-        Tuple[``str``, ``str``]:
-        1. The module path being
-            - the import path of the dataset/metric package if force_local_path is False: e.g. 'datasets.datasets.squad'
-            - the local path to the dataset/metric file if force_local_path is True: e.g. '/User/huggingface/datasets/datasets/squad/squad.py'
-        2. A hash string computed from the content of the dataset loading script.
+        DatasetModuleFactoryResult
     """
-    if script_version != "deprecated":
-        warnings.warn(
-            "'script_version' was renamed to 'revision' in version 1.13 and will be removed in 1.15.", FutureWarning
-        )
-        revision = script_version
     if download_config is None:
         download_config = DownloadConfig(**download_kwargs)
     download_config.extract_compressed_file = True
     download_config.force_extract = True
 
-    module_type = "dataset" if dataset else "metric"
-    name = list(filter(lambda x: x, path.replace(os.sep, "/").split("/")))[-1]
-    if not name.endswith(".py"):
-        name = name + ".py"
+    filename = list(filter(lambda x: x, path.replace(os.sep, "/").split("/")))[-1]
+    if not filename.endswith(".py"):
+        filename = filename + ".py"
+    combined_path = os.path.join(path, filename)
 
-    # Short name is name without the '.py' at the end (for the module)
-    short_name = name[:-3]
-
-    # first check if the module is packaged with the `datasets` package
-    def prepare_packaged_module(name):
-        try:
-            head_hf_s3(name, filename=name + ".py", dataset=dataset, max_retries=download_config.max_retries)
-        except Exception:
-            logger.debug(f"Couldn't head HF s3 for packaged dataset module '{name}'. Running in offline mode.")
-        return _PACKAGED_DATASETS_MODULES[name]
-
-    if dataset and path in _PACKAGED_DATASETS_MODULES:
-        output = prepare_packaged_module(path)
-        if return_resolved_file_path:
-            output += (None,)
-        if return_associated_base_path:
-            output += (None,)
-        if return_namespace:
-            output += (None,)
-        return output
-
-    # otherwise the module is added to the dynamic modules
-    dynamic_modules_path = dynamic_modules_path if dynamic_modules_path else init_dynamic_modules()
-    module_name_for_dynamic_modules = os.path.basename(dynamic_modules_path)
-    datasets_modules_path = os.path.join(dynamic_modules_path, "datasets")
-    datasets_modules_name = module_name_for_dynamic_modules + ".datasets"
-    metrics_modules_path = os.path.join(dynamic_modules_path, "metrics")
-    metrics_modules_name = module_name_for_dynamic_modules + ".metrics"
-
-    if force_local_path is None:
-        main_folder_path = os.path.join(datasets_modules_path if dataset else metrics_modules_path, short_name)
-    else:
-        main_folder_path = force_local_path
-
-    # We have several ways to find the processing file:
+    # We have several ways to get a dataset builder:
+    #
+    # - if path is the name of a packaged dataset module
+    #   -> use the packaged module (json, csv, etc.)
+    #
     # - if os.path.join(path, name) is a local python file
     #   -> use the module from the python file
     # - if path is a local directory (but no python file)
     #   -> use a packaged module (csv, text etc.) based on content of the directory
+    #
     # - if path has no "/" and is a module on github (in /datasets or in /metrics)
     #   -> use the module from the python file on github
     # - if path has one "/" and is dataset repository on the HF hub with a python file
     #   -> the module from the python file in the dataset repository
     # - if path has one "/" and is dataset repository on the HF hub without a python file
     #   -> use a packaged module (csv, text etc.) based on content of the repository
-    resource_type = "dataset" if dataset else "metric"
-    combined_path = os.path.join(path, name)
-    if path.endswith(name):
+
+    # Try packaged
+    if path in _PACKAGED_DATASETS_MODULES:
+        return PackagedDatasetModuleFactory(
+            path, data_files=data_files, download_config=download_config, download_mode=download_mode
+        ).get_module()
+    # Try locally
+    elif path.endswith(filename):
         if os.path.isfile(path):
-            file_path = path
-            local_path = path
-            base_path = os.path.dirname(path)
-            namespace = None
+            return LocalDatasetModuleFactoryWithScript(
+                path, download_mode=download_mode, dynamic_modules_path=dynamic_modules_path
+            ).get_module()
         else:
-            raise FileNotFoundError(f"Couldn't find a {resource_type} script at {relative_to_absolute_path(path)}")
+            raise FileNotFoundError(f"Couldn't find a dataset script at {relative_to_absolute_path(path)}")
     elif os.path.isfile(combined_path):
-        file_path = combined_path
-        local_path = combined_path
-        base_path = path
-        namespace = None
-    elif os.path.isfile(path):
-        file_path = path
-        local_path = path
-        base_path = os.path.dirname(path)
-        namespace = None
+        return LocalDatasetModuleFactoryWithScript(
+            combined_path, download_mode=download_mode, dynamic_modules_path=dynamic_modules_path
+        ).get_module()
     elif os.path.isdir(path):
-        resolved_data_files = _resolve_data_files_locally_or_by_urls(
-            path, data_files or "*", allowed_extensions=_EXTENSION_TO_MODULE.keys()
-        )
-        infered_module_name = _infer_module_for_data_files(resolved_data_files)
-        if not infered_module_name:
-            raise FileNotFoundError(f"No data files or {resource_type} script found in local directory {path}")
-        output = prepare_packaged_module(infered_module_name)
-        if return_resolved_file_path:
-            output += (None,)
-        if return_associated_base_path:
-            output += (path,)
-        if return_namespace:
-            output += (None,)
-        return output
-    else:
-        # Try github (canonical datasets/metrics) and then HF Hub (community datasets)
-        combined_path_abs = relative_to_absolute_path(combined_path)
-        expected_dir_for_combined_path_abs = os.path.dirname(combined_path_abs)
+        return LocalDatasetModuleFactoryWithoutScript(
+            path, data_files=data_files, download_mode=download_mode
+        ).get_module()
+    # Try remotely
+    elif is_relative_path(path) and path.count("/") <= 1 and not force_local_path:
         try:
-            try:
-                head_hf_s3(path, filename=name, dataset=dataset, max_retries=download_config.max_retries)
-            except Exception:
-                pass
-            revision = str(revision) if revision is not None else None
+            _raise_if_offline_mode_is_enabled()
             if path.count("/") == 0:  # canonical datasets/metrics: github path
-                file_path = hf_github_url(path=path, name=name, dataset=dataset, revision=revision)
-                namespace = None
-                try:
-                    local_path = cached_path(file_path, download_config=download_config)
-                except FileNotFoundError:
-                    if revision is not None:
-                        raise FileNotFoundError(
-                            f"Couldn't find a directory or a {resource_type} named '{path}' using version {revision}. "
-                            f"It doesn't exist locally at {expected_dir_for_combined_path_abs} or remotely at {file_path}"
-                        ) from None
-                    else:
-                        github_file_path = file_path
-                        file_path = hf_github_url(path=path, name=name, dataset=dataset, revision="master")
-                        try:
-                            local_path = cached_path(file_path, download_config=download_config)
-                            logger.warning(
-                                f"Couldn't find a directory or a {resource_type} named '{path}'. "
-                                f"It was picked from the master branch on github instead at {file_path}"
-                            )
-                        except FileNotFoundError:
-                            raise FileNotFoundError(
-                                f"Couldn't find a directory or a {resource_type} named '{path}'. "
-                                f"It doesn't exist locally at {expected_dir_for_combined_path_abs} or remotely at {github_file_path}"
-                            ) from None
+                return CanonicalDatasetModuleFactory(
+                    path,
+                    revision=revision,
+                    download_config=download_config,
+                    download_mode=download_mode,
+                    dynamic_modules_path=dynamic_modules_path,
+                ).get_module()
             elif path.count("/") == 1:  # users datasets/metrics: s3 path (hub for datasets and s3 for metrics)
-                file_path = hf_hub_url(path=path, name=name, revision=revision)
-                namespace = path[: path.index("/")]
-                if force_local_path is None:
-                    main_folder_path = os.path.join(
-                        datasets_modules_path if dataset else metrics_modules_path, f"{namespace}___{short_name}"
-                    )  # we have to use three underscores as in DatasetBuilder._relative_data_dir
-                if not dataset:
-                    # We don't have community metrics on the HF Hub
-                    raise FileNotFoundError(
-                        f"Couldn't find a {resource_type} in a directory at '{path}'. "
-                        f"It doesn't exist locally at {combined_path_abs}"
-                    )
+                hf_api = HfApi(config.HF_ENDPOINT)
                 try:
-                    local_path = cached_path(file_path, download_config=download_config)
-                except FileNotFoundError:
-                    hf_api = HfApi(config.HF_ENDPOINT)
-                    try:
-                        dataset_info = hf_api.dataset_info(
-                            repo_id=path, revision=revision, token=download_config.use_auth_token
-                        )
-                    except Exception as exc:
-                        raise FileNotFoundError(
-                            f"Couldn't find a directory or a {resource_type} named '{path}'. "
-                            f"It doesn't exist locally at {expected_dir_for_combined_path_abs} or remotely on {hf_api.endpoint}/datasets"
-                        ) from exc
-                    resolved_data_files = _resolve_data_files_in_dataset_repository(
-                        dataset_info,
-                        data_files if data_files is not None else "*",
-                        allowed_extensions=_EXTENSION_TO_MODULE.keys(),
+                    dataset_info = hf_api.dataset_info(
+                        repo_id=path,
+                        revision=revision,
+                        token=download_config.use_auth_token,
+                        timeout=100.0,
                     )
-                    infered_module_name = _infer_module_for_data_files(resolved_data_files)
-                    if not infered_module_name:
-                        raise FileNotFoundError(
-                            f"No data files found in dataset repository '{path}'. Local directory at {expected_dir_for_combined_path_abs} doesn't exist either."
-                        ) from None
-                    output = prepare_packaged_module(infered_module_name)
-                    if return_resolved_file_path:
-                        output += (None,)
-                    if return_associated_base_path:
-                        output += (url_or_path_parent(file_path),)
-                    if return_namespace:
-                        output += (namespace,)
-                    return output
-            else:
-                raise FileNotFoundError(
-                    f"Couldn't find a {resource_type} directory at '{path}'. "
-                    f"It doesn't exist locally at {expected_dir_for_combined_path_abs}"
-                )
-        except Exception as e:  # noqa: all the attempts failed, before raising the error we should check if the module already exists.
-            if os.path.isdir(main_folder_path):
-                hashes = [h for h in os.listdir(main_folder_path) if len(h) == 64]
-                if hashes:
-                    # get most recent
-                    def _get_modification_time(module_hash):
-                        return (Path(main_folder_path) / module_hash / name).stat().st_mtime
-
-                    hash = sorted(hashes, key=_get_modification_time)[-1]
-                    if namespace is None:
-                        module_path = ".".join(
-                            [datasets_modules_name if dataset else metrics_modules_name, short_name, hash, short_name]
-                        )
+                except Exception as e:  # noqa: catch any exception of hf_hub and consider that the dataset doesn't exist
+                    if isinstance(
+                        e,
+                        (
+                            OfflineModeIsEnabled,
+                            requests.exceptions.ConnectTimeout,
+                            requests.exceptions.ConnectionError,
+                        ),
+                    ):
+                        raise ConnectionError(f"Couldn't reach '{path}' on the Hub ({type(e).__name__})")
+                    elif "404" in str(e):
+                        msg = f"Dataset '{path}' doesn't exist on the Hub"
+                        raise FileNotFoundError(msg + f" at revision '{revision}'" if revision else msg)
                     else:
-                        module_path = ".".join(
-                            [
-                                datasets_modules_name if dataset else metrics_modules_name,
-                                f"{namespace}___{short_name}",
-                                hash,
-                                short_name,
-                            ]
-                        )
-                    logger.warning(
-                        f"Using the latest cached version of the module from {os.path.join(main_folder_path, hash)} "
-                        f"(last modified on {time.ctime(_get_modification_time(hash))}) since it "
-                        f"couldn't be found locally at {combined_path_abs}, or remotely ({type(e).__name__})."
-                    )
-                    output = (module_path, hash)
-                    if return_resolved_file_path:
-                        with open(os.path.join(main_folder_path, hash, short_name + ".json")) as cache_metadata:
-                            file_path = json.load(cache_metadata)["original file path"]
-                        output += (file_path,)
-                    if return_associated_base_path:
-                        output += (url_or_path_parent(file_path),)
-                    if return_namespace:
-                        output += (namespace,)
-                    return output
-            raise
-
-    # Load the module in two steps:
-    # 1. get the processing file on the local filesystem if it's not there (download to cache dir)
-    # 2. copy from the local file system inside the modules cache to import it
-
-    base_path = url_or_path_parent(file_path)  # remove the filename
-    dataset_infos = url_or_path_join(base_path, config.DATASETDICT_INFOS_FILENAME)
-
-    # Download the dataset infos file if available
-    try:
-        local_dataset_infos_path = cached_path(
-            dataset_infos,
-            download_config=download_config,
-        )
-    except (FileNotFoundError, ConnectionError):
-        local_dataset_infos_path = None
-
-    # Download external imports if needed
-    imports = get_imports(local_path)
-    local_imports = []
-    library_imports = []
-    for import_type, import_name, import_path, sub_directory in imports:
-        if import_type == "library":
-            library_imports.append((import_name, import_path))  # Import from a library
-            continue
-
-        if import_name == short_name:
-            raise ValueError(
-                f"Error in {module_type} script at {file_path}, importing relative {import_name} module "
-                f"but {import_name} is the name of the {module_type} script. "
-                f"Please change relative import {import_name} to another name and add a '# From: URL_OR_PATH' "
-                f"comment pointing to the original relative import file path."
-            )
-        if import_type == "internal":
-            url_or_filename = url_or_path_join(base_path, import_path + ".py")
-        elif import_type == "external":
-            url_or_filename = import_path
-        else:
-            raise ValueError("Wrong import_type")
-
-        local_import_path = cached_path(
-            url_or_filename,
-            download_config=download_config,
-        )
-        if sub_directory is not None:
-            local_import_path = os.path.join(local_import_path, sub_directory)
-        local_imports.append((import_name, local_import_path))
-
-    # Check library imports
-    needs_to_be_installed = []
-    for library_import_name, library_import_path in library_imports:
-        try:
-            lib = importlib.import_module(library_import_name)  # noqa F841
-        except ImportError:
-            needs_to_be_installed.append((library_import_name, library_import_path))
-    if needs_to_be_installed:
-        raise ImportError(
-            f"To be able to use this {module_type}, you need to install the following dependencies"
-            f"{[lib_name for lib_name, lib_path in needs_to_be_installed]} using 'pip install "
-            f"{' '.join([lib_path for lib_name, lib_path in needs_to_be_installed])}' for instance'"
-        )
-
-    # Define a directory with a unique name in our dataset or metric folder
-    # path is: ./datasets|metrics/dataset|metric_name/hash_from_code/script.py
-    # we use a hash to be able to have multiple versions of a dataset/metric processing file together
-    hash = files_to_hash([local_path] + [loc[1] for loc in local_imports])
-
-    if force_local_path is None:
-        hash_folder_path = os.path.join(main_folder_path, hash)
+                        raise e
+                if filename in [sibling.rfilename for sibling in dataset_info.siblings]:
+                    return CommunityDatasetModuleFactoryWithScript(
+                        path,
+                        revision=revision,
+                        download_config=download_config,
+                        download_mode=download_mode,
+                        dynamic_modules_path=dynamic_modules_path,
+                    ).get_module()
+                else:
+                    return CommunityDatasetModuleFactoryWithoutScript(
+                        path,
+                        revision=revision,
+                        data_files=data_files,
+                        download_config=download_config,
+                        download_mode=download_mode,
+                    ).get_module()
+        except Exception as e1:  # noqa: all the attempts failed, before raising the error we should check if the module is already cached.
+            try:
+                return CachedDatasetModuleFactory(path, dynamic_modules_path=dynamic_modules_path).get_module()
+            except Exception as e2:  # noqa: if it's not in the cache, then it doesn't exist.
+                if isinstance(e1, OfflineModeIsEnabled):
+                    raise ConnectionError(f"Couln't reach the Hugging Face Hub for dataset '{path}': {e1}") from None
+                if isinstance(e1, FileNotFoundError):
+                    raise FileNotFoundError(
+                        f"Couldn't find a dataset script at {relative_to_absolute_path(combined_path)} or any data file in the same directory. "
+                        f"Couldn't find '{path}' on the Hugging Face Hub either: {type(e1).__name__}: {e1}"
+                    ) from None
+                raise e1 from None
     else:
-        hash_folder_path = force_local_path
-
-    local_file_path = os.path.join(hash_folder_path, name)
-    dataset_infos_path = os.path.join(hash_folder_path, config.DATASETDICT_INFOS_FILENAME)
-
-    # Create the lock file where we know we have write permissions.
-    lock_path = (datasets_modules_path if dataset else metrics_modules_path) + f"{short_name}.lock"
-
-    # Prevent parallel disk operations
-    with FileLock(lock_path):
-        # Create main dataset/metrics folder if needed
-        if download_mode == GenerateMode.FORCE_REDOWNLOAD and os.path.exists(main_folder_path):
-            shutil.rmtree(main_folder_path)
+        raise FileNotFoundError(
+            f"Couldn't find a dataset script at {relative_to_absolute_path(combined_path)} or any data file in the same directory."
+        )
 
-        if not os.path.exists(main_folder_path):
-            logger.info(f"Creating main folder for {module_type} {file_path} at {main_folder_path}")
-            os.makedirs(main_folder_path, exist_ok=True)
-        else:
-            logger.info(f"Found main folder for {module_type} {file_path} at {main_folder_path}")
 
-        # add an __init__ file to the main dataset folder if needed
-        init_file_path = os.path.join(main_folder_path, "__init__.py")
-        if not os.path.exists(init_file_path):
-            with open(init_file_path, "w"):
-                pass
+def metric_module_factory(
+    path: str,
+    revision: Optional[Union[str, Version]] = None,
+    download_config: Optional[DownloadConfig] = None,
+    download_mode: Optional[GenerateMode] = None,
+    force_local_path: Optional[str] = None,
+    dynamic_modules_path: Optional[str] = None,
+    **download_kwargs,
+) -> MetricModuleFactoryResult:
+    r"""
+    Download/extract/cache a metric module.
 
-        # Create hash dataset folder if needed
-        if not os.path.exists(hash_folder_path):
-            logger.info(f"Creating specific version folder for {module_type} {file_path} at {hash_folder_path}")
-            os.makedirs(hash_folder_path)
-        else:
-            logger.info(f"Found specific version folder for {module_type} {file_path} at {hash_folder_path}")
+    Metrics codes are cached inside the the dynamic modules cache to allow easy import (avoid ugly sys.path tweaks).
 
-        # add an __init__ file to the hash dataset folder if needed
-        init_file_path = os.path.join(hash_folder_path, "__init__.py")
-        if not os.path.exists(init_file_path):
-            with open(init_file_path, "w"):
-                pass
+    Args:
 
-        # Copy dataset.py file in hash folder if needed
-        if not os.path.exists(local_file_path):
-            logger.info("Copying script file from %s to %s", file_path, local_file_path)
-            shutil.copyfile(local_path, local_file_path)
-        else:
-            logger.info("Found script file from %s to %s", file_path, local_file_path)
+        path (str): Path or name of the metric script.
 
-        # Copy dataset infos file if needed
-        if not os.path.exists(dataset_infos_path):
-            if local_dataset_infos_path is not None:
-                logger.info("Copying dataset infos file from %s to %s", dataset_infos, dataset_infos_path)
-                shutil.copyfile(local_dataset_infos_path, dataset_infos_path)
-            else:
-                logger.info("Couldn't find dataset infos file at %s", dataset_infos)
-        else:
-            if local_dataset_infos_path is not None and not filecmp.cmp(local_dataset_infos_path, dataset_infos_path):
-                logger.info("Updating dataset infos file from %s to %s", dataset_infos, dataset_infos_path)
-                shutil.copyfile(local_dataset_infos_path, dataset_infos_path)
-            else:
-                logger.info("Found dataset infos file from %s to %s", dataset_infos, dataset_infos_path)
+            - if ``path`` is a local metric script or a directory containing a local metric script (if the script has the same name as the directory):
+              -> load the module from the metric script
+              e.g. ``'./metrics/accuracy'`` or ``'./metrics/accuracy/accuracy.py'``.
+            - if ``path`` is a canonical metric (ex: `glue`, `squad`)
+              -> load the module from the metric script in the github repository at huggingface/datasets
+              e.g. ``'accuracy'`` or ``'rouge'``.
 
-        # Record metadata associating original dataset path with local unique folder
-        meta_path = local_file_path.split(".py")[0] + ".json"
-        if not os.path.exists(meta_path):
-            logger.info(f"Creating metadata file for {module_type} {file_path} at {meta_path}")
-            meta = {"original file path": file_path, "local file path": local_file_path}
-            # the filename is *.py in our case, so better rename to filenam.json instead of filename.py.json
-            with open(meta_path, "w", encoding="utf-8") as meta_file:
-                json.dump(meta, meta_file)
-        else:
-            logger.info(f"Found metadata file for {module_type} {file_path} at {meta_path}")
+        revision (Optional ``Union[str, datasets.Version]``):
+            If specified, the module will be loaded from the datasets repository at this version.
+            By default:
+            - it is set to the local version of the lib.
+            - it will also try to load it from the master branch if it's not available at the local version of the lib.
+            Specifying a version that is different from your local version of the lib might cause compatibility issues.
+        download_config (:class:`DownloadConfig`, optional): Specific download configuration parameters.
+        download_mode (:class:`GenerateMode`, default ``REUSE_DATASET_IF_EXISTS``): Download/generate mode.
+        force_local_path (Optional str): Optional path to a local path to download and prepare the script to.
+            Used to inspect or modify the script folder.
+        dynamic_modules_path (Optional str, defaults to HF_MODULES_CACHE / "datasets_modules", i.e. ~/.cache/huggingface/modules/datasets_modules):
+            Optional path to the directory in which the dynamic modules are saved. It must have been initialized with :obj:`init_dynamic_modules`.
+            By default the datasets and metrics are stored inside the `datasets_modules` module.
+        script_version:
+            .. deprecated:: 1.13
+                'script_version' was renamed to 'revision' in version 1.13 and will be removed in 1.15.
+        download_kwargs: optional attributes for DownloadConfig() which will override the attributes in download_config if supplied.
 
-        # Copy all the additional imports
-        for import_name, import_path in local_imports:
-            if os.path.isfile(import_path):
-                full_path_local_import = os.path.join(hash_folder_path, import_name + ".py")
-                if not os.path.exists(full_path_local_import):
-                    logger.info("Copying local import file from %s at %s", import_path, full_path_local_import)
-                    shutil.copyfile(import_path, full_path_local_import)
-                else:
-                    logger.info("Found local import file from %s at %s", import_path, full_path_local_import)
-            elif os.path.isdir(import_path):
-                full_path_local_import = os.path.join(hash_folder_path, import_name)
-                if not os.path.exists(full_path_local_import):
-                    logger.info("Copying local import directory from %s at %s", import_path, full_path_local_import)
-                    shutil.copytree(import_path, full_path_local_import)
-                else:
-                    logger.info("Found local import directory from %s at %s", import_path, full_path_local_import)
-            else:
-                raise OSError(f"Error with local import at {import_path}")
+    Returns:
+        MetricModuleFactoryResult
+    """
+    if download_config is None:
+        download_config = DownloadConfig(**download_kwargs)
+    download_config.extract_compressed_file = True
+    download_config.force_extract = True
 
-    if force_local_path is None:
-        if namespace is None:
-            module_path = ".".join(
-                [datasets_modules_name if dataset else metrics_modules_name, short_name, hash, short_name]
-            )
+    filename = list(filter(lambda x: x, path.replace(os.sep, "/").split("/")))[-1]
+    if not filename.endswith(".py"):
+        filename = filename + ".py"
+    combined_path = os.path.join(path, filename)
+    # Try locally
+    if path.endswith(filename):
+        if os.path.isfile(path):
+            return LocalMetricModuleFactory(
+                path, download_mode=download_mode, dynamic_modules_path=dynamic_modules_path
+            ).get_module()
         else:
-            module_path = ".".join(
-                [
-                    datasets_modules_name if dataset else metrics_modules_name,
-                    f"{namespace}___{short_name}",
-                    hash,
-                    short_name,
-                ]
-            )
+            raise FileNotFoundError(f"Couldn't find a metric script at {relative_to_absolute_path(path)}")
+    elif os.path.isfile(combined_path):
+        return LocalMetricModuleFactory(
+            combined_path, download_mode=download_mode, dynamic_modules_path=dynamic_modules_path
+        ).get_module()
+    elif is_relative_path(path) and path.count("/") == 0 and not force_local_path:
+        try:
+            return CanonicalMetricModuleFactory(
+                path,
+                revision=revision,
+                download_config=download_config,
+                download_mode=download_mode,
+                dynamic_modules_path=dynamic_modules_path,
+            ).get_module()
+        except Exception as e1:  # noqa: all the attempts failed, before raising the error we should check if the module is already cached.
+            try:
+                return CachedMetricModuleFactory(path, dynamic_modules_path=dynamic_modules_path).get_module()
+            except Exception as e2:  # noqa: if it's not in the cache, then it doesn't exist.
+                if not isinstance(e1, FileNotFoundError):
+                    raise e1 from None
+                raise FileNotFoundError(
+                    f"Couldn't find a metric script at {relative_to_absolute_path(combined_path)}. "
+                    f"Metric '{path}' doesn't exist on the Hugging Face Hub either."
+                ) from None
     else:
-        module_path = local_file_path
+        raise FileNotFoundError(f"Couldn't find a metric script at {relative_to_absolute_path(combined_path)}.")
 
-    # make the new module to be noticed by the import system
-    importlib.invalidate_caches()
 
-    output = (module_path, hash)
-    if return_resolved_file_path:
-        output += (file_path,)
-    if return_associated_base_path:
-        output += (base_path,)
-    if return_namespace:
-        output += (namespace,)
-    return output
+def prepare_module(
+    path: str,
+    revision: Optional[Union[str, Version]] = None,
+    download_config: Optional[DownloadConfig] = None,
+    download_mode: Optional[GenerateMode] = None,
+    dataset: bool = True,
+    force_local_path: Optional[str] = None,
+    dynamic_modules_path: Optional[str] = None,
+    data_files: Optional[Union[Dict, List, str]] = None,
+    script_version="deprecated",
+    **download_kwargs,
+) -> Union[Tuple[str, str], Tuple[str, str, Optional[str]]]:
+    """For backward compatibility. Please use dataset_module_factory or metric_module_factory instead."""
+    if script_version != "deprecated":
+        warnings.warn(
+            "'script_version' was renamed to 'revision' in version 1.13 and will be removed in 1.15.", FutureWarning
+        )
+        revision = script_version
+    if dataset:
+        results = dataset_module_factory(
+            path,
+            revision=revision,
+            download_config=download_config,
+            download_mode=download_mode,
+            force_local_path=force_local_path,
+            dynamic_modules_path=dynamic_modules_path,
+            data_files=data_files,
+            **download_kwargs,
+        )
+        return results.module_path, results.hash
+    else:
+        results = metric_module_factory(
+            path,
+            revision=revision,
+            download_config=download_config,
+            download_mode=download_mode,
+            force_local_path=force_local_path,
+            dynamic_modules_path=dynamic_modules_path,
+            **download_kwargs,
+        )
+        return results.module_path, results.hash
 
 
 def load_metric(
@@ -874,14 +1322,10 @@ def load_metric(
             "'script_version' was renamed to 'revision' in version 1.13 and will be removed in 1.15.", FutureWarning
         )
         revision = script_version
-    module_path, _ = prepare_module(
-        path,
-        revision=revision,
-        download_config=download_config,
-        download_mode=download_mode,
-        dataset=False,
-    )
-    metric_cls = import_main_class(module_path, dataset=False)
+    metric_module = metric_module_factory(
+        path, revision=revision, download_config=download_config, download_mode=download_mode
+    ).module_path
+    metric_cls = import_main_class(metric_module, dataset=False)
     metric = metric_cls(
         config_name=config_name,
         process_id=process_id,
@@ -972,53 +1416,30 @@ def load_dataset_builder(
             "'script_version' was renamed to 'revision' in version 1.13 and will be removed in 1.15.", FutureWarning
         )
         revision = script_version
-    # Download/copy dataset processing script
-    module_path, hash, base_path, namespace = prepare_module(
-        path,
-        revision=revision,
-        download_config=download_config,
-        download_mode=download_mode,
-        dataset=True,
-        return_associated_base_path=True,
-        return_namespace=True,
-        use_auth_token=use_auth_token,
-        data_files=data_files,
+    if use_auth_token is not None:
+        download_config = download_config.copy() if download_config else DownloadConfig()
+        download_config.use_auth_token = use_auth_token
+
+    dataset_module_factory_result = dataset_module_factory(
+        path, revision=revision, download_config=download_config, download_mode=download_mode, data_files=data_files
     )
 
     # Get dataset builder class from the processing script
-    builder_cls = import_main_class(module_path, dataset=True)
-
-    # For packaged builder used to load data from a dataset repository or dataset directory (no dataset script)
-    if module_path.startswith("datasets.") and path not in _PACKAGED_DATASETS_MODULES:
-        # Add a nice name to the configuratiom
-        if name is None:
-            name = path.split("/")[-1].split(os.sep)[-1]
-        # Resolve the data files
-        allowed_extensions = [
-            extension
-            for extension in _EXTENSION_TO_MODULE
-            if _EXTENSION_TO_MODULE[extension] == camelcase_to_snakecase(builder_cls.__name__)
+    dataset_module = dataset_module_factory_result.module_path
+    builder_cls = import_main_class(dataset_module)
+    builder_kwargs = dataset_module_factory_result.builder_kwargs
+    data_files = builder_kwargs.pop("data_files", data_files)
+    name = builder_kwargs.pop("name", name)
+    hash = builder_kwargs.pop("hash")
+
+    if path in _PACKAGED_DATASETS_MODULES and data_files is None:
+        error_msg = f"Please specify the data files to load for the {path} dataset builder."
+        example_extensions = [
+            extension for extension in _EXTENSION_TO_MODULE if _EXTENSION_TO_MODULE[extension] == path
         ]
-        data_files = data_files if data_files is not None else "*"
-        if base_path.startswith(config.HF_ENDPOINT):
-            dataset_info = HfApi(config.HF_ENDPOINT).dataset_info(path, revision=revision, token=use_auth_token)
-            data_files = _resolve_data_files_in_dataset_repository(
-                dataset_info, data_files, allowed_extensions=allowed_extensions
-            )
-        else:  # local dir
-            data_files = _resolve_data_files_locally_or_by_urls(
-                path, data_files, allowed_extensions=allowed_extensions
-            )
-    elif path in _PACKAGED_DATASETS_MODULES:
-        if data_files is None:
-            error_msg = f"Please specify the data files to load for the {path} dataset builder."
-            example_extensions = [
-                extension for extension in _EXTENSION_TO_MODULE if _EXTENSION_TO_MODULE[extension] == path
-            ]
-            if example_extensions:
-                error_msg += f'\nFor example `data_files={{"train": "path/to/data/train/*.{example_extensions[0]}"}}`'
-            raise ValueError(error_msg)
-        data_files = _resolve_data_files_locally_or_by_urls(".", data_files)
+        if example_extensions:
+            error_msg += f'\nFor example `data_files={{"train": "path/to/data/train/*.{example_extensions[0]}"}}`'
+        raise ValueError(error_msg)
 
     # Instantiate the dataset builder
     builder_instance: DatasetBuilder = builder_cls(
@@ -1027,10 +1448,9 @@ def load_dataset_builder(
         data_dir=data_dir,
         data_files=data_files,
         hash=hash,
-        base_path=base_path,
         features=features,
         use_auth_token=use_auth_token,
-        namespace=namespace,
+        **builder_kwargs,
         **config_kwargs,
     )
 
diff --git a/src/datasets/utils/file_utils.py b/src/datasets/utils/file_utils.py
index a0aed6c5f62..fb9a2b3612c 100644
--- a/src/datasets/utils/file_utils.py
+++ b/src/datasets/utils/file_utils.py
@@ -20,19 +20,17 @@
 from functools import partial
 from hashlib import sha256
 from pathlib import Path
-from typing import Dict, List, Optional, TypeVar, Union
+from typing import Dict, Optional, TypeVar, Union
 from urllib.parse import urlparse
 
 import numpy as np
 import posixpath
 import requests
-from tqdm.contrib.concurrent import thread_map
 
 from .. import __version__, config, utils
 from . import logging
 from .extract import ExtractManager
 from .filelock import FileLock
-from .tqdm_utils import tqdm
 
 
 logger = logging.get_logger(__name__)  # pylint: disable=invalid-name
@@ -492,24 +490,6 @@ def request_etag(url: str, use_auth_token: Optional[Union[str, bool]] = None) ->
     return etag
 
 
-def request_etags(
-    urls: List[str],
-    use_auth_token: Optional[Union[str, bool]] = None,
-    max_workers=64,
-    tqdm_kwargs: Optional[dict] = None,
-) -> List[Optional[str]]:
-    tqdm_kwargs = tqdm_kwargs if tqdm_kwargs is not None else {}
-    tqdm_kwargs["desc"] = tqdm_kwargs.get("desc", "Get ETags")
-    tqdm_kwargs["disable"] = tqdm_kwargs.get("disable", len(urls) <= 16 or logging.get_verbosity() == logging.NOTSET)
-    return thread_map(
-        partial(request_etag, use_auth_token=use_auth_token),
-        urls,
-        max_workers=max_workers,
-        tqdm_class=tqdm,
-        **tqdm_kwargs,
-    )
-
-
 def get_from_cache(
     url,
     cache_dir=None,
diff --git a/tests/conftest.py b/tests/conftest.py
index acfdf147d28..aecd7cb07e4 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -31,6 +31,12 @@ def set_test_cache_config(tmp_path_factory, monkeypatch):
     monkeypatch.setattr("datasets.config.EXTRACTED_DATASETS_PATH", str(test_extracted_datasets_path))
 
 
+@pytest.fixture(autouse=True)
+def set_update_download_counts_to_false(monkeypatch):
+    # don't take tests into account when counting downloads
+    monkeypatch.setattr("datasets.config.HF_UPDATE_DOWNLOAD_COUNTS", False)
+
+
 FILE_CONTENT = """\
     Text data.
     Second line of data."""
diff --git a/tests/test_data_files.py b/tests/test_data_files.py
new file mode 100644
index 00000000000..333725ba94a
--- /dev/null
+++ b/tests/test_data_files.py
@@ -0,0 +1,231 @@
+import os
+from pathlib import Path
+from unittest.mock import patch
+
+import pytest
+from huggingface_hub.hf_api import DatasetInfo
+
+from datasets.data_files import (
+    DataFilesDict,
+    DataFilesList,
+    Url,
+    resolve_patterns_in_dataset_repository,
+    resolve_patterns_locally_or_by_urls,
+)
+from datasets.fingerprint import Hasher
+from datasets.utils.file_utils import hf_hub_url
+
+
+_TEST_PATTERNS = ["*", "**/*", "*.txt", "data/*", "**/*.txt", "**/train.txt"]
+_FILES_TO_IGNORE = {".dummy", "README.md", "dummy_data.zip", "dataset_infos.json"}
+_TEST_PATTERNS_SIZES = dict([("*", 2), ("**/*", 2), ("*.txt", 2), ("data/*", 2), ("**/*.txt", 2), ("**/train.txt", 1)])
+
+_TEST_URL = "https://raw.githubusercontent.com/huggingface/datasets/9675a5a1e7b99a86f9c250f6ea5fa5d1e6d5cc7d/setup.py"
+
+
+@pytest.fixture
+def complex_data_dir(tmp_path):
+    data_dir = tmp_path / "complex_data_dir"
+    data_dir.mkdir()
+    (data_dir / "data").mkdir()
+    with open(data_dir / "data" / "train.txt", "w") as f:
+        f.write("foo\n" * 10)
+    with open(data_dir / "data" / "test.txt", "w") as f:
+        f.write("bar\n" * 10)
+    with open(data_dir / "README.md", "w") as f:
+        f.write("This is a readme")
+    with open(data_dir / ".dummy", "w") as f:
+        f.write("this is a dummy file that is not a data file")
+    return str(data_dir)
+
+
+@pytest.fixture
+def pattern_results(complex_data_dir):
+    return {
+        pattern: sorted(
+            [
+                str(path)
+                for path in Path(complex_data_dir).rglob(pattern)
+                if path.name not in _FILES_TO_IGNORE and path.is_file()
+            ]
+        )
+        for pattern in _TEST_PATTERNS
+    }
+
+
+@pytest.fixture
+def hub_dataset_info(complex_data_dir):
+    return DatasetInfo(
+        siblings=[
+            {"rfilename": path.relative_to(complex_data_dir).as_posix()}
+            for path in Path(complex_data_dir).rglob("*")
+            if path.is_file()
+        ],
+        sha="foobarfoobar",
+        id="foo",
+    )
+
+
+@pytest.fixture
+def hub_dataset_info_patterns_results(hub_dataset_info, complex_data_dir, pattern_results):
+    return {
+        pattern: [
+            hf_hub_url(
+                hub_dataset_info.id, Path(path).relative_to(complex_data_dir).as_posix(), revision=hub_dataset_info.sha
+            )
+            for path in pattern_results[pattern]
+        ]
+        for pattern in pattern_results
+    }
+
+
+@pytest.mark.parametrize("pattern", _TEST_PATTERNS)
+def test_pattern_results_fixture(pattern_results, pattern):
+    assert len(pattern_results[pattern]) == _TEST_PATTERNS_SIZES[pattern]
+    assert all(Path(path).is_file() for path in pattern_results[pattern])
+
+
+@pytest.mark.parametrize("pattern", _TEST_PATTERNS)
+def test_resolve_patterns_locally_or_by_urls(complex_data_dir, pattern, pattern_results):
+    resolved_data_files = resolve_patterns_locally_or_by_urls(complex_data_dir, [pattern])
+    assert sorted(str(f) for f in resolved_data_files) == pattern_results[pattern]
+    assert all(isinstance(path, Path) for path in resolved_data_files)
+
+
+def test_resolve_patterns_locally_or_by_urls_with_absolute_path(tmp_path, complex_data_dir):
+    abs_path = os.path.join(complex_data_dir, "data", "train.txt")
+    resolved_data_files = resolve_patterns_locally_or_by_urls(str(tmp_path / "blabla"), [abs_path])
+    assert len(resolved_data_files) == 1
+
+
+@pytest.mark.parametrize("pattern,size,extensions", [("*", 2, ["txt"]), ("*", 2, None), ("*", 0, ["blablabla"])])
+def test_resolve_patterns_locally_or_by_urls_with_extensions(complex_data_dir, pattern, size, extensions):
+    if size > 0:
+        resolved_data_files = resolve_patterns_locally_or_by_urls(
+            complex_data_dir, [pattern], allowed_extensions=extensions
+        )
+        assert len(resolved_data_files) == size
+    else:
+        with pytest.raises(FileNotFoundError):
+            resolve_patterns_locally_or_by_urls(complex_data_dir, pattern, allowed_extensions=extensions)
+
+
+def test_fail_resolve_patterns_locally_or_by_urls(complex_data_dir):
+    with pytest.raises(FileNotFoundError):
+        resolve_patterns_locally_or_by_urls(complex_data_dir, ["blablabla"])
+
+
+@pytest.mark.parametrize("pattern", _TEST_PATTERNS)
+def test_resolve_patterns_in_dataset_repository(hub_dataset_info, pattern, hub_dataset_info_patterns_results):
+    resolved_data_files = resolve_patterns_in_dataset_repository(hub_dataset_info, [pattern])
+    assert sorted(str(f) for f in resolved_data_files) == hub_dataset_info_patterns_results[pattern]
+    assert all(isinstance(url, Url) for url in resolved_data_files)
+
+
+@pytest.mark.parametrize("pattern,size,extensions", [("*", 2, ["txt"]), ("*", 2, None), ("*", 0, ["blablabla"])])
+def test_resolve_patterns_in_dataset_repository_with_extensions(hub_dataset_info, pattern, size, extensions):
+    if size > 0:
+        resolved_data_files = resolve_patterns_in_dataset_repository(
+            hub_dataset_info, [pattern], allowed_extensions=extensions
+        )
+        assert len(resolved_data_files) == size
+    else:
+        with pytest.raises(FileNotFoundError):
+            resolved_data_files = resolve_patterns_in_dataset_repository(
+                hub_dataset_info, [pattern], allowed_extensions=extensions
+            )
+
+
+def test_fail_resolve_patterns_in_dataset_repository(hub_dataset_info):
+    with pytest.raises(FileNotFoundError):
+        resolve_patterns_in_dataset_repository(hub_dataset_info, "blablabla")
+
+
+@pytest.mark.parametrize("pattern", _TEST_PATTERNS)
+def test_DataFilesList_from_hf_repo(hub_dataset_info, hub_dataset_info_patterns_results, pattern):
+    data_files_list = DataFilesList.from_hf_repo([pattern], hub_dataset_info)
+    assert sorted(str(f) for f in data_files_list) == hub_dataset_info_patterns_results[pattern]
+    assert all(isinstance(url, Url) for url in data_files_list)
+    assert len(data_files_list.origin_metadata) > 0
+
+
+@pytest.mark.parametrize("pattern", _TEST_PATTERNS)
+def test_DataFilesList_from_local_or_remote(complex_data_dir, pattern_results, pattern):
+    data_files_list = DataFilesList.from_local_or_remote([pattern], complex_data_dir)
+    assert sorted(str(f) for f in data_files_list) == pattern_results[pattern]
+    assert all(isinstance(path, Path) for path in data_files_list)
+    assert len(data_files_list.origin_metadata) > 0
+
+
+def test_DataFilesList_from_local_or_remote_with_extra_files(complex_data_dir, text_file):
+    data_files_list = DataFilesList.from_local_or_remote([_TEST_URL, str(text_file)], complex_data_dir)
+    assert list(data_files_list) == [Url(_TEST_URL), Path(text_file)]
+    assert len(data_files_list.origin_metadata) == 2
+
+
+@pytest.mark.parametrize("pattern", _TEST_PATTERNS)
+def test_DataFilesDict_from_hf_repo(hub_dataset_info, hub_dataset_info_patterns_results, pattern):
+    split_name = "train"
+    data_files = DataFilesDict.from_hf_repo({split_name: [pattern]}, hub_dataset_info)
+    assert all(isinstance(data_files_list, DataFilesList) for data_files_list in data_files.values())
+    assert sorted(str(f) for f in data_files[split_name]) == hub_dataset_info_patterns_results[pattern]
+    assert all(isinstance(url, Url) for url in data_files[split_name])
+
+
+@pytest.mark.parametrize("pattern", _TEST_PATTERNS)
+def test_DataFilesDict_from_local_or_remote(complex_data_dir, pattern_results, pattern):
+    split_name = "train"
+    data_files = DataFilesDict.from_local_or_remote({split_name: [pattern]}, complex_data_dir)
+    assert all(isinstance(data_files_list, DataFilesList) for data_files_list in data_files.values())
+    assert sorted(str(f) for f in data_files[split_name]) == pattern_results[pattern]
+    assert all(isinstance(url, Path) for url in data_files[split_name])
+
+
+def test_DataFilesDict_from_hf_repo_hashing(hub_dataset_info):
+    patterns = {"train": ["**/train.txt"], "test": ["**/test.txt"]}
+    data_files1 = DataFilesDict.from_hf_repo(patterns, hub_dataset_info)
+    data_files2 = DataFilesDict.from_hf_repo(patterns, hub_dataset_info)
+    assert Hasher.hash(data_files1) == Hasher.hash(data_files2)
+
+    data_files2 = DataFilesDict(sorted(data_files1.items(), reverse=True))
+    assert Hasher.hash(data_files1) == Hasher.hash(data_files2)
+
+    patterns2 = {"train": ["data/train.txt"], "test": ["data/test.txt"]}
+    data_files2 = DataFilesDict.from_hf_repo(patterns2, hub_dataset_info)
+    assert Hasher.hash(data_files1) == Hasher.hash(data_files2)
+
+    patterns2 = {"train": ["data/train.txt"], "test": ["data/train.txt"]}
+    data_files2 = DataFilesDict.from_hf_repo(patterns2, hub_dataset_info)
+    assert Hasher.hash(data_files1) != Hasher.hash(data_files2)
+
+    with patch.object(hub_dataset_info, "id", "blabla"):
+        data_files2 = DataFilesDict.from_hf_repo(patterns, hub_dataset_info)
+        assert Hasher.hash(data_files1) != Hasher.hash(data_files2)
+
+    with patch.object(hub_dataset_info, "sha", "blabla"):
+        data_files2 = DataFilesDict.from_hf_repo(patterns, hub_dataset_info)
+        assert Hasher.hash(data_files1) != Hasher.hash(data_files2)
+
+
+def test_DataFilesDict_from_hf_local_or_remote_hashing(text_file):
+    patterns = {"train": [_TEST_URL], "test": [str(text_file)]}
+    data_files1 = DataFilesDict.from_local_or_remote(patterns)
+    data_files2 = DataFilesDict.from_local_or_remote(patterns)
+    assert Hasher.hash(data_files1) == Hasher.hash(data_files2)
+
+    data_files2 = DataFilesDict(sorted(data_files1.items(), reverse=True))
+    assert Hasher.hash(data_files1) == Hasher.hash(data_files2)
+
+    patterns2 = {"train": [_TEST_URL], "test": [_TEST_URL]}
+    data_files2 = DataFilesDict.from_local_or_remote(patterns2)
+    assert Hasher.hash(data_files1) != Hasher.hash(data_files2)
+
+    with patch("datasets.data_files.request_etag") as mock_request_etag:
+        mock_request_etag.return_value = "blabla"
+        data_files2 = DataFilesDict.from_local_or_remote(patterns)
+        assert Hasher.hash(data_files1) != Hasher.hash(data_files2)
+
+    with patch("datasets.data_files.os.path.getmtime") as mock_getmtime:
+        mock_getmtime.return_value = 123
+        data_files2 = DataFilesDict.from_local_or_remote(patterns)
+        assert Hasher.hash(data_files1) != Hasher.hash(data_files2)
diff --git a/tests/test_load.py b/tests/test_load.py
index a54a3929a4b..26c94c580de 100644
--- a/tests/test_load.py
+++ b/tests/test_load.py
@@ -6,24 +6,32 @@
 import time
 from functools import partial
 from hashlib import sha256
-from pathlib import Path, PurePath
+from pathlib import Path
 from unittest import TestCase
 from unittest.mock import patch
 
 import pytest
 import requests
-from huggingface_hub.hf_api import DatasetInfo
 
 import datasets
 from datasets import SCRIPTS_VERSION, load_dataset, load_from_disk
 from datasets.arrow_dataset import Dataset
 from datasets.builder import DatasetBuilder
+from datasets.data_files import DataFilesDict
 from datasets.dataset_dict import DatasetDict, IterableDatasetDict
 from datasets.features import Features, Value
 from datasets.iterable_dataset import IterableDataset
 from datasets.load import (
-    _resolve_data_files_in_dataset_repository,
-    _resolve_data_files_locally_or_by_urls,
+    CachedDatasetModuleFactory,
+    CachedMetricModuleFactory,
+    CanonicalDatasetModuleFactory,
+    CanonicalMetricModuleFactory,
+    CommunityDatasetModuleFactoryWithoutScript,
+    CommunityDatasetModuleFactoryWithScript,
+    LocalDatasetModuleFactoryWithoutScript,
+    LocalDatasetModuleFactoryWithScript,
+    LocalMetricModuleFactory,
+    PackagedDatasetModuleFactory,
     prepare_module,
 )
 from datasets.utils.file_utils import DownloadConfig, is_remote_url
@@ -63,12 +71,29 @@ def _generate_examples(self, filepath, **kwargs):
                 yield i, {"text": line.strip()}
 """
 
-SAMPLE_DATASET_IDENTIFIER = "lhoestq/test"
-SAMPLE_DATASET_IDENTIFIER2 = "lhoestq/test2"
+SAMPLE_DATASET_IDENTIFIER = "lhoestq/test"  # has dataset script
+SAMPLE_DATASET_IDENTIFIER2 = "lhoestq/test2"  # only has data files
 SAMPLE_NOT_EXISTING_DATASET_IDENTIFIER = "lhoestq/_dummy"
 SAMPLE_DATASET_NAME_THAT_DOESNT_EXIST = "_dummy"
 
 
+METRIC_LOADING_SCRIPT_NAME = "__dummy_metric1__"
+
+METRIC_LOADING_SCRIPT_CODE = """
+import datasets
+from datasets import MetricInfo, Features, Value
+
+
+class __DummyMetric1__(datasets.Metric):
+
+    def _info(self):
+        return MetricInfo(features=Features({"predictions": Value("int"), "references": Value("int")}))
+
+    def _compute(self, predictions, references):
+        return {"__dummy_metric1__": sum(int(p == r) for p, r in zip(predictions, references))}
+"""
+
+
 @pytest.fixture
 def data_dir(tmp_path):
     data_dir = tmp_path / "data_dir"
@@ -122,6 +147,125 @@ def dataset_loading_script_dir_readonly(tmp_path):
     return dataset_loading_script_dir
 
 
+@pytest.fixture
+def metric_loading_script_dir(tmp_path):
+    script_name = METRIC_LOADING_SCRIPT_NAME
+    script_dir = tmp_path / script_name
+    script_dir.mkdir()
+    script_path = script_dir / f"{script_name}.py"
+    with open(script_path, "w") as f:
+        f.write(METRIC_LOADING_SCRIPT_CODE)
+    return str(script_dir)
+
+
+class ModuleFactoryTest(TestCase):
+    @pytest.fixture(autouse=True)
+    def inject_fixtures(self, jsonl_path, data_dir, dataset_loading_script_dir, metric_loading_script_dir):
+        self._jsonl_path = jsonl_path
+        self._data_dir = data_dir
+        self._dataset_loading_script_dir = dataset_loading_script_dir
+        self._metric_loading_script_dir = metric_loading_script_dir
+
+    def setUp(self):
+        self.hf_modules_cache = tempfile.mkdtemp()
+        self.cache_dir = tempfile.mkdtemp()
+        self.download_config = DownloadConfig(cache_dir=self.cache_dir)
+        self.dynamic_modules_path = datasets.load.init_dynamic_modules(
+            name="test_datasets_modules_" + os.path.basename(self.hf_modules_cache),
+            hf_modules_cache=self.hf_modules_cache,
+        )
+
+    def test_CanonicalDatasetModuleFactory(self):
+        # "wmt_t2t" has additional imports (internal)
+        factory = CanonicalDatasetModuleFactory(
+            "wmt_t2t", download_config=self.download_config, dynamic_modules_path=self.dynamic_modules_path
+        )
+        module_factory_result = factory.get_module()
+        assert importlib.import_module(module_factory_result.module_path) is not None
+
+    def test_CanonicalMetricModuleFactory(self):
+        # "bleu" requires additional imports (external from github)
+        factory = CanonicalMetricModuleFactory(
+            "bleu", download_config=self.download_config, dynamic_modules_path=self.dynamic_modules_path
+        )
+        module_factory_result = factory.get_module()
+        assert importlib.import_module(module_factory_result.module_path) is not None
+
+    def test_LocalMetricModuleFactory(self):
+        path = os.path.join(self._metric_loading_script_dir, f"{METRIC_LOADING_SCRIPT_NAME}.py")
+        factory = LocalMetricModuleFactory(
+            path, download_config=self.download_config, dynamic_modules_path=self.dynamic_modules_path
+        )
+        module_factory_result = factory.get_module()
+        assert importlib.import_module(module_factory_result.module_path) is not None
+
+    def test_LocalDatasetModuleFactoryWithScript(self):
+        path = os.path.join(self._dataset_loading_script_dir, f"{DATASET_LOADING_SCRIPT_NAME}.py")
+        factory = LocalDatasetModuleFactoryWithScript(
+            path, download_config=self.download_config, dynamic_modules_path=self.dynamic_modules_path
+        )
+        module_factory_result = factory.get_module()
+        assert importlib.import_module(module_factory_result.module_path) is not None
+
+    def test_LocalDatasetModuleFactoryWithoutScript(self):
+        factory = LocalDatasetModuleFactoryWithoutScript(self._data_dir)
+        module_factory_result = factory.get_module()
+        assert importlib.import_module(module_factory_result.module_path) is not None
+
+    def test_PackagedDatasetModuleFactory(self):
+        factory = PackagedDatasetModuleFactory(
+            "json", data_files=self._jsonl_path, download_config=self.download_config
+        )
+        module_factory_result = factory.get_module()
+        assert importlib.import_module(module_factory_result.module_path) is not None
+
+    def test_CommunityDatasetModuleFactoryWithoutScript(self):
+        factory = CommunityDatasetModuleFactoryWithoutScript(
+            SAMPLE_DATASET_IDENTIFIER2, download_config=self.download_config
+        )
+        module_factory_result = factory.get_module()
+        assert importlib.import_module(module_factory_result.module_path) is not None
+
+    def test_CommunityDatasetModuleFactoryWithScript(self):
+        factory = CommunityDatasetModuleFactoryWithScript(
+            SAMPLE_DATASET_IDENTIFIER,
+            download_config=self.download_config,
+            dynamic_modules_path=self.dynamic_modules_path,
+        )
+        module_factory_result = factory.get_module()
+        assert importlib.import_module(module_factory_result.module_path) is not None
+
+    def test_CachedDatasetModuleFactory(self):
+        path = os.path.join(self._dataset_loading_script_dir, f"{DATASET_LOADING_SCRIPT_NAME}.py")
+        factory = LocalDatasetModuleFactoryWithScript(
+            path, download_config=self.download_config, dynamic_modules_path=self.dynamic_modules_path
+        )
+        module_factory_result = factory.get_module()
+        for offline_mode in OfflineSimulationMode:
+            with offline(offline_mode):
+                factory = CachedDatasetModuleFactory(
+                    DATASET_LOADING_SCRIPT_NAME,
+                    dynamic_modules_path=self.dynamic_modules_path,
+                )
+                module_factory_result = factory.get_module()
+                assert importlib.import_module(module_factory_result.module_path) is not None
+
+    def test_CachedMetricModuleFactory(self):
+        path = os.path.join(self._metric_loading_script_dir, f"{METRIC_LOADING_SCRIPT_NAME}.py")
+        factory = LocalMetricModuleFactory(
+            path, download_config=self.download_config, dynamic_modules_path=self.dynamic_modules_path
+        )
+        module_factory_result = factory.get_module()
+        for offline_mode in OfflineSimulationMode:
+            with offline(offline_mode):
+                factory = CachedMetricModuleFactory(
+                    METRIC_LOADING_SCRIPT_NAME,
+                    dynamic_modules_path=self.dynamic_modules_path,
+                )
+                module_factory_result = factory.get_module()
+                assert importlib.import_module(module_factory_result.module_path) is not None
+
+
 class LoadTest(TestCase):
     @pytest.fixture(autouse=True)
     def inject_fixtures(self, caplog):
@@ -130,7 +274,7 @@ def inject_fixtures(self, caplog):
     def setUp(self):
         self.hf_modules_cache = tempfile.mkdtemp()
         self.dynamic_modules_path = datasets.load.init_dynamic_modules(
-            name="test_datasets_modules", hf_modules_cache=self.hf_modules_cache
+            name="test_datasets_modules2", hf_modules_cache=self.hf_modules_cache
         )
 
     def tearDown(self):
@@ -160,10 +304,9 @@ def test_prepare_module(self):
             dummy_code = "MY_DUMMY_VARIABLE = 'general kenobi'"
             module_dir = self._dummy_module_dir(tmp_dir, "__dummy_module_name1__", dummy_code)
             module_path = os.path.join(module_dir, "__dummy_module_name1__.py")
-            importable_module_path, module_hash, resolved_file_path = datasets.load.prepare_module(
-                module_path, dynamic_modules_path=self.dynamic_modules_path, return_resolved_file_path=True
+            importable_module_path, module_hash = datasets.load.prepare_module(
+                module_path, dynamic_modules_path=self.dynamic_modules_path
             )
-            self.assertEqual(resolved_file_path, module_path)
             dummy_module = importlib.import_module(importable_module_path)
             self.assertEqual(dummy_module.MY_DUMMY_VARIABLE, "general kenobi")
             self.assertEqual(module_hash, sha256(dummy_code.encode("utf-8")).hexdigest())
@@ -218,10 +361,11 @@ def test_load_dataset_canonical(self):
             with offline(offline_simulation_mode):
                 with self.assertRaises(ConnectionError) as context:
                     datasets.load_dataset("_dummy")
-                self.assertIn(
-                    f"https://raw.githubusercontent.com/huggingface/datasets/{scripts_version}/datasets/_dummy/_dummy.py",
-                    str(context.exception),
-                )
+                if offline_simulation_mode != OfflineSimulationMode.HF_DATASETS_OFFLINE_SET_TO_1:
+                    self.assertIn(
+                        f"https://raw.githubusercontent.com/huggingface/datasets/{scripts_version}/datasets/_dummy/_dummy.py",
+                        str(context.exception),
+                    )
 
     def test_load_dataset_users(self):
         with self.assertRaises(FileNotFoundError) as context:
@@ -234,10 +378,7 @@ def test_load_dataset_users(self):
             with offline(offline_simulation_mode):
                 with self.assertRaises(ConnectionError) as context:
                     datasets.load_dataset("lhoestq/_dummy")
-                self.assertIn(
-                    "lhoestq/_dummy",
-                    str(context.exception),
-                )
+                self.assertIn("lhoestq/_dummy", str(context.exception))
 
 
 def test_load_dataset_builder_for_absolute_script_dir(dataset_loading_script_dir, data_dir):
@@ -271,8 +412,8 @@ def test_load_dataset_builder_for_absolute_data_dir(complex_data_dir):
     assert isinstance(builder, DatasetBuilder)
     assert builder.name == "text"
     assert builder.config.name == Path(complex_data_dir).name
-    assert isinstance(builder.config.data_files, list)
-    assert len(builder.config.data_files) > 0
+    assert isinstance(builder.config.data_files, DataFilesDict)
+    assert len(builder.config.data_files["train"]) > 0
 
 
 def test_load_dataset_builder_for_relative_data_dir(complex_data_dir):
@@ -283,14 +424,15 @@ def test_load_dataset_builder_for_relative_data_dir(complex_data_dir):
         assert isinstance(builder, DatasetBuilder)
         assert builder.name == "text"
         assert builder.config.name == relative_data_dir
-        assert isinstance(builder.config.data_files, list)
-        assert len(builder.config.data_files) > 0
+        assert isinstance(builder.config.data_files, DataFilesDict)
+        assert len(builder.config.data_files["train"]) > 0
 
 
 def test_load_dataset_builder_for_community_dataset_with_script():
     builder = datasets.load_dataset_builder(SAMPLE_DATASET_IDENTIFIER)
     assert isinstance(builder, DatasetBuilder)
     assert builder.name == SAMPLE_DATASET_IDENTIFIER.split("/")[-1]
+    assert builder.config.name == "default"
     assert builder.info.features == Features({"text": Value("string")})
     namespace = SAMPLE_DATASET_IDENTIFIER[: SAMPLE_DATASET_IDENTIFIER.index("/")]
     assert builder._relative_data_dir().startswith(namespace)
@@ -301,11 +443,9 @@ def test_load_dataset_builder_for_community_dataset_without_script():
     builder = datasets.load_dataset_builder(SAMPLE_DATASET_IDENTIFIER2)
     assert isinstance(builder, DatasetBuilder)
     assert builder.name == "text"
-    assert builder.config.name == SAMPLE_DATASET_IDENTIFIER2.split("/")[-1]
-    namespace = SAMPLE_DATASET_IDENTIFIER[: SAMPLE_DATASET_IDENTIFIER.index("/")]
-    assert builder._relative_data_dir().startswith(namespace)
-    assert isinstance(builder.config.data_files, list)
-    assert len(builder.config.data_files) > 0
+    assert builder.config.name == SAMPLE_DATASET_IDENTIFIER2.replace("/", "___")
+    assert isinstance(builder.config.data_files, DataFilesDict)
+    assert len(builder.config.data_files["train"]) > 0
 
 
 def test_load_dataset_builder_fail():
@@ -403,17 +543,17 @@ def test_loading_from_the_datasets_hub():
 
 
 def test_loading_from_the_datasets_hub_with_use_auth_token():
-    from datasets.utils.file_utils import http_head
+    from requests import get
 
     def assert_auth(url, *args, headers, **kwargs):
         assert headers["authorization"] == "Bearer foo"
-        return http_head(url, *args, headers=headers, **kwargs)
+        return get(url, *args, headers=headers, **kwargs)
 
-    with patch("datasets.utils.file_utils.http_head") as mock_head:
+    with patch("requests.get") as mock_head:
         mock_head.side_effect = assert_auth
         with tempfile.TemporaryDirectory() as tmp_dir:
             with offline():
-                with pytest.raises(ConnectionError):
+                with pytest.raises((ConnectionError, requests.exceptions.ConnectionError)):
                     load_dataset(SAMPLE_NOT_EXISTING_DATASET_IDENTIFIER, cache_dir=tmp_dir, use_auth_token="foo")
         mock_head.assert_called()
 
@@ -520,98 +660,3 @@ def test_load_dataset_deletes_extracted_files(deleted, jsonl_gz_path, tmp_path):
         ds = load_dataset("json", split="train", data_files=data_files, cache_dir=cache_dir)
     assert ds[0] == {"col_1": "0", "col_2": 0, "col_3": 0.0}
     assert (sorted((cache_dir / "downloads" / "extracted").iterdir()) == []) is deleted
-
-
-@pytest.mark.parametrize(
-    "pattern,size", [("*", 2), ("**/*", 2), ("*.txt", 2), ("data/*", 2), ("**/*.txt", 2), ("**/train.txt", 1)]
-)
-def test_resolve_data_files_locally_or_by_urls(complex_data_dir, pattern, size):
-    resolved_data_files = _resolve_data_files_locally_or_by_urls(complex_data_dir, pattern)
-    files_to_ignore = {".dummy", "README.md"}
-    expected_resolved_data_files = [
-        path for path in Path(complex_data_dir).rglob(pattern) if path.name not in files_to_ignore and path.is_file()
-    ]
-    assert len(resolved_data_files) == size
-    assert sorted(resolved_data_files) == sorted(expected_resolved_data_files)
-    assert all(isinstance(path, Path) for path in resolved_data_files)
-    assert all(path.is_file() for path in resolved_data_files)
-
-
-def test_resolve_data_files_locally_or_by_urls_with_absolute_path(tmp_path, complex_data_dir):
-    abs_path = os.path.join(complex_data_dir, "data", "train.txt")
-    resolved_data_files = _resolve_data_files_locally_or_by_urls(str(tmp_path / "blabla"), abs_path)
-    assert len(resolved_data_files) == 1
-
-
-@pytest.mark.parametrize("pattern,size,extensions", [("*", 2, ["txt"]), ("*", 2, None), ("*", 0, ["blablabla"])])
-def test_resolve_data_files_locally_or_by_urls_with_extensions(complex_data_dir, pattern, size, extensions):
-    if size > 0:
-        resolved_data_files = _resolve_data_files_locally_or_by_urls(
-            complex_data_dir, pattern, allowed_extensions=extensions
-        )
-        assert len(resolved_data_files) == size
-    else:
-        with pytest.raises(FileNotFoundError):
-            _resolve_data_files_locally_or_by_urls(complex_data_dir, pattern, allowed_extensions=extensions)
-
-
-def test_fail_resolve_data_files_locally_or_by_urls(complex_data_dir):
-    with pytest.raises(FileNotFoundError):
-        _resolve_data_files_locally_or_by_urls(complex_data_dir, "blablabla")
-
-
-@pytest.mark.parametrize(
-    "pattern,size", [("*", 2), ("**/*", 2), ("*.txt", 2), ("data/*", 2), ("**/*.txt", 2), ("**/train.txt", 1)]
-)
-def test_resolve_data_files_in_dataset_repository(complex_data_dir, pattern, size):
-    dataset_info = DatasetInfo(
-        siblings=[
-            {"rfilename": path.relative_to(complex_data_dir).as_posix()}
-            for path in Path(complex_data_dir).rglob("*")
-            if path.is_file()
-        ]
-    )
-    resolved_data_files = _resolve_data_files_in_dataset_repository(dataset_info, pattern)
-    files_to_ignore = {".dummy", "README.md"}
-    expected_resolved_data_files = [
-        path.relative_to(complex_data_dir)
-        for path in Path(complex_data_dir).rglob(pattern)
-        if path.name not in files_to_ignore and path.is_file()
-    ]
-    assert len(resolved_data_files) == size
-    assert sorted(resolved_data_files) == sorted(expected_resolved_data_files)
-    assert all(isinstance(path, PurePath) for path in resolved_data_files)
-    assert all((Path(complex_data_dir) / path).is_file() for path in resolved_data_files)
-
-
-@pytest.mark.parametrize("pattern,size,extensions", [("*", 2, ["txt"]), ("*", 2, None), ("*", 0, ["blablabla"])])
-def test_resolve_data_files_in_dataset_repository_with_extensions(complex_data_dir, pattern, size, extensions):
-    dataset_info = DatasetInfo(
-        siblings=[
-            {"rfilename": path.relative_to(complex_data_dir).as_posix()}
-            for path in Path(complex_data_dir).rglob("*")
-            if path.is_file()
-        ]
-    )
-    if size > 0:
-        resolved_data_files = _resolve_data_files_in_dataset_repository(
-            dataset_info, pattern, allowed_extensions=extensions
-        )
-        assert len(resolved_data_files) == size
-    else:
-        with pytest.raises(FileNotFoundError):
-            resolved_data_files = _resolve_data_files_in_dataset_repository(
-                dataset_info, pattern, allowed_extensions=extensions
-            )
-
-
-def test_fail_resolve_data_files_in_dataset_repository(complex_data_dir):
-    dataset_info = DatasetInfo(
-        siblings=[
-            {"rfilename": path.relative_to(complex_data_dir).as_posix()}
-            for path in Path(complex_data_dir).rglob("*")
-            if path.is_file()
-        ]
-    )
-    with pytest.raises(FileNotFoundError):
-        _resolve_data_files_in_dataset_repository(dataset_info, "blablabla")
diff --git a/tests/test_metric_common.py b/tests/test_metric_common.py
index 038cd62507f..f57913c2900 100644
--- a/tests/test_metric_common.py
+++ b/tests/test_metric_common.py
@@ -62,7 +62,9 @@ class LocalMetricTest(parameterized.TestCase):
 
     def test_load_metric(self, metric_name):
         doctest.ELLIPSIS_MARKER = "[...]"
-        metric_module = importlib.import_module(datasets.load.prepare_module(os.path.join("metrics", metric_name))[0])
+        metric_module = importlib.import_module(
+            datasets.load.prepare_module(os.path.join("metrics", metric_name), dataset=False)[0]
+        )
         metric = datasets.load.import_main_class(metric_module.__name__, dataset=False)
         # check parameters
         parameters = inspect.signature(metric._compute).parameters
diff --git a/tests/utils.py b/tests/utils.py
index fe3701cd03b..34fef29f195 100644
--- a/tests/utils.py
+++ b/tests/utils.py
@@ -310,7 +310,8 @@ def offline_socket(*args, **kwargs):
     elif mode is OfflineSimulationMode.CONNECTION_TIMES_OUT:
         # inspired from https://stackoverflow.com/a/904609
         with patch("requests.request", timeout_request):
-            yield
+            with patch("requests.api.request", timeout_request):
+                yield
     elif mode is OfflineSimulationMode.HF_DATASETS_OFFLINE_SET_TO_1:
         with patch("datasets.config.HF_DATASETS_OFFLINE", True):
             yield


Code review comments:
diff_hunk: @@ -226,604 +236,1042 @@ def get_imports(file_path: str):
     return imports
 
 
-def _resolve_data_files_locally_or_by_urls(
-    base_path: str, patterns: Union[str, List[str], Dict], allowed_extensions: Optional[list] = None
-) -> Union[List[Path], Dict]:
+def _download_additional_modules(
+    name: str, base_path: str, imports: Tuple[str, str, str, str], download_config: Optional[DownloadConfig]
+) -> List[Tuple[str, str]]:
     """
-    Return the absolute paths to all the files that match the given patterns.
-    It also supports absolute paths in patterns.
-    If an URL is passed, it is returned as is."""
-    data_files_ignore = ["README.md", "config.json", "dataset_infos.json", "dummy_data.zip"]
-    if isinstance(patterns, str):
-        if is_remote_url(patterns):
-            return [patterns]
-        if is_relative_path(patterns):
-            glob_iter = list(Path(base_path).rglob(patterns))
+    Download additional module for a module <name>.py at URL (or local path) <base_path>/<name>.py
+    The imports must have been parsed first using ``get_imports``.
+
+    If some modules need to be installed with pip, an error is raised showing how to install them.
+    This function return the list of downloaded modules as tuples (import_name, module_file_path).
+
+    The downloaded modules can then be moved into an importable directory with ``_copy_script_and_other_resouces_in_importable_dir``.
+    """
+    local_imports = []
+    library_imports = []
+    for import_type, import_name, import_path, sub_directory in imports:
+        if import_type == "library":
+            library_imports.append((import_name, import_path))  # Import from a library
+            continue
+
+        if import_name == name:
+            raise ValueError(
+                f"Error in the {name} script, importing relative {import_name} module "
+                f"but {import_name} is the name of the script. "
+                f"Please change relative import {import_name} to another name and add a '# From: URL_OR_PATH' "
+                f"comment pointing to the original relative import file path."
+            )
+        if import_type == "internal":
+            url_or_filename = url_or_path_join(base_path, import_path + ".py")
+        elif import_type == "external":
+            url_or_filename = import_path
         else:
-            glob_iter = [Path(filepath) for filepath in glob.glob(patterns)]
+            raise ValueError("Wrong import_type")
 
-        matched_paths = [
-            filepath.resolve()
-            for filepath in glob_iter
-            if filepath.name not in data_files_ignore and not filepath.name.startswith(".") and filepath.is_file()
-        ]
-        if allowed_extensions is not None:
-            out = [
-                filepath
-                for filepath in matched_paths
-                if any(suffix[1:] in allowed_extensions for suffix in filepath.suffixes)
-            ]
-            if len(out) < len(matched_paths):
-                invalid_matched_files = list(set(matched_paths) - set(out))
-                logger.info(
-                    f"Some files matched the pattern '{patterns}' at {Path(base_path).resolve()} but don't have valid data file extensions: {invalid_matched_files}"
+        local_import_path = cached_path(
+            url_or_filename,
+            download_config=download_config,
+        )
+        if sub_directory is not None:
+            local_import_path = os.path.join(local_import_path, sub_directory)
+        local_imports.append((import_name, local_import_path))
+
+    # Check library imports
+    needs_to_be_installed = []
+    for library_import_name, library_import_path in library_imports:
+        try:
+            lib = importlib.import_module(library_import_name)  # noqa F841
+        except ImportError:
+            needs_to_be_installed.append((library_import_name, library_import_path))
+    if needs_to_be_installed:
+        raise ImportError(
+            f"To be able to use {name}, you need to install the following dependencies"
+            f"{[lib_name for lib_name, lib_path in needs_to_be_installed]} using 'pip install "
+            f"{' '.join([lib_path for lib_name, lib_path in needs_to_be_installed])}' for instance'"
+        )
+    return local_imports
+
+
+def _copy_script_and_other_resouces_in_importable_dir(
+    name: str,
+    importable_directory_path: str,
+    subdirrectory_name: str,
+    original_local_path: str,
+    local_imports: List[Tuple[str, str]],
+    additional_files: List[Tuple[str, str]],
+    download_mode: Optional[GenerateMode],
+) -> str:
+    """Copy a script and its required imports to an importable directory
+
+    Args:
+        name (str): name of the resource to load
+        importable_directory_path (str): path to the loadable folder in the dynamic modules directory
+        subdirrectory_name (str): name of the subdirectory in importable_directory_path in which to place the script
+        original_local_path (str): local path to the resource script
+        local_imports (List[Tuple[str, str]]): list of (destination_filename, import_file_to_copy)
+        additional_files (List[Tuple[str, str]]): list of (destination_filename, additional_file_to_copy)
+        download_mode (Optional[GenerateMode]): download mode
+
+    Return:
+        importable_local_file: path to an importable module with importlib.import_module
+    """
+
+    # Define a directory with a unique name in our dataset or metric folder
+    # path is: ./datasets|metrics/dataset|metric_name/hash_from_code/script.py
+    # we use a hash as subdirrectory_name to be able to have multiple versions of a dataset/metric processing file together
+    importable_subdirectory = os.path.join(importable_directory_path, subdirrectory_name)
+    importable_local_file = os.path.join(importable_subdirectory, name + ".py")
+
+    # Prevent parallel disk operations
+    lock_path = importable_directory_path + ".lock"
+    with FileLock(lock_path):
+        # Create main dataset/metrics folder if needed
+        if download_mode == GenerateMode.FORCE_REDOWNLOAD and os.path.exists(importable_directory_path):
+            shutil.rmtree(importable_directory_path)
+        os.makedirs(importable_directory_path, exist_ok=True)
+
+        # add an __init__ file to the main dataset folder if needed
+        init_file_path = os.path.join(importable_directory_path, "__init__.py")
+        if not os.path.exists(init_file_path):
+            with open(init_file_path, "w"):
+                pass
+
+        # Create hash dataset folder if needed
+        os.makedirs(importable_subdirectory, exist_ok=True)
+        # add an __init__ file to the hash dataset folder if needed
+        init_file_path = os.path.join(importable_subdirectory, "__init__.py")
+        if not os.path.exists(init_file_path):
+            with open(init_file_path, "w"):
+                pass
+
+        # Copy dataset.py file in hash folder if needed
+        if not os.path.exists(importable_local_file):
+            shutil.copyfile(original_local_path, importable_local_file)
+
+        # Record metadata associating original dataset path with local unique folder
+        meta_path = importable_local_file.split(".py")[0] + ".json"
+        if not os.path.exists(meta_path):
+            meta = {"original file path": original_local_path, "local file path": importable_local_file}
+            # the filename is *.py in our case, so better rename to filenam.json instead of filename.py.json
+            with open(meta_path, "w", encoding="utf-8") as meta_file:
+                json.dump(meta, meta_file)
+
+        # Copy all the additional imports
+        for import_name, import_path in local_imports:
+            if os.path.isfile(import_path):
+                full_path_local_import = os.path.join(importable_subdirectory, import_name + ".py")
+                if not os.path.exists(full_path_local_import):
+                    shutil.copyfile(import_path, full_path_local_import)
+            elif os.path.isdir(import_path):
+                full_path_local_import = os.path.join(importable_subdirectory, import_name)
+                if not os.path.exists(full_path_local_import):
+                    shutil.copytree(import_path, full_path_local_import)
+            else:
+                raise OSError(f"Error with local import at {import_path}")
+
+        # Copy aditional files like dataset infos file if needed
+        for file_name, original_path in additional_files:
+            destination_additional_path = os.path.join(importable_subdirectory, file_name)
+            if not os.path.exists(destination_additional_path) or not filecmp.cmp(
+                original_path, destination_additional_path
+            ):
+                shutil.copyfile(original_path, destination_additional_path)
+        return importable_local_file
+
+
+def infer_module_for_data_files(data_files_list: DataFilesList) -> Optional[str]:
+    extensions_counter = Counter(suffix[1:] for filepath in data_files_list for suffix in Path(filepath).suffixes)
+    if extensions_counter:
+        return _EXTENSION_TO_MODULE[extensions_counter.most_common(1)[0][0]]
+
+
+@dataclass
+class DatasetModuleFactoryResult:
+    module_path: str
+    hash: str
+    builder_kwargs: dict
+
+
+@dataclass
+class MetricModuleFactoryResult:
+    module_path: str
+    hash: str
+
+
+class _DatasetModuleFactory:
+    def get_module(self) -> DatasetModuleFactoryResult:
+        raise NotImplementedError
+
+
+class _MetricModuleFactory:
+    def get_module(self) -> MetricModuleFactoryResult:
+        raise NotImplementedError
+
+
+class CanonicalDatasetModuleFactory(_DatasetModuleFactory):
+    """Get the module of a canonical dataset. The dataset script is downloaded from GitHub."""
+
+    def __init__(
+        self,
+        name: str,
+        revision: Optional[Union[str, Version]] = None,
+        download_config: Optional[DownloadConfig] = None,
+        download_mode: Optional[GenerateMode] = None,
+        dynamic_modules_path: Optional[str] = None,
+    ):
+        self.name = name
+        self.revision = revision
+        self.download_config = download_config
+        self.download_mode = download_mode
+        self.dynamic_modules_path = dynamic_modules_path
+        assert self.name.count("/") == 0
+        increase_load_count(name, resource_type="dataset")
+
+    def download_dataset_script(self) -> str:
+        file_path = hf_github_url(path=self.name, name=self.name + ".py", revision=self.revision)
+        return cached_path(file_path, download_config=self.download_config)
+
+    def download_dataset_script_from_master(self) -> str:
+        file_path = hf_github_url(path=self.name, name=self.name + ".py", revision="master")
+        local_path = cached_path(file_path, download_config=self.download_config)
+        logger.warning(
+            f"Couldn't find a directory or a dataset named '{self.name}'. "
+            f"It was picked from the master branch on github instead at {file_path}"
+        )
path: src/datasets/load.py
text: @user1:
I wouldn't log this warning from within the `download_dataset_script_from_master()` method because:
- This implies this method is always dependent on some other method previously called. I would move this warning one level up, from within its caller (as the caller is responsible of calling the other method and this one).
- The warning is already present in the caller `get_module()` (see line 465)

@user1:
Also note that once you remove the warning, both methods have the same implementation with different `revision` parameter values.

@author:
Good catch !

@author:
I merged both methods, and only kept the warning one level up.

diff_hunk: @@ -226,604 +236,1042 @@ def get_imports(file_path: str):
     return imports
 
 
-def _resolve_data_files_locally_or_by_urls(
-    base_path: str, patterns: Union[str, List[str], Dict], allowed_extensions: Optional[list] = None
-) -> Union[List[Path], Dict]:
+def _download_additional_modules(
+    name: str, base_path: str, imports: Tuple[str, str, str, str], download_config: Optional[DownloadConfig]
+) -> List[Tuple[str, str]]:
     """
-    Return the absolute paths to all the files that match the given patterns.
-    It also supports absolute paths in patterns.
-    If an URL is passed, it is returned as is."""
-    data_files_ignore = ["README.md", "config.json", "dataset_infos.json", "dummy_data.zip"]
-    if isinstance(patterns, str):
-        if is_remote_url(patterns):
-            return [patterns]
-        if is_relative_path(patterns):
-            glob_iter = list(Path(base_path).rglob(patterns))
+    Download additional module for a module <name>.py at URL (or local path) <base_path>/<name>.py
+    The imports must have been parsed first using ``get_imports``.
+
+    If some modules need to be installed with pip, an error is raised showing how to install them.
+    This function return the list of downloaded modules as tuples (import_name, module_file_path).
+
+    The downloaded modules can then be moved into an importable directory with ``_copy_script_and_other_resouces_in_importable_dir``.
+    """
+    local_imports = []
+    library_imports = []
+    for import_type, import_name, import_path, sub_directory in imports:
+        if import_type == "library":
+            library_imports.append((import_name, import_path))  # Import from a library
+            continue
+
+        if import_name == name:
+            raise ValueError(
+                f"Error in the {name} script, importing relative {import_name} module "
+                f"but {import_name} is the name of the script. "
+                f"Please change relative import {import_name} to another name and add a '# From: URL_OR_PATH' "
+                f"comment pointing to the original relative import file path."
+            )
+        if import_type == "internal":
+            url_or_filename = url_or_path_join(base_path, import_path + ".py")
+        elif import_type == "external":
+            url_or_filename = import_path
         else:
-            glob_iter = [Path(filepath) for filepath in glob.glob(patterns)]
+            raise ValueError("Wrong import_type")
 
-        matched_paths = [
-            filepath.resolve()
-            for filepath in glob_iter
-            if filepath.name not in data_files_ignore and not filepath.name.startswith(".") and filepath.is_file()
-        ]
-        if allowed_extensions is not None:
-            out = [
-                filepath
-                for filepath in matched_paths
-                if any(suffix[1:] in allowed_extensions for suffix in filepath.suffixes)
-            ]
-            if len(out) < len(matched_paths):
-                invalid_matched_files = list(set(matched_paths) - set(out))
-                logger.info(
-                    f"Some files matched the pattern '{patterns}' at {Path(base_path).resolve()} but don't have valid data file extensions: {invalid_matched_files}"
+        local_import_path = cached_path(
+            url_or_filename,
+            download_config=download_config,
+        )
+        if sub_directory is not None:
+            local_import_path = os.path.join(local_import_path, sub_directory)
+        local_imports.append((import_name, local_import_path))
+
+    # Check library imports
+    needs_to_be_installed = []
+    for library_import_name, library_import_path in library_imports:
+        try:
+            lib = importlib.import_module(library_import_name)  # noqa F841
+        except ImportError:
+            needs_to_be_installed.append((library_import_name, library_import_path))
+    if needs_to_be_installed:
+        raise ImportError(
+            f"To be able to use {name}, you need to install the following dependencies"
+            f"{[lib_name for lib_name, lib_path in needs_to_be_installed]} using 'pip install "
+            f"{' '.join([lib_path for lib_name, lib_path in needs_to_be_installed])}' for instance'"
+        )
+    return local_imports
+
+
+def _copy_script_and_other_resouces_in_importable_dir(
path: src/datasets/load.py
text: @user1:
```suggestion
def _copy_script_and_other_resources_in_importable_dir(
```

diff_hunk: @@ -226,604 +236,1042 @@ def get_imports(file_path: str):
     return imports
 
 
-def _resolve_data_files_locally_or_by_urls(
-    base_path: str, patterns: Union[str, List[str], Dict], allowed_extensions: Optional[list] = None
-) -> Union[List[Path], Dict]:
+def _download_additional_modules(
+    name: str, base_path: str, imports: Tuple[str, str, str, str], download_config: Optional[DownloadConfig]
+) -> List[Tuple[str, str]]:
     """
-    Return the absolute paths to all the files that match the given patterns.
-    It also supports absolute paths in patterns.
-    If an URL is passed, it is returned as is."""
-    data_files_ignore = ["README.md", "config.json", "dataset_infos.json", "dummy_data.zip"]
-    if isinstance(patterns, str):
-        if is_remote_url(patterns):
-            return [patterns]
-        if is_relative_path(patterns):
-            glob_iter = list(Path(base_path).rglob(patterns))
+    Download additional module for a module <name>.py at URL (or local path) <base_path>/<name>.py
+    The imports must have been parsed first using ``get_imports``.
+
+    If some modules need to be installed with pip, an error is raised showing how to install them.
+    This function return the list of downloaded modules as tuples (import_name, module_file_path).
+
+    The downloaded modules can then be moved into an importable directory with ``_copy_script_and_other_resouces_in_importable_dir``.
+    """
+    local_imports = []
+    library_imports = []
+    for import_type, import_name, import_path, sub_directory in imports:
+        if import_type == "library":
+            library_imports.append((import_name, import_path))  # Import from a library
+            continue
+
+        if import_name == name:
+            raise ValueError(
+                f"Error in the {name} script, importing relative {import_name} module "
+                f"but {import_name} is the name of the script. "
+                f"Please change relative import {import_name} to another name and add a '# From: URL_OR_PATH' "
+                f"comment pointing to the original relative import file path."
+            )
+        if import_type == "internal":
+            url_or_filename = url_or_path_join(base_path, import_path + ".py")
+        elif import_type == "external":
+            url_or_filename = import_path
         else:
-            glob_iter = [Path(filepath) for filepath in glob.glob(patterns)]
+            raise ValueError("Wrong import_type")
 
-        matched_paths = [
-            filepath.resolve()
-            for filepath in glob_iter
-            if filepath.name not in data_files_ignore and not filepath.name.startswith(".") and filepath.is_file()
-        ]
-        if allowed_extensions is not None:
-            out = [
-                filepath
-                for filepath in matched_paths
-                if any(suffix[1:] in allowed_extensions for suffix in filepath.suffixes)
-            ]
-            if len(out) < len(matched_paths):
-                invalid_matched_files = list(set(matched_paths) - set(out))
-                logger.info(
-                    f"Some files matched the pattern '{patterns}' at {Path(base_path).resolve()} but don't have valid data file extensions: {invalid_matched_files}"
+        local_import_path = cached_path(
+            url_or_filename,
+            download_config=download_config,
+        )
+        if sub_directory is not None:
+            local_import_path = os.path.join(local_import_path, sub_directory)
+        local_imports.append((import_name, local_import_path))
+
+    # Check library imports
+    needs_to_be_installed = []
+    for library_import_name, library_import_path in library_imports:
+        try:
+            lib = importlib.import_module(library_import_name)  # noqa F841
+        except ImportError:
+            needs_to_be_installed.append((library_import_name, library_import_path))
+    if needs_to_be_installed:
+        raise ImportError(
+            f"To be able to use {name}, you need to install the following dependencies"
+            f"{[lib_name for lib_name, lib_path in needs_to_be_installed]} using 'pip install "
+            f"{' '.join([lib_path for lib_name, lib_path in needs_to_be_installed])}' for instance'"
+        )
+    return local_imports
+
+
+def _copy_script_and_other_resouces_in_importable_dir(
+    name: str,
+    importable_directory_path: str,
+    subdirrectory_name: str,
+    original_local_path: str,
+    local_imports: List[Tuple[str, str]],
+    additional_files: List[Tuple[str, str]],
+    download_mode: Optional[GenerateMode],
+) -> str:
+    """Copy a script and its required imports to an importable directory
+
+    Args:
+        name (str): name of the resource to load
+        importable_directory_path (str): path to the loadable folder in the dynamic modules directory
+        subdirrectory_name (str): name of the subdirectory in importable_directory_path in which to place the script
+        original_local_path (str): local path to the resource script
+        local_imports (List[Tuple[str, str]]): list of (destination_filename, import_file_to_copy)
+        additional_files (List[Tuple[str, str]]): list of (destination_filename, additional_file_to_copy)
+        download_mode (Optional[GenerateMode]): download mode
+
+    Return:
+        importable_local_file: path to an importable module with importlib.import_module
+    """
+
+    # Define a directory with a unique name in our dataset or metric folder
+    # path is: ./datasets|metrics/dataset|metric_name/hash_from_code/script.py
+    # we use a hash as subdirrectory_name to be able to have multiple versions of a dataset/metric processing file together
+    importable_subdirectory = os.path.join(importable_directory_path, subdirrectory_name)
+    importable_local_file = os.path.join(importable_subdirectory, name + ".py")
+
+    # Prevent parallel disk operations
+    lock_path = importable_directory_path + ".lock"
+    with FileLock(lock_path):
+        # Create main dataset/metrics folder if needed
+        if download_mode == GenerateMode.FORCE_REDOWNLOAD and os.path.exists(importable_directory_path):
+            shutil.rmtree(importable_directory_path)
+        os.makedirs(importable_directory_path, exist_ok=True)
+
+        # add an __init__ file to the main dataset folder if needed
+        init_file_path = os.path.join(importable_directory_path, "__init__.py")
+        if not os.path.exists(init_file_path):
+            with open(init_file_path, "w"):
+                pass
+
+        # Create hash dataset folder if needed
+        os.makedirs(importable_subdirectory, exist_ok=True)
+        # add an __init__ file to the hash dataset folder if needed
+        init_file_path = os.path.join(importable_subdirectory, "__init__.py")
+        if not os.path.exists(init_file_path):
+            with open(init_file_path, "w"):
+                pass
+
+        # Copy dataset.py file in hash folder if needed
+        if not os.path.exists(importable_local_file):
+            shutil.copyfile(original_local_path, importable_local_file)
+
+        # Record metadata associating original dataset path with local unique folder
+        meta_path = importable_local_file.split(".py")[0] + ".json"
+        if not os.path.exists(meta_path):
+            meta = {"original file path": original_local_path, "local file path": importable_local_file}
+            # the filename is *.py in our case, so better rename to filenam.json instead of filename.py.json
+            with open(meta_path, "w", encoding="utf-8") as meta_file:
+                json.dump(meta, meta_file)
+
+        # Copy all the additional imports
+        for import_name, import_path in local_imports:
+            if os.path.isfile(import_path):
+                full_path_local_import = os.path.join(importable_subdirectory, import_name + ".py")
+                if not os.path.exists(full_path_local_import):
+                    shutil.copyfile(import_path, full_path_local_import)
+            elif os.path.isdir(import_path):
+                full_path_local_import = os.path.join(importable_subdirectory, import_name)
+                if not os.path.exists(full_path_local_import):
+                    shutil.copytree(import_path, full_path_local_import)
+            else:
+                raise OSError(f"Error with local import at {import_path}")
+
+        # Copy aditional files like dataset infos file if needed
+        for file_name, original_path in additional_files:
+            destination_additional_path = os.path.join(importable_subdirectory, file_name)
+            if not os.path.exists(destination_additional_path) or not filecmp.cmp(
+                original_path, destination_additional_path
+            ):
+                shutil.copyfile(original_path, destination_additional_path)
+        return importable_local_file
+
+
+def infer_module_for_data_files(data_files_list: DataFilesList) -> Optional[str]:
+    extensions_counter = Counter(suffix[1:] for filepath in data_files_list for suffix in Path(filepath).suffixes)
+    if extensions_counter:
+        return _EXTENSION_TO_MODULE[extensions_counter.most_common(1)[0][0]]
+
+
+@dataclass
+class DatasetModuleFactoryResult:
+    module_path: str
+    hash: str
+    builder_kwargs: dict
+
+
+@dataclass
+class MetricModuleFactoryResult:
+    module_path: str
+    hash: str
+
+
+class _DatasetModuleFactory:
+    def get_module(self) -> DatasetModuleFactoryResult:
+        raise NotImplementedError
+
+
+class _MetricModuleFactory:
+    def get_module(self) -> MetricModuleFactoryResult:
+        raise NotImplementedError
+
+
+class CanonicalDatasetModuleFactory(_DatasetModuleFactory):
+    """Get the module of a canonical dataset. The dataset script is downloaded from GitHub."""
+
+    def __init__(
+        self,
+        name: str,
+        revision: Optional[Union[str, Version]] = None,
+        download_config: Optional[DownloadConfig] = None,
+        download_mode: Optional[GenerateMode] = None,
+        dynamic_modules_path: Optional[str] = None,
+    ):
+        self.name = name
+        self.revision = revision
+        self.download_config = download_config
+        self.download_mode = download_mode
+        self.dynamic_modules_path = dynamic_modules_path
+        assert self.name.count("/") == 0
+        increase_load_count(name, resource_type="dataset")
+
+    def download_dataset_script(self) -> str:
+        file_path = hf_github_url(path=self.name, name=self.name + ".py", revision=self.revision)
+        return cached_path(file_path, download_config=self.download_config)
+
+    def download_dataset_script_from_master(self) -> str:
+        file_path = hf_github_url(path=self.name, name=self.name + ".py", revision="master")
+        local_path = cached_path(file_path, download_config=self.download_config)
+        logger.warning(
+            f"Couldn't find a directory or a dataset named '{self.name}'. "
+            f"It was picked from the master branch on github instead at {file_path}"
+        )
+        return local_path
+
+    def download_dataset_infos_file(self, revision: Optional[str]) -> str:
+        dataset_infos = hf_github_url(path=self.name, name=config.DATASETDICT_INFOS_FILENAME, revision=revision)
+        # Download the dataset infos file if available
+        try:
+            return cached_path(
+                dataset_infos,
+                download_config=self.download_config,
+            )
+        except (FileNotFoundError, ConnectionError):
+            return None
+
+    def get_module(self) -> DatasetModuleFactoryResult:
+        # get script and other files
+        try:
+            local_path = self.download_dataset_script()
+            revision = self.revision
+        except FileNotFoundError:
+            if self.revision is not None:
+                raise
+            else:
+                local_path = self.download_dataset_script_from_master()
+                logger.warning(
+                    f"Dataset '{self.name}' not available for this version, using the one from the 'master' branch instead."
                 )
-        else:
-            out = matched_paths
-        if not out:
-            error_msg = f"Unable to resolve any data file that matches '{patterns}' at {Path(base_path).resolve()}"
-            if allowed_extensions is not None:
-                error_msg += f" with any supported extension {list(allowed_extensions)}"
-            raise FileNotFoundError(error_msg)
-        return out
-    elif isinstance(patterns, dict):
-        return {
-            k: _resolve_data_files_locally_or_by_urls(base_path, v, allowed_extensions=allowed_extensions)
-            for k, v in patterns.items()
+                revision = "master"
+        dataset_infos_path = self.download_dataset_infos_file(revision)
+        imports = get_imports(local_path)
+        local_imports = _download_additional_modules(
+            name=self.name,
+            base_path=hf_github_url(path=self.name, name="", revision=revision),
+            imports=imports,
+            download_config=self.download_config,
+        )
+        additional_files = [(config.DATASETDICT_INFOS_FILENAME, dataset_infos_path)] if dataset_infos_path else []
+        # the copy the script and the files in an importable directory
+        dynamic_modules_path = self.dynamic_modules_path if self.dynamic_modules_path else init_dynamic_modules()
+        importable_directory_path = os.path.join(dynamic_modules_path, "datasets", self.name)
+        Path(importable_directory_path).mkdir(parents=True, exist_ok=True)
+        (Path(importable_directory_path).parent / "__init__.py").touch(exist_ok=True)
+        hash = files_to_hash([local_path] + [loc[1] for loc in local_imports])
+        importable_local_file = _copy_script_and_other_resouces_in_importable_dir(
+            name=self.name,
+            importable_directory_path=importable_directory_path,
+            subdirrectory_name=hash,
+            original_local_path=local_path,
+            local_imports=local_imports,
+            additional_files=additional_files,
+            download_mode=self.download_mode,
+        )
+        logger.debug(f"Created importable dataset file at {importable_local_file}")
+        # make the new module to be noticed by the import system
+        importlib.invalidate_caches()
+        module_path = ".".join([os.path.basename(dynamic_modules_path), "datasets", self.name, hash, self.name])
+        builder_kwargs = {"hash": hash, "base_path": hf_github_url(self.name, "", revision=revision)}
+        return DatasetModuleFactoryResult(module_path, hash, builder_kwargs)
+
+
+class CanonicalMetricModuleFactory(_MetricModuleFactory):
+    """Get the module of a metric. The metric script is downloaded from GitHub."""
+
+    def __init__(
+        self,
+        name: str,
+        revision: Optional[Union[str, Version]] = None,
+        download_config: Optional[DownloadConfig] = None,
+        download_mode: Optional[GenerateMode] = None,
+        dynamic_modules_path: Optional[str] = None,
+    ):
+        self.name = name
+        self.revision = revision
+        self.download_config = download_config
+        self.download_mode = download_mode
+        self.dynamic_modules_path = dynamic_modules_path
+        assert self.name.count("/") == 0
+        increase_load_count(name, resource_type="metric")
+
+    def download_metric_script(self) -> str:
+        file_path = hf_github_url(path=self.name, name=self.name + ".py", revision=self.revision, dataset=False)
+        return cached_path(file_path, download_config=self.download_config)
+
+    def download_metric_script_from_master(self) -> str:
+        file_path = hf_github_url(path=self.name, name=self.name + ".py", revision="master", dataset=False)
+        local_path = cached_path(file_path, download_config=self.download_config)
+        logger.warning(
+            f"Couldn't find a directory or a metric named '{self.name}'. "
+            f"It was picked from the master branch on github instead at {file_path}"
+        )
+        return local_path
+
+    def get_module(self) -> MetricModuleFactoryResult:
+        # get script and other files
+        try:
+            local_path = self.download_metric_script()
+            revision = self.revision
+        except FileNotFoundError:
+            if self.revision is not None:
+                raise
+            else:
+                local_path = self.download_metric_script_from_master()
+                revision = "master"
path: src/datasets/load.py
text: @user1:
I know this logic was implemented before your PR, but I don't completely understand it.

- Which is the typical case where you don't pass revision (`self.revision is None`) and after the `FileNotFoundError` you then try a second time with `revision="master"`?
- On the other hand, if you pass explicitly `revision="master"` and `FileNotFoundError` is thrown, do you reaaly wan to try a second time with `revision="master"` as before?

@author:
Good questions !

For question 1: There can be a weeks between the time a dataset in added on master and the time a new version of `datasets` is released. This fallback on the `master` branch makes it possible for anyone to load a dataset that was just added on `master` without waiting for a release or installing from source. In particular, the https://hf.co/datasets page is synced with the `master` branch, therefore user expect to be able to load any dataset from the website, and this features enables this.

For question 2: you're right, we shouldn't try again if the user specified a `revision` (that could be master or any other revision). Thanks for catching this !

@user1:
Question 1: ~but if you don't pass revision in the first call, then "master" is used (as default), isn't it? Why calling a second time with `revision=master`?~
Sorry, `__version__` is used. OK.

@user1:
Question 2: I just saw I was confused: if you pass explicitly `revision=master` in the first call and then it fails, the error is raised and no second call is done.

@user1:
I make another question:
- Question 3: if a user has a specific value for `HF_SCRIPTS_VERSION` env variable, and then fails, does it make sense to make a second call and return the "master" revision?

@author:
Q3: It probably doesn't make sense to get the master version in this case. We should only look on master if nothing has been particularly specified by the user IMO. This way users can have control on the determinism of the library. I just changed that so that it only checks on master if both `revision` and `HF_SCRIPTS_VERSION ` are unspecified (which is the default). Let me know what you think

diff_hunk: @@ -226,604 +236,1042 @@ def get_imports(file_path: str):
     return imports
 
 
-def _resolve_data_files_locally_or_by_urls(
-    base_path: str, patterns: Union[str, List[str], Dict], allowed_extensions: Optional[list] = None
-) -> Union[List[Path], Dict]:
+def _download_additional_modules(
+    name: str, base_path: str, imports: Tuple[str, str, str, str], download_config: Optional[DownloadConfig]
+) -> List[Tuple[str, str]]:
     """
-    Return the absolute paths to all the files that match the given patterns.
-    It also supports absolute paths in patterns.
-    If an URL is passed, it is returned as is."""
-    data_files_ignore = ["README.md", "config.json", "dataset_infos.json", "dummy_data.zip"]
-    if isinstance(patterns, str):
-        if is_remote_url(patterns):
-            return [patterns]
-        if is_relative_path(patterns):
-            glob_iter = list(Path(base_path).rglob(patterns))
+    Download additional module for a module <name>.py at URL (or local path) <base_path>/<name>.py
+    The imports must have been parsed first using ``get_imports``.
+
+    If some modules need to be installed with pip, an error is raised showing how to install them.
+    This function return the list of downloaded modules as tuples (import_name, module_file_path).
+
+    The downloaded modules can then be moved into an importable directory with ``_copy_script_and_other_resouces_in_importable_dir``.
+    """
+    local_imports = []
+    library_imports = []
+    for import_type, import_name, import_path, sub_directory in imports:
+        if import_type == "library":
+            library_imports.append((import_name, import_path))  # Import from a library
+            continue
+
+        if import_name == name:
+            raise ValueError(
+                f"Error in the {name} script, importing relative {import_name} module "
+                f"but {import_name} is the name of the script. "
+                f"Please change relative import {import_name} to another name and add a '# From: URL_OR_PATH' "
+                f"comment pointing to the original relative import file path."
+            )
+        if import_type == "internal":
+            url_or_filename = url_or_path_join(base_path, import_path + ".py")
+        elif import_type == "external":
+            url_or_filename = import_path
         else:
-            glob_iter = [Path(filepath) for filepath in glob.glob(patterns)]
+            raise ValueError("Wrong import_type")
 
-        matched_paths = [
-            filepath.resolve()
-            for filepath in glob_iter
-            if filepath.name not in data_files_ignore and not filepath.name.startswith(".") and filepath.is_file()
-        ]
-        if allowed_extensions is not None:
-            out = [
-                filepath
-                for filepath in matched_paths
-                if any(suffix[1:] in allowed_extensions for suffix in filepath.suffixes)
-            ]
-            if len(out) < len(matched_paths):
-                invalid_matched_files = list(set(matched_paths) - set(out))
-                logger.info(
-                    f"Some files matched the pattern '{patterns}' at {Path(base_path).resolve()} but don't have valid data file extensions: {invalid_matched_files}"
+        local_import_path = cached_path(
+            url_or_filename,
+            download_config=download_config,
+        )
+        if sub_directory is not None:
+            local_import_path = os.path.join(local_import_path, sub_directory)
+        local_imports.append((import_name, local_import_path))
+
+    # Check library imports
+    needs_to_be_installed = []
+    for library_import_name, library_import_path in library_imports:
+        try:
+            lib = importlib.import_module(library_import_name)  # noqa F841
+        except ImportError:
+            needs_to_be_installed.append((library_import_name, library_import_path))
+    if needs_to_be_installed:
+        raise ImportError(
+            f"To be able to use {name}, you need to install the following dependencies"
+            f"{[lib_name for lib_name, lib_path in needs_to_be_installed]} using 'pip install "
+            f"{' '.join([lib_path for lib_name, lib_path in needs_to_be_installed])}' for instance'"
+        )
+    return local_imports
+
+
+def _copy_script_and_other_resouces_in_importable_dir(
+    name: str,
+    importable_directory_path: str,
+    subdirrectory_name: str,
+    original_local_path: str,
+    local_imports: List[Tuple[str, str]],
+    additional_files: List[Tuple[str, str]],
+    download_mode: Optional[GenerateMode],
+) -> str:
+    """Copy a script and its required imports to an importable directory
+
+    Args:
+        name (str): name of the resource to load
+        importable_directory_path (str): path to the loadable folder in the dynamic modules directory
+        subdirrectory_name (str): name of the subdirectory in importable_directory_path in which to place the script
+        original_local_path (str): local path to the resource script
+        local_imports (List[Tuple[str, str]]): list of (destination_filename, import_file_to_copy)
+        additional_files (List[Tuple[str, str]]): list of (destination_filename, additional_file_to_copy)
+        download_mode (Optional[GenerateMode]): download mode
+
+    Return:
+        importable_local_file: path to an importable module with importlib.import_module
+    """
+
+    # Define a directory with a unique name in our dataset or metric folder
+    # path is: ./datasets|metrics/dataset|metric_name/hash_from_code/script.py
+    # we use a hash as subdirrectory_name to be able to have multiple versions of a dataset/metric processing file together
+    importable_subdirectory = os.path.join(importable_directory_path, subdirrectory_name)
+    importable_local_file = os.path.join(importable_subdirectory, name + ".py")
+
+    # Prevent parallel disk operations
+    lock_path = importable_directory_path + ".lock"
+    with FileLock(lock_path):
+        # Create main dataset/metrics folder if needed
+        if download_mode == GenerateMode.FORCE_REDOWNLOAD and os.path.exists(importable_directory_path):
+            shutil.rmtree(importable_directory_path)
+        os.makedirs(importable_directory_path, exist_ok=True)
+
+        # add an __init__ file to the main dataset folder if needed
+        init_file_path = os.path.join(importable_directory_path, "__init__.py")
+        if not os.path.exists(init_file_path):
+            with open(init_file_path, "w"):
+                pass
+
+        # Create hash dataset folder if needed
+        os.makedirs(importable_subdirectory, exist_ok=True)
+        # add an __init__ file to the hash dataset folder if needed
+        init_file_path = os.path.join(importable_subdirectory, "__init__.py")
+        if not os.path.exists(init_file_path):
+            with open(init_file_path, "w"):
+                pass
+
+        # Copy dataset.py file in hash folder if needed
+        if not os.path.exists(importable_local_file):
+            shutil.copyfile(original_local_path, importable_local_file)
+
+        # Record metadata associating original dataset path with local unique folder
+        meta_path = importable_local_file.split(".py")[0] + ".json"
+        if not os.path.exists(meta_path):
+            meta = {"original file path": original_local_path, "local file path": importable_local_file}
+            # the filename is *.py in our case, so better rename to filenam.json instead of filename.py.json
+            with open(meta_path, "w", encoding="utf-8") as meta_file:
+                json.dump(meta, meta_file)
+
+        # Copy all the additional imports
+        for import_name, import_path in local_imports:
+            if os.path.isfile(import_path):
+                full_path_local_import = os.path.join(importable_subdirectory, import_name + ".py")
+                if not os.path.exists(full_path_local_import):
+                    shutil.copyfile(import_path, full_path_local_import)
+            elif os.path.isdir(import_path):
+                full_path_local_import = os.path.join(importable_subdirectory, import_name)
+                if not os.path.exists(full_path_local_import):
+                    shutil.copytree(import_path, full_path_local_import)
+            else:
+                raise OSError(f"Error with local import at {import_path}")
+
+        # Copy aditional files like dataset infos file if needed
+        for file_name, original_path in additional_files:
+            destination_additional_path = os.path.join(importable_subdirectory, file_name)
+            if not os.path.exists(destination_additional_path) or not filecmp.cmp(
+                original_path, destination_additional_path
+            ):
+                shutil.copyfile(original_path, destination_additional_path)
+        return importable_local_file
+
+
+def infer_module_for_data_files(data_files_list: DataFilesList) -> Optional[str]:
+    extensions_counter = Counter(suffix[1:] for filepath in data_files_list for suffix in Path(filepath).suffixes)
+    if extensions_counter:
+        return _EXTENSION_TO_MODULE[extensions_counter.most_common(1)[0][0]]
+
+
+@dataclass
+class DatasetModuleFactoryResult:
+    module_path: str
+    hash: str
+    builder_kwargs: dict
+
+
+@dataclass
+class MetricModuleFactoryResult:
+    module_path: str
+    hash: str
+
+
+class _DatasetModuleFactory:
+    def get_module(self) -> DatasetModuleFactoryResult:
+        raise NotImplementedError
+
+
+class _MetricModuleFactory:
+    def get_module(self) -> MetricModuleFactoryResult:
+        raise NotImplementedError
+
+
+class CanonicalDatasetModuleFactory(_DatasetModuleFactory):
+    """Get the module of a canonical dataset. The dataset script is downloaded from GitHub."""
+
+    def __init__(
+        self,
+        name: str,
+        revision: Optional[Union[str, Version]] = None,
+        download_config: Optional[DownloadConfig] = None,
+        download_mode: Optional[GenerateMode] = None,
+        dynamic_modules_path: Optional[str] = None,
+    ):
+        self.name = name
+        self.revision = revision
+        self.download_config = download_config
+        self.download_mode = download_mode
+        self.dynamic_modules_path = dynamic_modules_path
+        assert self.name.count("/") == 0
+        increase_load_count(name, resource_type="dataset")
+
+    def download_dataset_script(self) -> str:
+        file_path = hf_github_url(path=self.name, name=self.name + ".py", revision=self.revision)
+        return cached_path(file_path, download_config=self.download_config)
+
+    def download_dataset_script_from_master(self) -> str:
+        file_path = hf_github_url(path=self.name, name=self.name + ".py", revision="master")
+        local_path = cached_path(file_path, download_config=self.download_config)
+        logger.warning(
+            f"Couldn't find a directory or a dataset named '{self.name}'. "
+            f"It was picked from the master branch on github instead at {file_path}"
+        )
+        return local_path
+
+    def download_dataset_infos_file(self, revision: Optional[str]) -> str:
+        dataset_infos = hf_github_url(path=self.name, name=config.DATASETDICT_INFOS_FILENAME, revision=revision)
+        # Download the dataset infos file if available
+        try:
+            return cached_path(
+                dataset_infos,
+                download_config=self.download_config,
+            )
+        except (FileNotFoundError, ConnectionError):
+            return None
+
+    def get_module(self) -> DatasetModuleFactoryResult:
+        # get script and other files
+        try:
+            local_path = self.download_dataset_script()
+            revision = self.revision
+        except FileNotFoundError:
+            if self.revision is not None:
+                raise
+            else:
+                local_path = self.download_dataset_script_from_master()
+                logger.warning(
+                    f"Dataset '{self.name}' not available for this version, using the one from the 'master' branch instead."
                 )
-        else:
-            out = matched_paths
-        if not out:
-            error_msg = f"Unable to resolve any data file that matches '{patterns}' at {Path(base_path).resolve()}"
-            if allowed_extensions is not None:
-                error_msg += f" with any supported extension {list(allowed_extensions)}"
-            raise FileNotFoundError(error_msg)
-        return out
-    elif isinstance(patterns, dict):
-        return {
-            k: _resolve_data_files_locally_or_by_urls(base_path, v, allowed_extensions=allowed_extensions)
-            for k, v in patterns.items()
+                revision = "master"
+        dataset_infos_path = self.download_dataset_infos_file(revision)
+        imports = get_imports(local_path)
+        local_imports = _download_additional_modules(
+            name=self.name,
+            base_path=hf_github_url(path=self.name, name="", revision=revision),
+            imports=imports,
+            download_config=self.download_config,
+        )
+        additional_files = [(config.DATASETDICT_INFOS_FILENAME, dataset_infos_path)] if dataset_infos_path else []
+        # the copy the script and the files in an importable directory
+        dynamic_modules_path = self.dynamic_modules_path if self.dynamic_modules_path else init_dynamic_modules()
+        importable_directory_path = os.path.join(dynamic_modules_path, "datasets", self.name)
+        Path(importable_directory_path).mkdir(parents=True, exist_ok=True)
+        (Path(importable_directory_path).parent / "__init__.py").touch(exist_ok=True)
+        hash = files_to_hash([local_path] + [loc[1] for loc in local_imports])
+        importable_local_file = _copy_script_and_other_resouces_in_importable_dir(
+            name=self.name,
+            importable_directory_path=importable_directory_path,
+            subdirrectory_name=hash,
+            original_local_path=local_path,
+            local_imports=local_imports,
+            additional_files=additional_files,
+            download_mode=self.download_mode,
+        )
+        logger.debug(f"Created importable dataset file at {importable_local_file}")
path: src/datasets/load.py
text: @user1:
Maybe create a method with this code?

Note that this code is repeated in several other classes.

@author:
Which lines would you like to see in another method ?

@user1:
Sorry, something went wrong... I try again...

diff_hunk: @@ -226,604 +236,1042 @@ def get_imports(file_path: str):
     return imports
 
 
-def _resolve_data_files_locally_or_by_urls(
-    base_path: str, patterns: Union[str, List[str], Dict], allowed_extensions: Optional[list] = None
-) -> Union[List[Path], Dict]:
+def _download_additional_modules(
+    name: str, base_path: str, imports: Tuple[str, str, str, str], download_config: Optional[DownloadConfig]
+) -> List[Tuple[str, str]]:
     """
-    Return the absolute paths to all the files that match the given patterns.
-    It also supports absolute paths in patterns.
-    If an URL is passed, it is returned as is."""
-    data_files_ignore = ["README.md", "config.json", "dataset_infos.json", "dummy_data.zip"]
-    if isinstance(patterns, str):
-        if is_remote_url(patterns):
-            return [patterns]
-        if is_relative_path(patterns):
-            glob_iter = list(Path(base_path).rglob(patterns))
+    Download additional module for a module <name>.py at URL (or local path) <base_path>/<name>.py
+    The imports must have been parsed first using ``get_imports``.
+
+    If some modules need to be installed with pip, an error is raised showing how to install them.
+    This function return the list of downloaded modules as tuples (import_name, module_file_path).
+
+    The downloaded modules can then be moved into an importable directory with ``_copy_script_and_other_resouces_in_importable_dir``.
+    """
+    local_imports = []
+    library_imports = []
+    for import_type, import_name, import_path, sub_directory in imports:
+        if import_type == "library":
+            library_imports.append((import_name, import_path))  # Import from a library
+            continue
+
+        if import_name == name:
+            raise ValueError(
+                f"Error in the {name} script, importing relative {import_name} module "
+                f"but {import_name} is the name of the script. "
+                f"Please change relative import {import_name} to another name and add a '# From: URL_OR_PATH' "
+                f"comment pointing to the original relative import file path."
+            )
+        if import_type == "internal":
+            url_or_filename = url_or_path_join(base_path, import_path + ".py")
+        elif import_type == "external":
+            url_or_filename = import_path
         else:
-            glob_iter = [Path(filepath) for filepath in glob.glob(patterns)]
+            raise ValueError("Wrong import_type")
 
-        matched_paths = [
-            filepath.resolve()
-            for filepath in glob_iter
-            if filepath.name not in data_files_ignore and not filepath.name.startswith(".") and filepath.is_file()
-        ]
-        if allowed_extensions is not None:
-            out = [
-                filepath
-                for filepath in matched_paths
-                if any(suffix[1:] in allowed_extensions for suffix in filepath.suffixes)
-            ]
-            if len(out) < len(matched_paths):
-                invalid_matched_files = list(set(matched_paths) - set(out))
-                logger.info(
-                    f"Some files matched the pattern '{patterns}' at {Path(base_path).resolve()} but don't have valid data file extensions: {invalid_matched_files}"
+        local_import_path = cached_path(
+            url_or_filename,
+            download_config=download_config,
+        )
+        if sub_directory is not None:
+            local_import_path = os.path.join(local_import_path, sub_directory)
+        local_imports.append((import_name, local_import_path))
+
+    # Check library imports
+    needs_to_be_installed = []
+    for library_import_name, library_import_path in library_imports:
+        try:
+            lib = importlib.import_module(library_import_name)  # noqa F841
+        except ImportError:
+            needs_to_be_installed.append((library_import_name, library_import_path))
+    if needs_to_be_installed:
+        raise ImportError(
+            f"To be able to use {name}, you need to install the following dependencies"
+            f"{[lib_name for lib_name, lib_path in needs_to_be_installed]} using 'pip install "
+            f"{' '.join([lib_path for lib_name, lib_path in needs_to_be_installed])}' for instance'"
+        )
+    return local_imports
+
+
+def _copy_script_and_other_resouces_in_importable_dir(
+    name: str,
+    importable_directory_path: str,
+    subdirrectory_name: str,
+    original_local_path: str,
+    local_imports: List[Tuple[str, str]],
+    additional_files: List[Tuple[str, str]],
+    download_mode: Optional[GenerateMode],
+) -> str:
+    """Copy a script and its required imports to an importable directory
+
+    Args:
+        name (str): name of the resource to load
+        importable_directory_path (str): path to the loadable folder in the dynamic modules directory
+        subdirrectory_name (str): name of the subdirectory in importable_directory_path in which to place the script
+        original_local_path (str): local path to the resource script
+        local_imports (List[Tuple[str, str]]): list of (destination_filename, import_file_to_copy)
+        additional_files (List[Tuple[str, str]]): list of (destination_filename, additional_file_to_copy)
+        download_mode (Optional[GenerateMode]): download mode
+
+    Return:
+        importable_local_file: path to an importable module with importlib.import_module
+    """
+
+    # Define a directory with a unique name in our dataset or metric folder
+    # path is: ./datasets|metrics/dataset|metric_name/hash_from_code/script.py
+    # we use a hash as subdirrectory_name to be able to have multiple versions of a dataset/metric processing file together
+    importable_subdirectory = os.path.join(importable_directory_path, subdirrectory_name)
+    importable_local_file = os.path.join(importable_subdirectory, name + ".py")
+
+    # Prevent parallel disk operations
+    lock_path = importable_directory_path + ".lock"
+    with FileLock(lock_path):
+        # Create main dataset/metrics folder if needed
+        if download_mode == GenerateMode.FORCE_REDOWNLOAD and os.path.exists(importable_directory_path):
+            shutil.rmtree(importable_directory_path)
+        os.makedirs(importable_directory_path, exist_ok=True)
+
+        # add an __init__ file to the main dataset folder if needed
+        init_file_path = os.path.join(importable_directory_path, "__init__.py")
+        if not os.path.exists(init_file_path):
+            with open(init_file_path, "w"):
+                pass
+
+        # Create hash dataset folder if needed
+        os.makedirs(importable_subdirectory, exist_ok=True)
+        # add an __init__ file to the hash dataset folder if needed
+        init_file_path = os.path.join(importable_subdirectory, "__init__.py")
+        if not os.path.exists(init_file_path):
+            with open(init_file_path, "w"):
+                pass
+
+        # Copy dataset.py file in hash folder if needed
+        if not os.path.exists(importable_local_file):
+            shutil.copyfile(original_local_path, importable_local_file)
+
+        # Record metadata associating original dataset path with local unique folder
+        meta_path = importable_local_file.split(".py")[0] + ".json"
+        if not os.path.exists(meta_path):
+            meta = {"original file path": original_local_path, "local file path": importable_local_file}
+            # the filename is *.py in our case, so better rename to filenam.json instead of filename.py.json
+            with open(meta_path, "w", encoding="utf-8") as meta_file:
+                json.dump(meta, meta_file)
+
+        # Copy all the additional imports
+        for import_name, import_path in local_imports:
+            if os.path.isfile(import_path):
+                full_path_local_import = os.path.join(importable_subdirectory, import_name + ".py")
+                if not os.path.exists(full_path_local_import):
+                    shutil.copyfile(import_path, full_path_local_import)
+            elif os.path.isdir(import_path):
+                full_path_local_import = os.path.join(importable_subdirectory, import_name)
+                if not os.path.exists(full_path_local_import):
+                    shutil.copytree(import_path, full_path_local_import)
+            else:
+                raise OSError(f"Error with local import at {import_path}")
+
+        # Copy aditional files like dataset infos file if needed
+        for file_name, original_path in additional_files:
+            destination_additional_path = os.path.join(importable_subdirectory, file_name)
+            if not os.path.exists(destination_additional_path) or not filecmp.cmp(
+                original_path, destination_additional_path
+            ):
+                shutil.copyfile(original_path, destination_additional_path)
+        return importable_local_file
+
+
+def infer_module_for_data_files(data_files_list: DataFilesList) -> Optional[str]:
+    extensions_counter = Counter(suffix[1:] for filepath in data_files_list for suffix in Path(filepath).suffixes)
+    if extensions_counter:
+        return _EXTENSION_TO_MODULE[extensions_counter.most_common(1)[0][0]]
+
+
+@dataclass
+class DatasetModuleFactoryResult:
+    module_path: str
+    hash: str
+    builder_kwargs: dict
+
+
+@dataclass
+class MetricModuleFactoryResult:
+    module_path: str
+    hash: str
+
+
+class _DatasetModuleFactory:
+    def get_module(self) -> DatasetModuleFactoryResult:
+        raise NotImplementedError
+
+
+class _MetricModuleFactory:
+    def get_module(self) -> MetricModuleFactoryResult:
+        raise NotImplementedError
+
+
+class CanonicalDatasetModuleFactory(_DatasetModuleFactory):
+    """Get the module of a canonical dataset. The dataset script is downloaded from GitHub."""
+
+    def __init__(
+        self,
+        name: str,
+        revision: Optional[Union[str, Version]] = None,
+        download_config: Optional[DownloadConfig] = None,
+        download_mode: Optional[GenerateMode] = None,
+        dynamic_modules_path: Optional[str] = None,
+    ):
+        self.name = name
+        self.revision = revision
+        self.download_config = download_config
+        self.download_mode = download_mode
+        self.dynamic_modules_path = dynamic_modules_path
+        assert self.name.count("/") == 0
+        increase_load_count(name, resource_type="dataset")
+
+    def download_dataset_script(self) -> str:
+        file_path = hf_github_url(path=self.name, name=self.name + ".py", revision=self.revision)
+        return cached_path(file_path, download_config=self.download_config)
+
+    def download_dataset_script_from_master(self) -> str:
+        file_path = hf_github_url(path=self.name, name=self.name + ".py", revision="master")
+        local_path = cached_path(file_path, download_config=self.download_config)
+        logger.warning(
+            f"Couldn't find a directory or a dataset named '{self.name}'. "
+            f"It was picked from the master branch on github instead at {file_path}"
+        )
+        return local_path
+
+    def download_dataset_infos_file(self, revision: Optional[str]) -> str:
+        dataset_infos = hf_github_url(path=self.name, name=config.DATASETDICT_INFOS_FILENAME, revision=revision)
+        # Download the dataset infos file if available
+        try:
+            return cached_path(
+                dataset_infos,
+                download_config=self.download_config,
+            )
+        except (FileNotFoundError, ConnectionError):
+            return None
+
+    def get_module(self) -> DatasetModuleFactoryResult:
+        # get script and other files
+        try:
+            local_path = self.download_dataset_script()
+            revision = self.revision
+        except FileNotFoundError:
+            if self.revision is not None:
+                raise
+            else:
+                local_path = self.download_dataset_script_from_master()
+                logger.warning(
+                    f"Dataset '{self.name}' not available for this version, using the one from the 'master' branch instead."
                 )
-        else:
-            out = matched_paths
-        if not out:
-            error_msg = f"Unable to resolve any data file that matches '{patterns}' at {Path(base_path).resolve()}"
-            if allowed_extensions is not None:
-                error_msg += f" with any supported extension {list(allowed_extensions)}"
-            raise FileNotFoundError(error_msg)
-        return out
-    elif isinstance(patterns, dict):
-        return {
-            k: _resolve_data_files_locally_or_by_urls(base_path, v, allowed_extensions=allowed_extensions)
-            for k, v in patterns.items()
+                revision = "master"
+        dataset_infos_path = self.download_dataset_infos_file(revision)
+        imports = get_imports(local_path)
+        local_imports = _download_additional_modules(
+            name=self.name,
+            base_path=hf_github_url(path=self.name, name="", revision=revision),
+            imports=imports,
+            download_config=self.download_config,
+        )
+        additional_files = [(config.DATASETDICT_INFOS_FILENAME, dataset_infos_path)] if dataset_infos_path else []
+        # the copy the script and the files in an importable directory
path: src/datasets/load.py
text: @user1:
```suggestion
        # copy the script and the files in an importable directory
```

diff_hunk: @@ -226,604 +236,1042 @@ def get_imports(file_path: str):
     return imports
 
 
-def _resolve_data_files_locally_or_by_urls(
-    base_path: str, patterns: Union[str, List[str], Dict], allowed_extensions: Optional[list] = None
-) -> Union[List[Path], Dict]:
+def _download_additional_modules(
+    name: str, base_path: str, imports: Tuple[str, str, str, str], download_config: Optional[DownloadConfig]
+) -> List[Tuple[str, str]]:
     """
-    Return the absolute paths to all the files that match the given patterns.
-    It also supports absolute paths in patterns.
-    If an URL is passed, it is returned as is."""
-    data_files_ignore = ["README.md", "config.json", "dataset_infos.json", "dummy_data.zip"]
-    if isinstance(patterns, str):
-        if is_remote_url(patterns):
-            return [patterns]
-        if is_relative_path(patterns):
-            glob_iter = list(Path(base_path).rglob(patterns))
+    Download additional module for a module <name>.py at URL (or local path) <base_path>/<name>.py
+    The imports must have been parsed first using ``get_imports``.
+
+    If some modules need to be installed with pip, an error is raised showing how to install them.
+    This function return the list of downloaded modules as tuples (import_name, module_file_path).
+
+    The downloaded modules can then be moved into an importable directory with ``_copy_script_and_other_resouces_in_importable_dir``.
+    """
+    local_imports = []
+    library_imports = []
+    for import_type, import_name, import_path, sub_directory in imports:
+        if import_type == "library":
+            library_imports.append((import_name, import_path))  # Import from a library
+            continue
+
+        if import_name == name:
+            raise ValueError(
+                f"Error in the {name} script, importing relative {import_name} module "
+                f"but {import_name} is the name of the script. "
+                f"Please change relative import {import_name} to another name and add a '# From: URL_OR_PATH' "
+                f"comment pointing to the original relative import file path."
+            )
+        if import_type == "internal":
+            url_or_filename = url_or_path_join(base_path, import_path + ".py")
+        elif import_type == "external":
+            url_or_filename = import_path
         else:
-            glob_iter = [Path(filepath) for filepath in glob.glob(patterns)]
+            raise ValueError("Wrong import_type")
 
-        matched_paths = [
-            filepath.resolve()
-            for filepath in glob_iter
-            if filepath.name not in data_files_ignore and not filepath.name.startswith(".") and filepath.is_file()
-        ]
-        if allowed_extensions is not None:
-            out = [
-                filepath
-                for filepath in matched_paths
-                if any(suffix[1:] in allowed_extensions for suffix in filepath.suffixes)
-            ]
-            if len(out) < len(matched_paths):
-                invalid_matched_files = list(set(matched_paths) - set(out))
-                logger.info(
-                    f"Some files matched the pattern '{patterns}' at {Path(base_path).resolve()} but don't have valid data file extensions: {invalid_matched_files}"
+        local_import_path = cached_path(
+            url_or_filename,
+            download_config=download_config,
+        )
+        if sub_directory is not None:
+            local_import_path = os.path.join(local_import_path, sub_directory)
+        local_imports.append((import_name, local_import_path))
+
+    # Check library imports
+    needs_to_be_installed = []
+    for library_import_name, library_import_path in library_imports:
+        try:
+            lib = importlib.import_module(library_import_name)  # noqa F841
+        except ImportError:
+            needs_to_be_installed.append((library_import_name, library_import_path))
+    if needs_to_be_installed:
+        raise ImportError(
+            f"To be able to use {name}, you need to install the following dependencies"
+            f"{[lib_name for lib_name, lib_path in needs_to_be_installed]} using 'pip install "
+            f"{' '.join([lib_path for lib_name, lib_path in needs_to_be_installed])}' for instance'"
+        )
+    return local_imports
+
+
+def _copy_script_and_other_resouces_in_importable_dir(
+    name: str,
+    importable_directory_path: str,
+    subdirrectory_name: str,
+    original_local_path: str,
+    local_imports: List[Tuple[str, str]],
+    additional_files: List[Tuple[str, str]],
+    download_mode: Optional[GenerateMode],
+) -> str:
+    """Copy a script and its required imports to an importable directory
+
+    Args:
+        name (str): name of the resource to load
+        importable_directory_path (str): path to the loadable folder in the dynamic modules directory
+        subdirrectory_name (str): name of the subdirectory in importable_directory_path in which to place the script
+        original_local_path (str): local path to the resource script
+        local_imports (List[Tuple[str, str]]): list of (destination_filename, import_file_to_copy)
+        additional_files (List[Tuple[str, str]]): list of (destination_filename, additional_file_to_copy)
+        download_mode (Optional[GenerateMode]): download mode
+
+    Return:
+        importable_local_file: path to an importable module with importlib.import_module
+    """
+
+    # Define a directory with a unique name in our dataset or metric folder
+    # path is: ./datasets|metrics/dataset|metric_name/hash_from_code/script.py
+    # we use a hash as subdirrectory_name to be able to have multiple versions of a dataset/metric processing file together
+    importable_subdirectory = os.path.join(importable_directory_path, subdirrectory_name)
+    importable_local_file = os.path.join(importable_subdirectory, name + ".py")
+
+    # Prevent parallel disk operations
+    lock_path = importable_directory_path + ".lock"
+    with FileLock(lock_path):
+        # Create main dataset/metrics folder if needed
+        if download_mode == GenerateMode.FORCE_REDOWNLOAD and os.path.exists(importable_directory_path):
+            shutil.rmtree(importable_directory_path)
+        os.makedirs(importable_directory_path, exist_ok=True)
+
+        # add an __init__ file to the main dataset folder if needed
+        init_file_path = os.path.join(importable_directory_path, "__init__.py")
+        if not os.path.exists(init_file_path):
+            with open(init_file_path, "w"):
+                pass
+
+        # Create hash dataset folder if needed
+        os.makedirs(importable_subdirectory, exist_ok=True)
+        # add an __init__ file to the hash dataset folder if needed
+        init_file_path = os.path.join(importable_subdirectory, "__init__.py")
+        if not os.path.exists(init_file_path):
+            with open(init_file_path, "w"):
+                pass
+
+        # Copy dataset.py file in hash folder if needed
+        if not os.path.exists(importable_local_file):
+            shutil.copyfile(original_local_path, importable_local_file)
+
+        # Record metadata associating original dataset path with local unique folder
+        meta_path = importable_local_file.split(".py")[0] + ".json"
+        if not os.path.exists(meta_path):
+            meta = {"original file path": original_local_path, "local file path": importable_local_file}
+            # the filename is *.py in our case, so better rename to filenam.json instead of filename.py.json
+            with open(meta_path, "w", encoding="utf-8") as meta_file:
+                json.dump(meta, meta_file)
+
+        # Copy all the additional imports
+        for import_name, import_path in local_imports:
+            if os.path.isfile(import_path):
+                full_path_local_import = os.path.join(importable_subdirectory, import_name + ".py")
+                if not os.path.exists(full_path_local_import):
+                    shutil.copyfile(import_path, full_path_local_import)
+            elif os.path.isdir(import_path):
+                full_path_local_import = os.path.join(importable_subdirectory, import_name)
+                if not os.path.exists(full_path_local_import):
+                    shutil.copytree(import_path, full_path_local_import)
+            else:
+                raise OSError(f"Error with local import at {import_path}")
+
+        # Copy aditional files like dataset infos file if needed
+        for file_name, original_path in additional_files:
+            destination_additional_path = os.path.join(importable_subdirectory, file_name)
+            if not os.path.exists(destination_additional_path) or not filecmp.cmp(
+                original_path, destination_additional_path
+            ):
+                shutil.copyfile(original_path, destination_additional_path)
+        return importable_local_file
+
+
+def infer_module_for_data_files(data_files_list: DataFilesList) -> Optional[str]:
+    extensions_counter = Counter(suffix[1:] for filepath in data_files_list for suffix in Path(filepath).suffixes)
+    if extensions_counter:
+        return _EXTENSION_TO_MODULE[extensions_counter.most_common(1)[0][0]]
+
+
+@dataclass
+class DatasetModuleFactoryResult:
+    module_path: str
+    hash: str
+    builder_kwargs: dict
+
+
+@dataclass
+class MetricModuleFactoryResult:
+    module_path: str
+    hash: str
+
+
+class _DatasetModuleFactory:
+    def get_module(self) -> DatasetModuleFactoryResult:
+        raise NotImplementedError
+
+
+class _MetricModuleFactory:
+    def get_module(self) -> MetricModuleFactoryResult:
+        raise NotImplementedError
+
+
+class CanonicalDatasetModuleFactory(_DatasetModuleFactory):
+    """Get the module of a canonical dataset. The dataset script is downloaded from GitHub."""
+
+    def __init__(
+        self,
+        name: str,
+        revision: Optional[Union[str, Version]] = None,
+        download_config: Optional[DownloadConfig] = None,
+        download_mode: Optional[GenerateMode] = None,
+        dynamic_modules_path: Optional[str] = None,
+    ):
+        self.name = name
+        self.revision = revision
+        self.download_config = download_config
+        self.download_mode = download_mode
+        self.dynamic_modules_path = dynamic_modules_path
+        assert self.name.count("/") == 0
+        increase_load_count(name, resource_type="dataset")
+
+    def download_dataset_script(self) -> str:
+        file_path = hf_github_url(path=self.name, name=self.name + ".py", revision=self.revision)
+        return cached_path(file_path, download_config=self.download_config)
+
+    def download_dataset_script_from_master(self) -> str:
+        file_path = hf_github_url(path=self.name, name=self.name + ".py", revision="master")
+        local_path = cached_path(file_path, download_config=self.download_config)
+        logger.warning(
+            f"Couldn't find a directory or a dataset named '{self.name}'. "
+            f"It was picked from the master branch on github instead at {file_path}"
+        )
+        return local_path
+
+    def download_dataset_infos_file(self, revision: Optional[str]) -> str:
+        dataset_infos = hf_github_url(path=self.name, name=config.DATASETDICT_INFOS_FILENAME, revision=revision)
+        # Download the dataset infos file if available
+        try:
+            return cached_path(
+                dataset_infos,
+                download_config=self.download_config,
+            )
+        except (FileNotFoundError, ConnectionError):
+            return None
+
+    def get_module(self) -> DatasetModuleFactoryResult:
+        # get script and other files
+        try:
+            local_path = self.download_dataset_script()
+            revision = self.revision
+        except FileNotFoundError:
+            if self.revision is not None:
+                raise
+            else:
+                local_path = self.download_dataset_script_from_master()
+                logger.warning(
+                    f"Dataset '{self.name}' not available for this version, using the one from the 'master' branch instead."
                 )
-        else:
-            out = matched_paths
-        if not out:
-            error_msg = f"Unable to resolve any data file that matches '{patterns}' at {Path(base_path).resolve()}"
-            if allowed_extensions is not None:
-                error_msg += f" with any supported extension {list(allowed_extensions)}"
-            raise FileNotFoundError(error_msg)
-        return out
-    elif isinstance(patterns, dict):
-        return {
-            k: _resolve_data_files_locally_or_by_urls(base_path, v, allowed_extensions=allowed_extensions)
-            for k, v in patterns.items()
+                revision = "master"
+        dataset_infos_path = self.download_dataset_infos_file(revision)
+        imports = get_imports(local_path)
+        local_imports = _download_additional_modules(
+            name=self.name,
+            base_path=hf_github_url(path=self.name, name="", revision=revision),
+            imports=imports,
+            download_config=self.download_config,
+        )
+        additional_files = [(config.DATASETDICT_INFOS_FILENAME, dataset_infos_path)] if dataset_infos_path else []
+        # the copy the script and the files in an importable directory
+        dynamic_modules_path = self.dynamic_modules_path if self.dynamic_modules_path else init_dynamic_modules()
+        importable_directory_path = os.path.join(dynamic_modules_path, "datasets", self.name)
+        Path(importable_directory_path).mkdir(parents=True, exist_ok=True)
+        (Path(importable_directory_path).parent / "__init__.py").touch(exist_ok=True)
+        hash = files_to_hash([local_path] + [loc[1] for loc in local_imports])
+        importable_local_file = _copy_script_and_other_resouces_in_importable_dir(
+            name=self.name,
+            importable_directory_path=importable_directory_path,
+            subdirrectory_name=hash,
+            original_local_path=local_path,
+            local_imports=local_imports,
+            additional_files=additional_files,
+            download_mode=self.download_mode,
+        )
+        logger.debug(f"Created importable dataset file at {importable_local_file}")
+        # make the new module to be noticed by the import system
+        importlib.invalidate_caches()
+        module_path = ".".join([os.path.basename(dynamic_modules_path), "datasets", self.name, hash, self.name])
+        builder_kwargs = {"hash": hash, "base_path": hf_github_url(self.name, "", revision=revision)}
+        return DatasetModuleFactoryResult(module_path, hash, builder_kwargs)
+
+
+class CanonicalMetricModuleFactory(_MetricModuleFactory):
+    """Get the module of a metric. The metric script is downloaded from GitHub."""
+
+    def __init__(
+        self,
+        name: str,
+        revision: Optional[Union[str, Version]] = None,
+        download_config: Optional[DownloadConfig] = None,
+        download_mode: Optional[GenerateMode] = None,
+        dynamic_modules_path: Optional[str] = None,
+    ):
+        self.name = name
+        self.revision = revision
+        self.download_config = download_config
+        self.download_mode = download_mode
+        self.dynamic_modules_path = dynamic_modules_path
+        assert self.name.count("/") == 0
+        increase_load_count(name, resource_type="metric")
+
+    def download_metric_script(self) -> str:
+        file_path = hf_github_url(path=self.name, name=self.name + ".py", revision=self.revision, dataset=False)
+        return cached_path(file_path, download_config=self.download_config)
+
+    def download_metric_script_from_master(self) -> str:
path: src/datasets/load.py
text: @user1:
The same as above for the naming of these 2 methods.

Also note that their implementation is the same (apart from the param `dataset=False`). But not too important as this is only repeated twice.

diff_hunk: @@ -0,0 +1,328 @@
+import glob
+import os
+from functools import partial
+from pathlib import Path, PurePath
+from typing import Dict, List, Optional, Tuple, Union
+
+import huggingface_hub
+from tqdm.contrib.concurrent import thread_map
+
+from .splits import Split
+from .utils import logging
+from .utils.file_utils import is_relative_path, is_remote_url, request_etag
+from .utils.tqdm_utils import tqdm
+
+
+DEFAULT_SPLIT = str(Split.TRAIN)
+
+
+logger = logging.get_logger(__name__)
+
+
+class Url(str):
+    pass
+
+
+def _sanitize_patterns(patterns: Union[Dict, List, str, None]) -> Dict[str, Union[List[str], "DataFilesList"]]:
+    """
+    Take the data_files patterns from the user, and format them into a dictionary.
+    Each key is the name of the split, and each value is a list of data files patterns (paths or urls).
+    The default split is "train".
+
+    Returns:
+        patterns: dictionary of split_name -> list_of _atterns
+    """
+    if patterns is None:
+        return {DEFAULT_SPLIT: ["*"]}
+    if isinstance(patterns, dict):
+        return {str(key): value if isinstance(value, list) else [value] for key, value in patterns.items()}
+    elif isinstance(patterns, str):
+        return {DEFAULT_SPLIT: [patterns]}
+    else:
+        return {DEFAULT_SPLIT: list(patterns)}
+
+
+def _resolve_single_pattern_locally(
+    base_path: str, pattern: str, allowed_extensions: Optional[List[str]] = None
+) -> List[Path]:
+    """
+    Return the absolute paths to all the files that match the given patterns.
+    It also supports absolute paths in patterns.
+    If an URL is passed, it is returned as is.
+    """
+    data_files_ignore = ["README.md", "config.json", "dataset_infos.json", "dummy_data.zip"]
+    if is_relative_path(pattern):
+        glob_iter = list(Path(base_path).rglob(pattern))
+    else:
+        glob_iter = [Path(filepath) for filepath in glob.glob(pattern)]
+
+    matched_paths = [
+        filepath.resolve()
+        for filepath in glob_iter
+        if filepath.name not in data_files_ignore and not filepath.name.startswith(".") and filepath.is_file()
+    ]
+    if allowed_extensions is not None:
+        out = [
+            filepath
+            for filepath in matched_paths
+            if any(suffix[1:] in allowed_extensions for suffix in filepath.suffixes)
+        ]
+        if len(out) < len(matched_paths):
+            invalid_matched_files = list(set(matched_paths) - set(out))
+            logger.info(
+                f"Some files matched the pattern '{pattern}' at {Path(base_path).resolve()} but don't have valid data file extensions: {invalid_matched_files}"
+            )
+    else:
+        out = matched_paths
+    if not out:
+        error_msg = f"Unable to resolve any data file that matches '{pattern}' at {Path(base_path).resolve()}"
+        if allowed_extensions is not None:
+            error_msg += f" with any supported extension {list(allowed_extensions)}"
+        raise FileNotFoundError(error_msg)
+    return out
+
+
+def resolve_patterns_locally_or_by_urls(
+    base_path: str, patterns: List[str], allowed_extensions: Optional[List[str]] = None
+) -> List[Union[Path, Url]]:
+    """
+    Resolve the paths and URLs of the data files from the patterns passed by the user.
+    URLs are just returned as is.
+
+    Examples:
+
+        >>> import huggingface_hub
+        >>> from datasets.data_files import resolve_patterns_in_dataset_repository
+        >>> base_path = /Users/username/Desktop/hf/datasets
+        >>> resolve_patterns_locally_or_by_urls(base_path, ["src/**/*.yaml"])
+        [PosixPath('/Users/quentinlhoest/Desktop/hf/shirte/datasets/src/datasets/utils/resources/readme_structure.yaml')]
+
+    Args:
+        base_path (str): Base path to use when resolving relative paths.
+        patterns (List[str]): Unix patterns or paths or URLs of the data files to resolve.
+            The paths can be absolute or relative to base_path.
+        allowed_extensions (Optional[list], optional): White-list of file extensions to use. Defaults to None (all extensions).
+            For example: allowed_extensions=["csv", "json", "txt", "parquet"]
+
+    Returns:
+        List[Union[Path, Url]]: List of paths or URLs to the local or remote files that match the patterns.
+    """
+    data_files = []
+    for pattern in patterns:
+        if is_remote_url(pattern):
+            data_files.append(Url(pattern))
+        else:
+            data_files.extend(_resolve_single_pattern_locally(base_path, pattern, allowed_extensions))
+    return data_files
+
+
+def _exec_patterns_in_dataset_repository(
+    dataset_info: huggingface_hub.hf_api.DatasetInfo,
+    patterns: List[str],
+    allowed_extensions: Optional[list] = None,
+) -> List[PurePath]:
+    data_files_ignore = ["README.md", "config.json", "dataset_infos.json", "dummy_data.zip"]
+    all_data_files = [
+        PurePath("/" + dataset_file.rfilename) for dataset_file in dataset_info.siblings
+    ]  # add a / at the beginning to make the pattern **/* match files at the root
+    filenames = []
+    for pattern in patterns:
+        matched_paths = [
+            filepath.relative_to("/")
+            for filepath in all_data_files
+            if filepath.name not in data_files_ignore and not filepath.name.startswith(".") and filepath.match(pattern)
+        ]
+        if allowed_extensions is not None:
+            out = [
+                filepath
+                for filepath in matched_paths
+                if any(suffix[1:] in allowed_extensions for suffix in filepath.suffixes)
+            ]
+            if len(out) < len(matched_paths):
+                invalid_matched_files = list(set(matched_paths) - set(out))
+                logger.info(
+                    f"Some files matched the pattern {pattern} in dataset repository {dataset_info.id} but don't have valid data file extensions: {invalid_matched_files}"
+                )
+        else:
+            out = matched_paths
+        if not out:
+            error_msg = f"Unable to resolve data_file {pattern} in dataset repository {dataset_info.id}"
+            if allowed_extensions is not None:
+                error_msg += f" with any supported extension {list(allowed_extensions)}"
+            raise FileNotFoundError(error_msg)
+        filenames.extend(out)
+    return filenames
+
+
+def resolve_patterns_in_dataset_repository(
+    dataset_info: huggingface_hub.hf_api.DatasetInfo,
+    patterns: List[str],
+    allowed_extensions: Optional[list] = None,
+) -> List[Url]:
+    """
+    Resolve the URLs of the data files from the patterns passed by the user.
+
+    Examples:
+
+        >>> import huggingface_hub
+        >>> from datasets.data_files import resolve_patterns_in_dataset_repository
+        >>> dataset_info = huggingface_hub.HfApi().dataset_info("lhoestq/demo1")
+        >>> resolve_patterns_in_dataset_repository(dataset_info, ["*.csv"])
+        ['https://huggingface.co/datasets/lhoestq/demo1/resolve/0ca0d9f35b390ad11516095aeb27fd30cfe72578/data/test.csv',
+        'https://huggingface.co/datasets/lhoestq/demo1/resolve/0ca0d9f35b390ad11516095aeb27fd30cfe72578/data/train.csv']
+
+    Args:
+        dataset_info (huggingface_hub.hf_api.DatasetInfo): dataset info obtained using the hugginggace_hub.HfApi
+        patterns (List[str]): Unix patterns or paths of the files in the dataset repository.
+            The paths should be relative to the root of the repository.
+        allowed_extensions (Optional[list], optional): White-list of file extensions to use. Defaults to None (all extensions).
+            For example: allowed_extensions=["csv", "json", "txt", "parquet"]
+
+    Returns:
+        List[Url]: List of URLs to the files in the dataset repository that match the patterns.
+    """
+    data_files_urls: List[Url] = []
+    for filename in _exec_patterns_in_dataset_repository(dataset_info, patterns, allowed_extensions):
+        data_files_urls.append(
+            Url(
+                huggingface_hub.hf_hub_url(
+                    dataset_info.id, filename.as_posix(), revision=dataset_info.sha, repo_type="dataset"
+                )
+            )
+        )
+    return data_files_urls
+
+
+def _get_single_origin_metadata_locally_or_by_urls(
+    data_file: Union[Path, Url], use_auth_token: Optional[Union[bool, str]] = None
+) -> Tuple[str]:
+    if isinstance(data_file, Url):
+        data_file = str(data_file)
+        return (request_etag(data_file, use_auth_token=use_auth_token),)
+    else:
+        data_file = str(data_file.resolve())
+        return (str(os.path.getmtime(data_file)),)
+
+
+def _get_origin_metadata_locally_or_by_urls(
+    data_files: List[Union[Path, Url]], max_workers=64, use_auth_token: Optional[Union[bool, str]] = None
+) -> Tuple[str]:
+    return thread_map(
+        partial(_get_single_origin_metadata_locally_or_by_urls, use_auth_token=use_auth_token),
+        data_files,
+        max_workers=max_workers,
+        tqdm_class=tqdm,
+        desc="Resolving data files",
+        disable=len(data_files) <= 16 or logging.get_verbosity() == logging.NOTSET,
+    )
+
+
+class DataFilesList(List[Union[Path, Url]]):
+    """
+    List of data files (absolute local paths or URLs).
+    It has two construction methods given the user's data files patterns :
+    - ``from_hf_repo``: resolve patterns inside a dataset repository
+    - ``from_local_or_remote``: resolve patterns from a local path
+
+    Moreover DataFilesList has an additional attribute ``origin_metadata``.
+    It can store:
+    - the last modified time of local files
+    - ETag of remote files
+    - commit sha of a dataset repository
+
+    Thanks to this additional attribute, it is possible to hash the list
+    and get a different hash if and only if at least one file changed.
+    This is useful for caching Dataset objects that are obtained from a list of data files.
+    """
+
+    def __init__(self, data_files: List[Union[Path, Url]], origin_metadata: List[Tuple[str]]):
+        super().__init__(data_files)
+        self.origin_metadata = origin_metadata
+
+    @staticmethod
path: src/datasets/data_files.py
text: @user2:
What about `@user1` instead and:
```python
return cls(data_files, origin_metadata)
```

diff_hunk: @@ -226,604 +236,1042 @@ def get_imports(file_path: str):
     return imports
 
 
-def _resolve_data_files_locally_or_by_urls(
-    base_path: str, patterns: Union[str, List[str], Dict], allowed_extensions: Optional[list] = None
-) -> Union[List[Path], Dict]:
+def _download_additional_modules(
+    name: str, base_path: str, imports: Tuple[str, str, str, str], download_config: Optional[DownloadConfig]
+) -> List[Tuple[str, str]]:
     """
-    Return the absolute paths to all the files that match the given patterns.
-    It also supports absolute paths in patterns.
-    If an URL is passed, it is returned as is."""
-    data_files_ignore = ["README.md", "config.json", "dataset_infos.json", "dummy_data.zip"]
-    if isinstance(patterns, str):
-        if is_remote_url(patterns):
-            return [patterns]
-        if is_relative_path(patterns):
-            glob_iter = list(Path(base_path).rglob(patterns))
+    Download additional module for a module <name>.py at URL (or local path) <base_path>/<name>.py
+    The imports must have been parsed first using ``get_imports``.
+
+    If some modules need to be installed with pip, an error is raised showing how to install them.
+    This function return the list of downloaded modules as tuples (import_name, module_file_path).
+
+    The downloaded modules can then be moved into an importable directory with ``_copy_script_and_other_resouces_in_importable_dir``.
+    """
+    local_imports = []
+    library_imports = []
+    for import_type, import_name, import_path, sub_directory in imports:
+        if import_type == "library":
+            library_imports.append((import_name, import_path))  # Import from a library
+            continue
+
+        if import_name == name:
+            raise ValueError(
+                f"Error in the {name} script, importing relative {import_name} module "
+                f"but {import_name} is the name of the script. "
+                f"Please change relative import {import_name} to another name and add a '# From: URL_OR_PATH' "
+                f"comment pointing to the original relative import file path."
+            )
+        if import_type == "internal":
+            url_or_filename = url_or_path_join(base_path, import_path + ".py")
+        elif import_type == "external":
+            url_or_filename = import_path
         else:
-            glob_iter = [Path(filepath) for filepath in glob.glob(patterns)]
+            raise ValueError("Wrong import_type")
 
-        matched_paths = [
-            filepath.resolve()
-            for filepath in glob_iter
-            if filepath.name not in data_files_ignore and not filepath.name.startswith(".") and filepath.is_file()
-        ]
-        if allowed_extensions is not None:
-            out = [
-                filepath
-                for filepath in matched_paths
-                if any(suffix[1:] in allowed_extensions for suffix in filepath.suffixes)
-            ]
-            if len(out) < len(matched_paths):
-                invalid_matched_files = list(set(matched_paths) - set(out))
-                logger.info(
-                    f"Some files matched the pattern '{patterns}' at {Path(base_path).resolve()} but don't have valid data file extensions: {invalid_matched_files}"
+        local_import_path = cached_path(
+            url_or_filename,
+            download_config=download_config,
+        )
+        if sub_directory is not None:
+            local_import_path = os.path.join(local_import_path, sub_directory)
+        local_imports.append((import_name, local_import_path))
+
+    # Check library imports
+    needs_to_be_installed = []
+    for library_import_name, library_import_path in library_imports:
+        try:
+            lib = importlib.import_module(library_import_name)  # noqa F841
+        except ImportError:
+            needs_to_be_installed.append((library_import_name, library_import_path))
+    if needs_to_be_installed:
+        raise ImportError(
+            f"To be able to use {name}, you need to install the following dependencies"
+            f"{[lib_name for lib_name, lib_path in needs_to_be_installed]} using 'pip install "
+            f"{' '.join([lib_path for lib_name, lib_path in needs_to_be_installed])}' for instance'"
+        )
+    return local_imports
+
+
+def _copy_script_and_other_resouces_in_importable_dir(
+    name: str,
+    importable_directory_path: str,
+    subdirrectory_name: str,
+    original_local_path: str,
+    local_imports: List[Tuple[str, str]],
+    additional_files: List[Tuple[str, str]],
+    download_mode: Optional[GenerateMode],
+) -> str:
+    """Copy a script and its required imports to an importable directory
+
+    Args:
+        name (str): name of the resource to load
+        importable_directory_path (str): path to the loadable folder in the dynamic modules directory
+        subdirrectory_name (str): name of the subdirectory in importable_directory_path in which to place the script
+        original_local_path (str): local path to the resource script
+        local_imports (List[Tuple[str, str]]): list of (destination_filename, import_file_to_copy)
+        additional_files (List[Tuple[str, str]]): list of (destination_filename, additional_file_to_copy)
+        download_mode (Optional[GenerateMode]): download mode
+
+    Return:
+        importable_local_file: path to an importable module with importlib.import_module
+    """
+
+    # Define a directory with a unique name in our dataset or metric folder
+    # path is: ./datasets|metrics/dataset|metric_name/hash_from_code/script.py
+    # we use a hash as subdirrectory_name to be able to have multiple versions of a dataset/metric processing file together
+    importable_subdirectory = os.path.join(importable_directory_path, subdirrectory_name)
+    importable_local_file = os.path.join(importable_subdirectory, name + ".py")
+
+    # Prevent parallel disk operations
+    lock_path = importable_directory_path + ".lock"
+    with FileLock(lock_path):
+        # Create main dataset/metrics folder if needed
+        if download_mode == GenerateMode.FORCE_REDOWNLOAD and os.path.exists(importable_directory_path):
+            shutil.rmtree(importable_directory_path)
+        os.makedirs(importable_directory_path, exist_ok=True)
+
+        # add an __init__ file to the main dataset folder if needed
+        init_file_path = os.path.join(importable_directory_path, "__init__.py")
+        if not os.path.exists(init_file_path):
+            with open(init_file_path, "w"):
+                pass
+
+        # Create hash dataset folder if needed
+        os.makedirs(importable_subdirectory, exist_ok=True)
+        # add an __init__ file to the hash dataset folder if needed
+        init_file_path = os.path.join(importable_subdirectory, "__init__.py")
+        if not os.path.exists(init_file_path):
+            with open(init_file_path, "w"):
+                pass
+
+        # Copy dataset.py file in hash folder if needed
+        if not os.path.exists(importable_local_file):
+            shutil.copyfile(original_local_path, importable_local_file)
+
+        # Record metadata associating original dataset path with local unique folder
+        meta_path = importable_local_file.split(".py")[0] + ".json"
+        if not os.path.exists(meta_path):
+            meta = {"original file path": original_local_path, "local file path": importable_local_file}
+            # the filename is *.py in our case, so better rename to filenam.json instead of filename.py.json
+            with open(meta_path, "w", encoding="utf-8") as meta_file:
+                json.dump(meta, meta_file)
+
+        # Copy all the additional imports
+        for import_name, import_path in local_imports:
+            if os.path.isfile(import_path):
+                full_path_local_import = os.path.join(importable_subdirectory, import_name + ".py")
+                if not os.path.exists(full_path_local_import):
+                    shutil.copyfile(import_path, full_path_local_import)
+            elif os.path.isdir(import_path):
+                full_path_local_import = os.path.join(importable_subdirectory, import_name)
+                if not os.path.exists(full_path_local_import):
+                    shutil.copytree(import_path, full_path_local_import)
+            else:
+                raise OSError(f"Error with local import at {import_path}")
+
+        # Copy aditional files like dataset infos file if needed
+        for file_name, original_path in additional_files:
+            destination_additional_path = os.path.join(importable_subdirectory, file_name)
+            if not os.path.exists(destination_additional_path) or not filecmp.cmp(
+                original_path, destination_additional_path
+            ):
+                shutil.copyfile(original_path, destination_additional_path)
+        return importable_local_file
+
+
+def infer_module_for_data_files(data_files_list: DataFilesList) -> Optional[str]:
+    extensions_counter = Counter(suffix[1:] for filepath in data_files_list for suffix in Path(filepath).suffixes)
+    if extensions_counter:
+        return _EXTENSION_TO_MODULE[extensions_counter.most_common(1)[0][0]]
+
+
+@dataclass
+class DatasetModuleFactoryResult:
+    module_path: str
+    hash: str
+    builder_kwargs: dict
+
+
+@dataclass
+class MetricModuleFactoryResult:
path: src/datasets/load.py
text: @user1:
Also renaming to `MetricModule`?

diff_hunk: @@ -226,604 +236,1042 @@ def get_imports(file_path: str):
     return imports
 
 
-def _resolve_data_files_locally_or_by_urls(
-    base_path: str, patterns: Union[str, List[str], Dict], allowed_extensions: Optional[list] = None
-) -> Union[List[Path], Dict]:
+def _download_additional_modules(
+    name: str, base_path: str, imports: Tuple[str, str, str, str], download_config: Optional[DownloadConfig]
+) -> List[Tuple[str, str]]:
     """
-    Return the absolute paths to all the files that match the given patterns.
-    It also supports absolute paths in patterns.
-    If an URL is passed, it is returned as is."""
-    data_files_ignore = ["README.md", "config.json", "dataset_infos.json", "dummy_data.zip"]
-    if isinstance(patterns, str):
-        if is_remote_url(patterns):
-            return [patterns]
-        if is_relative_path(patterns):
-            glob_iter = list(Path(base_path).rglob(patterns))
+    Download additional module for a module <name>.py at URL (or local path) <base_path>/<name>.py
+    The imports must have been parsed first using ``get_imports``.
+
+    If some modules need to be installed with pip, an error is raised showing how to install them.
+    This function return the list of downloaded modules as tuples (import_name, module_file_path).
+
+    The downloaded modules can then be moved into an importable directory with ``_copy_script_and_other_resouces_in_importable_dir``.
+    """
+    local_imports = []
+    library_imports = []
+    for import_type, import_name, import_path, sub_directory in imports:
+        if import_type == "library":
+            library_imports.append((import_name, import_path))  # Import from a library
+            continue
+
+        if import_name == name:
+            raise ValueError(
+                f"Error in the {name} script, importing relative {import_name} module "
+                f"but {import_name} is the name of the script. "
+                f"Please change relative import {import_name} to another name and add a '# From: URL_OR_PATH' "
+                f"comment pointing to the original relative import file path."
+            )
+        if import_type == "internal":
+            url_or_filename = url_or_path_join(base_path, import_path + ".py")
+        elif import_type == "external":
+            url_or_filename = import_path
         else:
-            glob_iter = [Path(filepath) for filepath in glob.glob(patterns)]
+            raise ValueError("Wrong import_type")
 
-        matched_paths = [
-            filepath.resolve()
-            for filepath in glob_iter
-            if filepath.name not in data_files_ignore and not filepath.name.startswith(".") and filepath.is_file()
-        ]
-        if allowed_extensions is not None:
-            out = [
-                filepath
-                for filepath in matched_paths
-                if any(suffix[1:] in allowed_extensions for suffix in filepath.suffixes)
-            ]
-            if len(out) < len(matched_paths):
-                invalid_matched_files = list(set(matched_paths) - set(out))
-                logger.info(
-                    f"Some files matched the pattern '{patterns}' at {Path(base_path).resolve()} but don't have valid data file extensions: {invalid_matched_files}"
+        local_import_path = cached_path(
+            url_or_filename,
+            download_config=download_config,
+        )
+        if sub_directory is not None:
+            local_import_path = os.path.join(local_import_path, sub_directory)
+        local_imports.append((import_name, local_import_path))
+
+    # Check library imports
+    needs_to_be_installed = []
+    for library_import_name, library_import_path in library_imports:
+        try:
+            lib = importlib.import_module(library_import_name)  # noqa F841
+        except ImportError:
+            needs_to_be_installed.append((library_import_name, library_import_path))
+    if needs_to_be_installed:
+        raise ImportError(
+            f"To be able to use {name}, you need to install the following dependencies"
+            f"{[lib_name for lib_name, lib_path in needs_to_be_installed]} using 'pip install "
+            f"{' '.join([lib_path for lib_name, lib_path in needs_to_be_installed])}' for instance'"
+        )
+    return local_imports
+
+
+def _copy_script_and_other_resouces_in_importable_dir(
+    name: str,
+    importable_directory_path: str,
+    subdirrectory_name: str,
+    original_local_path: str,
+    local_imports: List[Tuple[str, str]],
+    additional_files: List[Tuple[str, str]],
+    download_mode: Optional[GenerateMode],
+) -> str:
+    """Copy a script and its required imports to an importable directory
+
+    Args:
+        name (str): name of the resource to load
+        importable_directory_path (str): path to the loadable folder in the dynamic modules directory
+        subdirrectory_name (str): name of the subdirectory in importable_directory_path in which to place the script
+        original_local_path (str): local path to the resource script
+        local_imports (List[Tuple[str, str]]): list of (destination_filename, import_file_to_copy)
+        additional_files (List[Tuple[str, str]]): list of (destination_filename, additional_file_to_copy)
+        download_mode (Optional[GenerateMode]): download mode
+
+    Return:
+        importable_local_file: path to an importable module with importlib.import_module
+    """
+
+    # Define a directory with a unique name in our dataset or metric folder
+    # path is: ./datasets|metrics/dataset|metric_name/hash_from_code/script.py
+    # we use a hash as subdirrectory_name to be able to have multiple versions of a dataset/metric processing file together
+    importable_subdirectory = os.path.join(importable_directory_path, subdirrectory_name)
+    importable_local_file = os.path.join(importable_subdirectory, name + ".py")
+
+    # Prevent parallel disk operations
+    lock_path = importable_directory_path + ".lock"
+    with FileLock(lock_path):
+        # Create main dataset/metrics folder if needed
+        if download_mode == GenerateMode.FORCE_REDOWNLOAD and os.path.exists(importable_directory_path):
+            shutil.rmtree(importable_directory_path)
+        os.makedirs(importable_directory_path, exist_ok=True)
+
+        # add an __init__ file to the main dataset folder if needed
+        init_file_path = os.path.join(importable_directory_path, "__init__.py")
+        if not os.path.exists(init_file_path):
+            with open(init_file_path, "w"):
+                pass
+
+        # Create hash dataset folder if needed
+        os.makedirs(importable_subdirectory, exist_ok=True)
+        # add an __init__ file to the hash dataset folder if needed
+        init_file_path = os.path.join(importable_subdirectory, "__init__.py")
+        if not os.path.exists(init_file_path):
+            with open(init_file_path, "w"):
+                pass
+
+        # Copy dataset.py file in hash folder if needed
+        if not os.path.exists(importable_local_file):
+            shutil.copyfile(original_local_path, importable_local_file)
+
+        # Record metadata associating original dataset path with local unique folder
+        meta_path = importable_local_file.split(".py")[0] + ".json"
+        if not os.path.exists(meta_path):
+            meta = {"original file path": original_local_path, "local file path": importable_local_file}
+            # the filename is *.py in our case, so better rename to filenam.json instead of filename.py.json
+            with open(meta_path, "w", encoding="utf-8") as meta_file:
+                json.dump(meta, meta_file)
+
+        # Copy all the additional imports
+        for import_name, import_path in local_imports:
+            if os.path.isfile(import_path):
+                full_path_local_import = os.path.join(importable_subdirectory, import_name + ".py")
+                if not os.path.exists(full_path_local_import):
+                    shutil.copyfile(import_path, full_path_local_import)
+            elif os.path.isdir(import_path):
+                full_path_local_import = os.path.join(importable_subdirectory, import_name)
+                if not os.path.exists(full_path_local_import):
+                    shutil.copytree(import_path, full_path_local_import)
+            else:
+                raise OSError(f"Error with local import at {import_path}")
+
+        # Copy aditional files like dataset infos file if needed
+        for file_name, original_path in additional_files:
+            destination_additional_path = os.path.join(importable_subdirectory, file_name)
+            if not os.path.exists(destination_additional_path) or not filecmp.cmp(
+                original_path, destination_additional_path
+            ):
+                shutil.copyfile(original_path, destination_additional_path)
+        return importable_local_file
+
+
+def infer_module_for_data_files(data_files_list: DataFilesList) -> Optional[str]:
+    extensions_counter = Counter(suffix[1:] for filepath in data_files_list for suffix in Path(filepath).suffixes)
+    if extensions_counter:
+        return _EXTENSION_TO_MODULE[extensions_counter.most_common(1)[0][0]]
+
+
+@dataclass
+class DatasetModuleFactoryResult:
+    module_path: str
+    hash: str
+    builder_kwargs: dict
+
+
+@dataclass
+class MetricModuleFactoryResult:
+    module_path: str
+    hash: str
+
+
+class _DatasetModuleFactory:
+    def get_module(self) -> DatasetModuleFactoryResult:
+        raise NotImplementedError
+
+
+class _MetricModuleFactory:
+    def get_module(self) -> MetricModuleFactoryResult:
+        raise NotImplementedError
+
+
+class CanonicalDatasetModuleFactory(_DatasetModuleFactory):
+    """Get the module of a canonical dataset. The dataset script is downloaded from GitHub."""
+
+    def __init__(
+        self,
+        name: str,
+        revision: Optional[Union[str, Version]] = None,
+        download_config: Optional[DownloadConfig] = None,
+        download_mode: Optional[GenerateMode] = None,
+        dynamic_modules_path: Optional[str] = None,
+    ):
+        self.name = name
+        self.revision = revision
+        self.download_config = download_config
+        self.download_mode = download_mode
+        self.dynamic_modules_path = dynamic_modules_path
+        assert self.name.count("/") == 0
+        increase_load_count(name, resource_type="dataset")
+
+    def download_dataset_script(self) -> str:
+        file_path = hf_github_url(path=self.name, name=self.name + ".py", revision=self.revision)
+        return cached_path(file_path, download_config=self.download_config)
+
+    def download_dataset_script_from_master(self) -> str:
path: src/datasets/load.py
text: @user1:
Following with the philosophy of the factory design pattern (a common interface with different implementations), I would call these 2 methods:
- `download_script` or `download_loading_script`
- `download_script_from_master` or `download_loading_script_from_master`
analogously to the method `get_module` (we don't call it `get_dataset_module`).

@author:
I renamed it to `download_loading_script`, and removed the method "from_master" (see previous comment about merging the two methods)

diff_hunk: @@ -226,604 +236,1042 @@ def get_imports(file_path: str):
     return imports
 
 
-def _resolve_data_files_locally_or_by_urls(
-    base_path: str, patterns: Union[str, List[str], Dict], allowed_extensions: Optional[list] = None
-) -> Union[List[Path], Dict]:
+def _download_additional_modules(
+    name: str, base_path: str, imports: Tuple[str, str, str, str], download_config: Optional[DownloadConfig]
+) -> List[Tuple[str, str]]:
     """
-    Return the absolute paths to all the files that match the given patterns.
-    It also supports absolute paths in patterns.
-    If an URL is passed, it is returned as is."""
-    data_files_ignore = ["README.md", "config.json", "dataset_infos.json", "dummy_data.zip"]
-    if isinstance(patterns, str):
-        if is_remote_url(patterns):
-            return [patterns]
-        if is_relative_path(patterns):
-            glob_iter = list(Path(base_path).rglob(patterns))
+    Download additional module for a module <name>.py at URL (or local path) <base_path>/<name>.py
+    The imports must have been parsed first using ``get_imports``.
+
+    If some modules need to be installed with pip, an error is raised showing how to install them.
+    This function return the list of downloaded modules as tuples (import_name, module_file_path).
+
+    The downloaded modules can then be moved into an importable directory with ``_copy_script_and_other_resouces_in_importable_dir``.
+    """
+    local_imports = []
+    library_imports = []
+    for import_type, import_name, import_path, sub_directory in imports:
+        if import_type == "library":
+            library_imports.append((import_name, import_path))  # Import from a library
+            continue
+
+        if import_name == name:
+            raise ValueError(
+                f"Error in the {name} script, importing relative {import_name} module "
+                f"but {import_name} is the name of the script. "
+                f"Please change relative import {import_name} to another name and add a '# From: URL_OR_PATH' "
+                f"comment pointing to the original relative import file path."
+            )
+        if import_type == "internal":
+            url_or_filename = url_or_path_join(base_path, import_path + ".py")
+        elif import_type == "external":
+            url_or_filename = import_path
         else:
-            glob_iter = [Path(filepath) for filepath in glob.glob(patterns)]
+            raise ValueError("Wrong import_type")
 
-        matched_paths = [
-            filepath.resolve()
-            for filepath in glob_iter
-            if filepath.name not in data_files_ignore and not filepath.name.startswith(".") and filepath.is_file()
-        ]
-        if allowed_extensions is not None:
-            out = [
-                filepath
-                for filepath in matched_paths
-                if any(suffix[1:] in allowed_extensions for suffix in filepath.suffixes)
-            ]
-            if len(out) < len(matched_paths):
-                invalid_matched_files = list(set(matched_paths) - set(out))
-                logger.info(
-                    f"Some files matched the pattern '{patterns}' at {Path(base_path).resolve()} but don't have valid data file extensions: {invalid_matched_files}"
+        local_import_path = cached_path(
+            url_or_filename,
+            download_config=download_config,
+        )
+        if sub_directory is not None:
+            local_import_path = os.path.join(local_import_path, sub_directory)
+        local_imports.append((import_name, local_import_path))
+
+    # Check library imports
+    needs_to_be_installed = []
+    for library_import_name, library_import_path in library_imports:
+        try:
+            lib = importlib.import_module(library_import_name)  # noqa F841
+        except ImportError:
+            needs_to_be_installed.append((library_import_name, library_import_path))
+    if needs_to_be_installed:
+        raise ImportError(
+            f"To be able to use {name}, you need to install the following dependencies"
+            f"{[lib_name for lib_name, lib_path in needs_to_be_installed]} using 'pip install "
+            f"{' '.join([lib_path for lib_name, lib_path in needs_to_be_installed])}' for instance'"
+        )
+    return local_imports
+
+
+def _copy_script_and_other_resouces_in_importable_dir(
+    name: str,
+    importable_directory_path: str,
+    subdirrectory_name: str,
+    original_local_path: str,
+    local_imports: List[Tuple[str, str]],
+    additional_files: List[Tuple[str, str]],
+    download_mode: Optional[GenerateMode],
+) -> str:
+    """Copy a script and its required imports to an importable directory
+
+    Args:
+        name (str): name of the resource to load
+        importable_directory_path (str): path to the loadable folder in the dynamic modules directory
+        subdirrectory_name (str): name of the subdirectory in importable_directory_path in which to place the script
+        original_local_path (str): local path to the resource script
+        local_imports (List[Tuple[str, str]]): list of (destination_filename, import_file_to_copy)
+        additional_files (List[Tuple[str, str]]): list of (destination_filename, additional_file_to_copy)
+        download_mode (Optional[GenerateMode]): download mode
+
+    Return:
+        importable_local_file: path to an importable module with importlib.import_module
+    """
+
+    # Define a directory with a unique name in our dataset or metric folder
+    # path is: ./datasets|metrics/dataset|metric_name/hash_from_code/script.py
+    # we use a hash as subdirrectory_name to be able to have multiple versions of a dataset/metric processing file together
+    importable_subdirectory = os.path.join(importable_directory_path, subdirrectory_name)
+    importable_local_file = os.path.join(importable_subdirectory, name + ".py")
+
+    # Prevent parallel disk operations
+    lock_path = importable_directory_path + ".lock"
+    with FileLock(lock_path):
+        # Create main dataset/metrics folder if needed
+        if download_mode == GenerateMode.FORCE_REDOWNLOAD and os.path.exists(importable_directory_path):
+            shutil.rmtree(importable_directory_path)
+        os.makedirs(importable_directory_path, exist_ok=True)
+
+        # add an __init__ file to the main dataset folder if needed
+        init_file_path = os.path.join(importable_directory_path, "__init__.py")
+        if not os.path.exists(init_file_path):
+            with open(init_file_path, "w"):
+                pass
+
+        # Create hash dataset folder if needed
+        os.makedirs(importable_subdirectory, exist_ok=True)
+        # add an __init__ file to the hash dataset folder if needed
+        init_file_path = os.path.join(importable_subdirectory, "__init__.py")
+        if not os.path.exists(init_file_path):
+            with open(init_file_path, "w"):
+                pass
+
+        # Copy dataset.py file in hash folder if needed
+        if not os.path.exists(importable_local_file):
+            shutil.copyfile(original_local_path, importable_local_file)
+
+        # Record metadata associating original dataset path with local unique folder
+        meta_path = importable_local_file.split(".py")[0] + ".json"
+        if not os.path.exists(meta_path):
+            meta = {"original file path": original_local_path, "local file path": importable_local_file}
+            # the filename is *.py in our case, so better rename to filenam.json instead of filename.py.json
+            with open(meta_path, "w", encoding="utf-8") as meta_file:
+                json.dump(meta, meta_file)
+
+        # Copy all the additional imports
+        for import_name, import_path in local_imports:
+            if os.path.isfile(import_path):
+                full_path_local_import = os.path.join(importable_subdirectory, import_name + ".py")
+                if not os.path.exists(full_path_local_import):
+                    shutil.copyfile(import_path, full_path_local_import)
+            elif os.path.isdir(import_path):
+                full_path_local_import = os.path.join(importable_subdirectory, import_name)
+                if not os.path.exists(full_path_local_import):
+                    shutil.copytree(import_path, full_path_local_import)
+            else:
+                raise OSError(f"Error with local import at {import_path}")
+
+        # Copy aditional files like dataset infos file if needed
+        for file_name, original_path in additional_files:
+            destination_additional_path = os.path.join(importable_subdirectory, file_name)
+            if not os.path.exists(destination_additional_path) or not filecmp.cmp(
+                original_path, destination_additional_path
+            ):
+                shutil.copyfile(original_path, destination_additional_path)
+        return importable_local_file
+
+
+def infer_module_for_data_files(data_files_list: DataFilesList) -> Optional[str]:
+    extensions_counter = Counter(suffix[1:] for filepath in data_files_list for suffix in Path(filepath).suffixes)
+    if extensions_counter:
+        return _EXTENSION_TO_MODULE[extensions_counter.most_common(1)[0][0]]
+
+
+@dataclass
+class DatasetModuleFactoryResult:
path: src/datasets/load.py
text: @user1:
Maybe renaming to `DatasetModule`?