Skip to content

Config

InvalidSnkConfigError

Bases: SnkConfigError, ValueError

Thrown if the given SNK config appears to have an invalid format.

Source code in src/snk_cli/config/config.py 
14
15
16
17
class InvalidSnkConfigError(SnkConfigError, ValueError):
    """
    Thrown if the given SNK config appears to have an invalid format.
    """

MissingSnkConfigError

Bases: SnkConfigError, FileNotFoundError

Thrown if the given SNK config file cannot be found.

Source code in src/snk_cli/config/config.py 
19
20
21
22
class MissingSnkConfigError(SnkConfigError, FileNotFoundError):
    """
    Thrown if the given SNK config file cannot be found.
    """

SnkConfig dataclass

A dataclass for storing Snakemake workflow configuration.

Attributes:

Name Type Description
art str

The art to display in the CLI. Defaults to None.
logo str

The logo to display in the CLI. Defaults to None.
tagline str

The tagline to display in the CLI. Defaults to "A Snakemake workflow CLI generated with Snk".
font str

The font size for the CLI. Defaults to "small".
version Optional[str]

The version of the workflow. Defaults to None.
conda bool

Whether to use conda for managing environments. Defaults to True.
resources List[Path]

List of paths to additional resources. Defaults to an empty list.
symlink_resources bool

Whether to symlink resources instead of copying them. Defaults to False.
skip_missing bool

Whether to skip missing CLI options. Defaults to False.
additional_snakemake_args List[str]

List of additional Snakemake command-line arguments. Defaults to an empty list.
commands List[str]

List of subcommands to include in the CLI. Defaults to ["run", "script", "env", "profile", "info", "config"].
cli dict

Dictionary of CLI options and their values. Defaults to an empty dictionary.
_snk_config_path Path

The path to the SNK config file. Defaults to None.
Methods

from_path(snk_config_path: Path) -> SnkConfig: Load and validate Snk config from file.

from_workflow_dir(workflow_dir_path: Path, create_if_not_exists: bool = False) -> SnkConfig: Load and validate SNK config from workflow directory.

validate_resources(resources: List[Path]) -> None: Validate resources.

add_resources(resources: List[Path], workflow_dir_path: Path = None) -> None: Add resources to the SNK config.

to_yaml(path: Path) -> None: Write SNK config to YAML file.

save() -> None: Save SNK config.

