File size: 1,509 Bytes
058c80a 5b41acf 058c80a 54b3a72 646d90e 54b3a72 646d90e 24df49f 54b3a72 646d90e 54b3a72 5b41acf 54b3a72 5b41acf 54b3a72 058c80a 54b3a72 5b41acf 646d90e 54b3a72 5b41acf 24df49f 058c80a 54b3a72 5b41acf 54b3a72 058c80a 5b41acf 54b3a72 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 |
from json.decoder import JSONDecodeError
from typing import Any, Dict, Optional
from .artifact import Artifact, UnitxtArtifactNotFoundError, fetch_artifact
from .logging_utils import get_logger
from .parsing_utils import parse_key_equals_value_string_to_dict
from .register import _reset_env_local_catalogs, register_all_artifacts
from .settings_utils import get_settings
from .standard import DatasetRecipe
logger = get_logger()
settings = get_settings()
def fetch(artifact_name: str, overwrite_kwargs: Optional[Dict[str, Any]]=None):
try:
artifact, _ = fetch_artifact(artifact_name, overwrite_kwargs=overwrite_kwargs)
return artifact
except (UnitxtArtifactNotFoundError, JSONDecodeError):
return None
def parse(query: str) -> dict:
return parse_key_equals_value_string_to_dict(query)
def get_dataset_artifact(dataset, overwrite_kwargs: Optional[Dict[str, Any]]=None):
if isinstance(dataset, DatasetRecipe):
return dataset
assert isinstance(
dataset, str
), "dataset should be string description of recipe, or recipe object."
_reset_env_local_catalogs()
register_all_artifacts()
recipe = fetch(dataset, overwrite_kwargs=overwrite_kwargs)
if recipe is None:
args = parse(dataset)
if "__type__" not in args:
args["__type__"] = settings.default_recipe
if overwrite_kwargs is not None:
args.update(overwrite_kwargs)
recipe = Artifact.from_dict(args)
return recipe
|