Skip to content

parse_dagster_definition

kedro_dagster.datasets.partitioned_dataset.parse_dagster_definition(config)

Parse and instantiate a partition definition class using a config.

Parameters

Name Type Description Default
config dict[str, Any]

Partition definition config. Must contain a type key with a fully-qualified class name or a class object.

 required

Returns

Type Description
tuple[type[PartitionsDefinition], dict[str, Any]]

Class object and remaining config.

Source Code

View on GitHub

Show/Hide source 
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
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
def parse_dagster_definition(
    config: dict[str, Any],
) -> tuple[type[Any], dict[str, Any]]:
    """Parse and instantiate a partition definition class using a config.

    Parameters
    ----------
    config : dict[str, Any]
        Partition definition config. Must contain a ``type`` key with a fully-qualified class
        name or a class object.

    Returns
    -------
    tuple[type[PartitionsDefinition], dict[str, Any]]
        Class object and remaining config.
    """
    config = copy.deepcopy(config)
    definition_type = config.pop(TYPE_KEY)

    class_obj: type[Any] | None = None
    error_msg = None
    if isinstance(definition_type, str):
        if len(definition_type.strip(".")) != len(definition_type):
            raise TypeError("'type' class path does not support relative paths or paths ending with a dot.")
        class_paths = (prefix + definition_type for prefix in _DEFAULT_PACKAGES)

        # Try to resolve the class by attempting a few import paths; record the last error message if all fail
        for class_path in class_paths:
            try:
                tmp, error_msg = _load_obj(class_path)  # Try to load partition class
            except TypeError:  # noqa: BLE001
                try:
                    tmp = _load_obj(class_path)
                except Exception as exc:
                    error_msg = str(exc)
                    raise TypeError(f"Error loading class '{class_path}': {error_msg}") from exc

            if tmp is not None:
                class_obj = tmp
                break

        if class_obj is None:  # If no valid class was found, raise an error
            default_error_msg = f"Class '{definition_type}' not found, is this a typo?"
            raise TypeError(f"{error_msg if error_msg else default_error_msg}")

    if class_obj is None:
        class_obj = definition_type

    return class_obj, config