Source code in src/snk_cli/config/config.py 
 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
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
@dataclass
class SnkConfig:
    """
    A dataclass for storing Snakemake workflow configuration.

    Attributes:
      art (str, optional): The art to display in the CLI. Defaults to None.
      logo (str, optional): The logo to display in the CLI. Defaults to None.
      tagline (str): The tagline to display in the CLI. Defaults to "A Snakemake workflow CLI generated with Snk".
      font (str): The font size for the CLI. Defaults to "small".
      version (Optional[str], optional): The version of the workflow. Defaults to None.
      conda (bool): Whether to use conda for managing environments. Defaults to True.
      resources (List[Path]): List of paths to additional resources. Defaults to an empty list.
      symlink_resources (bool): Whether to symlink resources instead of copying them. Defaults to False.
      skip_missing (bool): Whether to skip missing CLI options. Defaults to False.
      additional_snakemake_args (List[str]): List of additional Snakemake command-line arguments. Defaults to an empty list.
      commands (List[str]): List of subcommands to include in the CLI. Defaults to ["run", "script", "env", "profile", "info", "config"].
      cli (dict): Dictionary of CLI options and their values. Defaults to an empty dictionary.
      _snk_config_path (Path): The path to the SNK config file. Defaults to None.

    Methods:
      from_path(snk_config_path: Path) -> SnkConfig:
        Load and validate Snk config from file.

      from_workflow_dir(workflow_dir_path: Path, create_if_not_exists: bool = False) -> SnkConfig:
        Load and validate SNK config from workflow directory.

      validate_resources(resources: List[Path]) -> None:
        Validate resources.

      add_resources(resources: List[Path], workflow_dir_path: Path = None) -> None:
        Add resources to the SNK config.

      to_yaml(path: Path) -> None:
        Write SNK config to YAML file.

      save() -> None:
        Save SNK config.
    """

    art: str = None
    logo: str = None
    tagline: str = "A Snakemake workflow CLI generated with Snk"
    font: str = "small"
    version: Optional[str] = None
    conda: bool = True
    resources: List[Path] = field(default_factory=list)
    symlink_resources: bool = False
    skip_missing: bool = False # skip any missing cli options (i.e. those not in the snk file)
    additional_snakemake_args: List[str] = field(default_factory=list)
    commands: List[str] = field(default_factory=lambda: ["run", "script", "env", "profile", "info", "config"])
    cli: dict = field(default_factory=dict)
    _snk_config_path: Path = None

    @classmethod
    def from_path(cls, snk_config_path: Path):
        """
        Load and validate Snk config from file.
        Args:
          snk_config_path (Path): Path to the SNK config file.
        Returns:
          SnkConfig: A SnkConfig object.
        Raises:
          FileNotFoundError: If the SNK config file is not found.
        Examples:
          >>> SnkConfig.from_path(Path("snk.yaml"))
          SnkConfig(art=None, logo=None, tagline='A Snakemake workflow CLI generated with Snk', font='small', resources=[], annotations={}, symlink_resources=False, _snk_config_path=PosixPath('snk.yaml'))
        """
        if not snk_config_path.exists():
            raise MissingSnkConfigError(
                f"Could not find SNK config file: {snk_config_path}"
            ) from FileNotFoundError
        # raise error if file is empty
        if snk_config_path.stat().st_size == 0:
            raise InvalidSnkConfigError(f"SNK config file is empty: {snk_config_path}") from ValueError

        snk_config_dict = snakemake.load_configfile(snk_config_path)
        snk_config_dict["version"] = get_version_from_config(snk_config_path, snk_config_dict)
        if "annotations" in snk_config_dict:
            # TODO: remove annotations in the future
            snk_config_dict["cli"] = snk_config_dict["annotations"]
            del snk_config_dict["annotations"]
        if "conda_required" in snk_config_dict:
            # TODO: remove conda_required in the future
            snk_config_dict["conda"] = snk_config_dict["conda_required"]
            del snk_config_dict["conda_required"]
        snk_config = cls(**snk_config_dict)
        snk_config.resources = [
            snk_config_path.parent / resource for resource in snk_config.resources
        ]
        snk_config.validate_resources(snk_config.resources)
        snk_config._snk_config_path = snk_config_path
        return snk_config

    @classmethod
    def from_workflow_dir(
        cls, workflow_dir_path: Path, create_if_not_exists: bool = False
    ):
        """
        Load and validate SNK config from workflow directory.
        Args:
          workflow_dir_path (Path): Path to the workflow directory.
          create_if_not_exists (bool): Whether to create a SNK config file if one does not exist.
        Returns:
          SnkConfig: A SnkConfig object.
        Raises:
          FileNotFoundError: If the SNK config file is not found.
        Examples:
          >>> SnkConfig.from_workflow_dir(Path("workflow"))
          SnkConfig(art=None, logo=None, tagline='A Snakemake workflow CLI generated with Snk', font='small', resources=[], annotations={}, symlink_resources=False, _snk_config_path=PosixPath('workflow/snk.yaml'))
        """
        if (workflow_dir_path / "snk.yaml").exists():
            return cls.from_path(workflow_dir_path / "snk.yaml")
        elif (workflow_dir_path / ".snk").exists():
            import warnings

            warnings.warn(
                "Use of .snk will be deprecated in the future. Please use snk.yaml instead.",
                DeprecationWarning,
            )
            return cls.from_path(workflow_dir_path / ".snk")
        elif create_if_not_exists:
            snk_config = cls(_snk_config_path=workflow_dir_path / "snk.yaml")
            return snk_config
        else:
            raise FileNotFoundError(
                f"Could not find SNK config file in workflow directory: {workflow_dir_path}"
            )

    def validate_resources(self, resources):
        """
        Validate resources.
        Args:
          resources (List[Path]): List of resources to validate.
        Raises:
          FileNotFoundError: If a resource is not found.
        Notes:
          This function does not modify the resources list.
        Examples:
          >>> SnkConfig.validate_resources([Path("resource1.txt"), Path("resource2.txt")])
        """
        for resource in resources:
            if not resource.exists():
                raise FileNotFoundError(f"Could not find resource: {resource}")

    def add_resources(self, resources: List[Path], workflow_dir_path: Path = None):
        """
        Add resources to the SNK config.
        Args:
          resources (List[Path]): List of resources to add.
          workflow_dir_path (Path): Path to the workflow directory.
        Returns:
          None
        Side Effects:
          Adds the resources to the SNK config.
        Examples:
          >>> snk_config = SnkConfig()
          >>> snk_config.add_resources([Path("resource1.txt"), Path("resource2.txt")], Path("workflow"))
        """
        processed = []
        for resource in resources:
            if workflow_dir_path and not resource.is_absolute():
                resource = workflow_dir_path / resource
            processed.append(resource)
        self.validate_resources(processed)
        self.resources.extend(processed)

    def to_yaml(self, path: Path) -> None:
        """
        Write SNK config to YAML file.
        Args:
          path (Path): Path to write the YAML file to.
        Returns:
          None
        Side Effects:
          Writes the SNK config to the specified path.
        Examples:
          >>> snk_config = SnkConfig()
          >>> snk_config.to_yaml(Path("snk.yaml"))
        """
        config_dict = {k: v for k, v in vars(self).items() if not k.startswith("_")}
        with open(path, "w") as f:
            yaml.dump(config_dict, f)

    def save(self) -> None:
        """
        Save SNK config.
        Args:
          path (Path): Path to write the YAML file to.
        Returns:
          None
        Side Effects:
          Writes the SNK config to the path specified by _snk_config_path.
        Examples:
          >>> snk_config = SnkConfig()
          >>> snk_config.save()
        """
        self.to_yaml(self._snk_config_path)

add_resources(resources, workflow_dir_path=None)

Add resources to the SNK config.

Parameters:

Name Type Description Default
resources List[Path]

List of resources to add.

 required
workflow_dir_path Path

Path to the workflow directory.

 None

Returns:

Type Description

None
Side Effects

Adds the resources to the SNK config.

Examples:

>>> snk_config = SnkConfig()
>>> snk_config.add_resources([Path("resource1.txt"), Path("resource2.txt")], Path("workflow"))
Source code in src/snk_cli/config/config.py 
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
def add_resources(self, resources: List[Path], workflow_dir_path: Path = None):
    """
    Add resources to the SNK config.
    Args:
      resources (List[Path]): List of resources to add.
      workflow_dir_path (Path): Path to the workflow directory.
    Returns:
      None
    Side Effects:
      Adds the resources to the SNK config.
    Examples:
      >>> snk_config = SnkConfig()
      >>> snk_config.add_resources([Path("resource1.txt"), Path("resource2.txt")], Path("workflow"))
    """
    processed = []
    for resource in resources:
        if workflow_dir_path and not resource.is_absolute():
            resource = workflow_dir_path / resource
        processed.append(resource)
    self.validate_resources(processed)
    self.resources.extend(processed)

from_path(snk_config_path) classmethod

Load and validate Snk config from file.

Parameters:

Name Type Description Default
snk_config_path Path

Path to the SNK config file.

 required

Returns:

Name Type Description
SnkConfig

A SnkConfig object.

Raises:

Type Description
FileNotFoundError

If the SNK config file is not found.

Examples:

>>> SnkConfig.from_path(Path("snk.yaml"))
SnkConfig(art=None, logo=None, tagline='A Snakemake workflow CLI generated with Snk', font='small', resources=[], annotations={}, symlink_resources=False, _snk_config_path=PosixPath('snk.yaml'))
Source code in src/snk_cli/config/config.py 
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
@classmethod
def from_path(cls, snk_config_path: Path):
    """
    Load and validate Snk config from file.
    Args:
      snk_config_path (Path): Path to the SNK config file.
    Returns:
      SnkConfig: A SnkConfig object.
    Raises:
      FileNotFoundError: If the SNK config file is not found.
    Examples:
      >>> SnkConfig.from_path(Path("snk.yaml"))
      SnkConfig(art=None, logo=None, tagline='A Snakemake workflow CLI generated with Snk', font='small', resources=[], annotations={}, symlink_resources=False, _snk_config_path=PosixPath('snk.yaml'))
    """
    if not snk_config_path.exists():
        raise MissingSnkConfigError(
            f"Could not find SNK config file: {snk_config_path}"
        ) from FileNotFoundError
    # raise error if file is empty
    if snk_config_path.stat().st_size == 0:
        raise InvalidSnkConfigError(f"SNK config file is empty: {snk_config_path}") from ValueError

    snk_config_dict = snakemake.load_configfile(snk_config_path)
    snk_config_dict["version"] = get_version_from_config(snk_config_path, snk_config_dict)
    if "annotations" in snk_config_dict:
        # TODO: remove annotations in the future
        snk_config_dict["cli"] = snk_config_dict["annotations"]
        del snk_config_dict["annotations"]
    if "conda_required" in snk_config_dict:
        # TODO: remove conda_required in the future
        snk_config_dict["conda"] = snk_config_dict["conda_required"]
        del snk_config_dict["conda_required"]
    snk_config = cls(**snk_config_dict)
    snk_config.resources = [
        snk_config_path.parent / resource for resource in snk_config.resources
    ]
    snk_config.validate_resources(snk_config.resources)
    snk_config._snk_config_path = snk_config_path
    return snk_config

from_workflow_dir(workflow_dir_path, create_if_not_exists=False) classmethod

Load and validate SNK config from workflow directory.

Parameters:

Name Type Description Default
workflow_dir_path Path

Path to the workflow directory.

 required
create_if_not_exists bool

Whether to create a SNK config file if one does not exist.

 False

Returns:

Name Type Description
SnkConfig

A SnkConfig object.

Raises:

Type Description
FileNotFoundError

If the SNK config file is not found.

Examples:

>>> SnkConfig.from_workflow_dir(Path("workflow"))
SnkConfig(art=None, logo=None, tagline='A Snakemake workflow CLI generated with Snk', font='small', resources=[], annotations={}, symlink_resources=False, _snk_config_path=PosixPath('workflow/snk.yaml'))
Source code in src/snk_cli/config/config.py 
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
@classmethod
def from_workflow_dir(
    cls, workflow_dir_path: Path, create_if_not_exists: bool = False
):
    """
    Load and validate SNK config from workflow directory.
    Args:
      workflow_dir_path (Path): Path to the workflow directory.
      create_if_not_exists (bool): Whether to create a SNK config file if one does not exist.
    Returns:
      SnkConfig: A SnkConfig object.
    Raises:
      FileNotFoundError: If the SNK config file is not found.
    Examples:
      >>> SnkConfig.from_workflow_dir(Path("workflow"))
      SnkConfig(art=None, logo=None, tagline='A Snakemake workflow CLI generated with Snk', font='small', resources=[], annotations={}, symlink_resources=False, _snk_config_path=PosixPath('workflow/snk.yaml'))
    """
    if (workflow_dir_path / "snk.yaml").exists():
        return cls.from_path(workflow_dir_path / "snk.yaml")
    elif (workflow_dir_path / ".snk").exists():
        import warnings

        warnings.warn(
            "Use of .snk will be deprecated in the future. Please use snk.yaml instead.",
            DeprecationWarning,
        )
        return cls.from_path(workflow_dir_path / ".snk")
    elif create_if_not_exists:
        snk_config = cls(_snk_config_path=workflow_dir_path / "snk.yaml")
        return snk_config
    else:
        raise FileNotFoundError(
            f"Could not find SNK config file in workflow directory: {workflow_dir_path}"
        )

save()

Save SNK config.

Parameters:

Name Type Description Default
path Path

Path to write the YAML file to.

 required

Returns:

Type Description
None

None
Side Effects

Writes the SNK config to the path specified by _snk_config_path.

Examples:

>>> snk_config = SnkConfig()
>>> snk_config.save()
Source code in src/snk_cli/config/config.py 
208
209
210
211
212
213
214
215
216
217
218
219
220
221
def save(self) -> None:
    """
    Save SNK config.
    Args:
      path (Path): Path to write the YAML file to.
    Returns:
      None
    Side Effects:
      Writes the SNK config to the path specified by _snk_config_path.
    Examples:
      >>> snk_config = SnkConfig()
      >>> snk_config.save()
    """
    self.to_yaml(self._snk_config_path)

to_yaml(path)

Write SNK config to YAML file.

Parameters:

Name Type Description Default
path Path

Path to write the YAML file to.

 required

Returns:

Type Description
None

None
Side Effects

Writes the SNK config to the specified path.

Examples:

>>> snk_config = SnkConfig()
>>> snk_config.to_yaml(Path("snk.yaml"))
Source code in src/snk_cli/config/config.py 
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
def to_yaml(self, path: Path) -> None:
    """
    Write SNK config to YAML file.
    Args:
      path (Path): Path to write the YAML file to.
    Returns:
      None
    Side Effects:
      Writes the SNK config to the specified path.
    Examples:
      >>> snk_config = SnkConfig()
      >>> snk_config.to_yaml(Path("snk.yaml"))
    """
    config_dict = {k: v for k, v in vars(self).items() if not k.startswith("_")}
    with open(path, "w") as f:
        yaml.dump(config_dict, f)

validate_resources(resources)

Validate resources.

Parameters:

Name Type Description Default
resources List[Path]

List of resources to validate.

 required

Raises:

Type Description
FileNotFoundError

If a resource is not found.
Notes

This function does not modify the resources list.

Examples:

>>> SnkConfig.validate_resources([Path("resource1.txt"), Path("resource2.txt")])
Source code in src/snk_cli/config/config.py 
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
def validate_resources(self, resources):
    """
    Validate resources.
    Args:
      resources (List[Path]): List of resources to validate.
    Raises:
      FileNotFoundError: If a resource is not found.
    Notes:
      This function does not modify the resources list.
    Examples:
      >>> SnkConfig.validate_resources([Path("resource1.txt"), Path("resource2.txt")])
    """
    for resource in resources:
        if not resource.exists():
            raise FileNotFoundError(f"Could not find resource: {resource}")

SnkConfigError

Bases: Exception

Base class for all SNK config exceptions.

Source code in src/snk_cli/config/config.py 
 9
10
11
12
class SnkConfigError(Exception):
    """
    Base class for all SNK config exceptions.
    """

get_config_from_workflow_dir(workflow_dir_path)

Get the config file from a workflow directory.

Parameters:

Name Type Description Default
workflow_dir_path Path

Path to the workflow directory.

 required

Returns:

Name Type Description
Path

Path to the config file, or None if not found.

Examples:

>>> get_config_from_workflow_dir(Path("workflow"))
PosixPath('workflow/config.yaml')
Source code in src/snk_cli/config/config.py 
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
def get_config_from_workflow_dir(workflow_dir_path: Path):
    """
    Get the config file from a workflow directory.
    Args:
      workflow_dir_path (Path): Path to the workflow directory.
    Returns:
      Path: Path to the config file, or None if not found.
    Examples:
      >>> get_config_from_workflow_dir(Path("workflow"))
      PosixPath('workflow/config.yaml')
    """
    for path in [
        Path("config") / "config.yaml",
        Path("config") / "config.yml",
        "config.yaml",
        "config.yml",
    ]:
        if (workflow_dir_path / path).exists():
            return workflow_dir_path / path
    return None

load_workflow_snakemake_config(workflow_dir_path)

Load the Snakemake config from a workflow directory.

Parameters:

Name Type Description Default
workflow_dir_path Path

Path to the workflow directory.

 required

Returns:

Name Type Description
dict

The Snakemake config.

Examples:

>>> load_workflow_snakemake_config(Path("workflow"))
{'inputs': {'data': 'data.txt'}, 'outputs': {'results': 'results.txt'}}
Source code in src/snk_cli/config/config.py 
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
def load_workflow_snakemake_config(workflow_dir_path: Path):
    """
    Load the Snakemake config from a workflow directory.
    Args:
      workflow_dir_path (Path): Path to the workflow directory.
    Returns:
      dict: The Snakemake config.
    Examples:
      >>> load_workflow_snakemake_config(Path("workflow"))
      {'inputs': {'data': 'data.txt'}, 'outputs': {'results': 'results.txt'}}
    """
    workflow_config_path = get_config_from_workflow_dir(workflow_dir_path)
    if not workflow_config_path or not workflow_config_path.exists():
        return {}
    return snakemake.load_configfile(workflow_config_path)

get_version_from_config(config_path, config_dict=None)

Get the version from the config file or config dictionary.

Parameters:

Name Type Description Default
config_path Path

Path to the config file.

 required
config_dict dict

Config dictionary. Defaults to None.

 None

Returns:

Name Type Description
str str

The version.

Raises:

Type Description
FileNotFoundError

If the version file (about.py) is not found.
KeyError

If the version key is not found in the version file.

Examples:

>>> get_version_from_config(Path("config.yaml"))
'0.1.0'
>>> get_version_from_config(Path("config.yaml"), {"version": "0.2.0"})
'0.2.0'
Source code in src/snk_cli/config/utils.py 
 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
45
46
47
48
def get_version_from_config(config_path: Path, config_dict: dict = None) -> str:
    """
    Get the version from the config file or config dictionary.

    Args:
      config_path (Path): Path to the config file.
      config_dict (dict, optional): Config dictionary. Defaults to None.

    Returns:
      str: The version.

    Raises:
      FileNotFoundError: If the version file (__about__.py) is not found.
      KeyError: If the __version__ key is not found in the version file.

    Examples:
      >>> get_version_from_config(Path("config.yaml"))
      '0.1.0'
      >>> get_version_from_config(Path("config.yaml"), {"version": "0.2.0"})
      '0.2.0'
    """
    if not config_dict:
        config_dict = load_configfile(config_path)

    if "version" not in config_dict:
        return None 
    if config_dict["version"] is None:
        return None
    version = str(config_dict["version"])
    if "__about__.py" in version:
        # load version from __about__.py
        about_path = config_path.parent / version
        if not about_path.exists():
            raise FileNotFoundError(
                f"Could not find version file: {about_path}"
            )
        about = {}
        exec(about_path.read_text(), about)
        try:
            version = about["__version__"]
        except KeyError as e:
            raise KeyError(
                f"Could not find __version__ in file: {about_path}"
            ) from e
    return version