Skip to content

util

SSB-project utils.

Nothing dataclass

Bases: Option[T]

An 'Option' type that doesn't contain a value.

Source code in ssb_project_cli/ssb_project/util.py
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
@dataclass
class Nothing(Option[T]):
    """An 'Option' type that doesn't contain a value."""

    def get_or_else(self, default: U) -> U:
        """Get value if it exists or return default."""
        return default

    def map(self, func: Callable[[T], U]) -> Option[U]:
        """Map function over value if it exists."""
        return Nothing()

    def is_empty(self) -> bool:
        """Check if the Option type contains a value."""
        return True

get_or_else(default)

Get value if it exists or return default.

Source code in ssb_project_cli/ssb_project/util.py
69
70
71
def get_or_else(self, default: U) -> U:
    """Get value if it exists or return default."""
    return default

is_empty()

Check if the Option type contains a value.

Source code in ssb_project_cli/ssb_project/util.py
77
78
79
def is_empty(self) -> bool:
    """Check if the Option type contains a value."""
    return True

map(func)

Map function over value if it exists.

Source code in ssb_project_cli/ssb_project/util.py
73
74
75
def map(self, func: Callable[[T], U]) -> Option[U]:
    """Map function over value if it exists."""
    return Nothing()

Option

Bases: Generic[T]

Represents a value that might be empty.

Source code in ssb_project_cli/ssb_project/util.py
30
31
32
33
34
35
36
37
38
39
40
41
42
43
class Option(Generic[T]):
    """Represents a value that might be empty."""

    def get_or_else(self, default: U) -> Union[T, U]:
        """Get value if it exists or return default."""
        raise NotImplementedError

    def map(self, func: Callable[[T], U]) -> "Option[U]":
        """Map function over value if it exists."""
        raise NotImplementedError

    def is_empty(self) -> bool:
        """Check if the Option type contains a value."""
        raise NotImplementedError

get_or_else(default)

Get value if it exists or return default.

Source code in ssb_project_cli/ssb_project/util.py
33
34
35
def get_or_else(self, default: U) -> Union[T, U]:
    """Get value if it exists or return default."""
    raise NotImplementedError

is_empty()

Check if the Option type contains a value.

Source code in ssb_project_cli/ssb_project/util.py
41
42
43
def is_empty(self) -> bool:
    """Check if the Option type contains a value."""
    raise NotImplementedError

map(func)

Map function over value if it exists.

Source code in ssb_project_cli/ssb_project/util.py
37
38
39
def map(self, func: Callable[[T], U]) -> "Option[U]":
    """Map function over value if it exists."""
    raise NotImplementedError

Some dataclass

Bases: Option[T]

An 'Option' type that contains a value.

Source code in ssb_project_cli/ssb_project/util.py
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
@dataclass
class Some(Option[T]):
    """An 'Option' type that contains a value."""

    value: T

    def get_or_else(self, default: U) -> T:
        """Get value if it exists or return default."""
        return self.value

    def map(self, func: Callable[[T], U]) -> Option[U]:
        """Map function over value if it exists."""
        return Some(func(self.value))

    def is_empty(self) -> bool:
        """Check if the Option type contains a value."""
        return False

get_or_else(default)

Get value if it exists or return default.

Source code in ssb_project_cli/ssb_project/util.py
52
53
54
def get_or_else(self, default: U) -> T:
    """Get value if it exists or return default."""
    return self.value

is_empty()

Check if the Option type contains a value.

Source code in ssb_project_cli/ssb_project/util.py
60
61
62
def is_empty(self) -> bool:
    """Check if the Option type contains a value."""
    return False

map(func)

Map function over value if it exists.

Source code in ssb_project_cli/ssb_project/util.py
56
57
58
def map(self, func: Callable[[T], U]) -> Option[U]:
    """Map function over value if it exists."""
    return Some(func(self.value))

create_error_log(log, calling_function, home_path=HOME_PATH)

Creates a file with log of error in the current folder.

Parameters:

Name Type Description Default
log str

The content of the error log.

required
calling_function str

The function in which the error occurred. Used to give a more descriptive name to error log file.

required
home_path Path

System home path

HOME_PATH
Source code in ssb_project_cli/ssb_project/util.py
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
def create_error_log(
    log: str, calling_function: str, home_path: Path = HOME_PATH
) -> None:
    """Creates a file with log of error in the current folder.

    Args:
        log: The content of the error log.
        calling_function: The function in which the error occurred. Used to give a more descriptive name to error log file.
        home_path: System home path
    """
    try:
        error_logs_path = f"{home_path}/ssb-project-cli/.error_logs"
        if not os.path.exists(error_logs_path):
            os.makedirs(error_logs_path)
        filename = f"{calling_function}-error-{int(time.time())}.txt"
        with open(f"{error_logs_path}/{filename}", "w+") as f:
            f.write(log)
            print(f"Detailed error information saved to {error_logs_path}/{filename}")
            print(
                f"You can find the full debug log here {error_logs_path}/ssb-project-debug.log"
            )
            print(
                f"❗️You can try deleting '.poetry/cache' in your project directory or '{home_path}/.cache/pypoetry'. Cache could be causing problems"
            )
    except Exception as e:
        print(f"Error while attempting to write the log file: {e}")

execute_command(command, command_shortname, success_desc=None, failure_desc=None, cwd=None)

Execute command and handle failure/success cases.

Parameters:

Name Type Description Default
command Union[str, list[str]]

The command to be executed. For example "poetry install".

required
command_shortname str

For example: "poetry-install". Used to create descriptive error log file.

required
success_desc Optional[str]

For example: "Poetry install ran successfully".

None
failure_desc Optional[str]

For example: "Something went wrong while running poetry install".

None
cwd Optional[Path]

The current working directory.

None

Returns:

Type Description
CompletedProcess[bytes]

The result of the subprocess.

Source code in ssb_project_cli/ssb_project/util.py
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
def execute_command(
    command: Union[str, list[str]],
    command_shortname: str,
    success_desc: Optional[str] = None,
    failure_desc: Optional[str] = None,
    cwd: Optional[Path] = None,
) -> subprocess.CompletedProcess[bytes]:
    """Execute command and handle failure/success cases.

    Args:
        command: The command to be executed. For example "poetry install".
        command_shortname: For example: "poetry-install". Used to create descriptive error log file.
        success_desc: For example: "Poetry install ran successfully".
        failure_desc: For example: "Something went wrong while running poetry install".
        cwd: The current working directory.

    Returns:
        The result of the subprocess.
    """
    if isinstance(command, str):
        command = command.split(" ")
    result = subprocess.run(  # noqa S603
        command,
        capture_output=True,
        cwd=cwd,
    )

    if result.returncode != 0:
        calling_function = command_shortname
        log = str(result)
        if failure_desc:
            print(failure_desc)
        else:
            print("Error while running: " + " ".join(command))
        create_error_log(log, calling_function)
        sys.exit(1)
    else:
        if success_desc:
            print(success_desc)

    return result

get_kernels_dict()

Gets installed kernel specifications.

Returns:

Name Type Description
kernel_dict dict[str, dict[str, str]]

Dictionary of installed kernel specifications

Source code in ssb_project_cli/ssb_project/util.py
174
175
176
177
178
179
180
def get_kernels_dict() -> dict[str, dict[str, str]]:
    """Gets installed kernel specifications.

    Returns:
        kernel_dict: Dictionary of installed kernel specifications
    """
    return kernelspec_manager.get_all_specs()

get_project_name_and_root_path(project_path=None)

Get the name and root of the project.

  • First source: .cruft.json
  • Second source: pyproject.toml
  • Final fallback: project root directory name.

Parameters:

Name Type Description Default
project_path Path | None

Optionally supply a path to the project. If not supplied, use the current working directory.

None

Returns:

Name Type Description
project_name str | None

The name of the project.

project_root Path | None

Path of the root directory of the project.

Source code in ssb_project_cli/ssb_project/util.py
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
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
def get_project_name_and_root_path(
    project_path: Path | None = None,
) -> tuple[str | None, Path | None]:
    """Get the name and root of the project.

    - First source: `.cruft.json`
    - Second source: `pyproject.toml`
    - Final fallback: project root directory name.

    Args:
        project_path: Optionally supply a path to the project. If not supplied, use the current working directory.

    Returns:
        project_name: The name of the project.
        project_root: Path of the root directory of the project.
    """
    cruft_json_name = ".cruft.json"
    pyproject_name = "pyproject.toml"
    origin = project_path or Path.cwd()
    if not origin.exists():
        return None, None
    paths = [origin]
    # List of current path and all parents up to the filesystem root
    paths.extend(origin.parents)

    for path in paths:
        if {i.name for i in path.iterdir()}.intersection(
            {cruft_json_name, pyproject_name, ".git"}
        ):
            try:
                # Attempt to source from Cruft first
                name = json.loads((path / cruft_json_name).read_text())["context"][
                    "cookiecutter"
                ]["project_name"]
                return (
                    name,
                    path,
                )
            except (KeyError, FileNotFoundError, json.JSONDecodeError):
                # Fall back to pyproject.toml
                try:
                    name = tomli.loads((path / pyproject_name).read_text())["tool"][
                        "poetry"
                    ]["name"]
                    return (
                        name,
                        path,
                    )
                except (KeyError, FileNotFoundError):
                    # Final fall back to project root directory name
                    return path.name, path
    return None, None

handle_no_kernel_argument(no_kernel)

Handle the 'no_kernel' parameter and environment variable.

The CLI flag is always prioritised, otherwise it falls back to the environment variable and then lastly defaults to False.

Source code in ssb_project_cli/ssb_project/util.py
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
def handle_no_kernel_argument(no_kernel: bool) -> bool:
    """Handle the 'no_kernel' parameter and environment variable.

    The CLI flag is always prioritised, otherwise it falls back to the environment
    variable and then lastly defaults to False.
    """
    if no_kernel:
        return no_kernel
    env_var_no_kernel = os.environ.get("NO_KERNEL")
    if env_var_no_kernel is None:  # handle NO_KERNEL is undefined case
        return False
    elif env_var_no_kernel not in ["True", "False"]:
        print(
            f"""
              The value of the 'NO_KERNEL' environment variable is {os.environ["NO_KERNEL"]}.
              The only valid values are True and False.
            """
        )
        exit(1)
    else:
        return bool(env_var_no_kernel)

remove_kernel_spec(kernel_name)

Remove a kernel spec.

Source code in ssb_project_cli/ssb_project/util.py
183
184
185
def remove_kernel_spec(kernel_name: str) -> None:
    """Remove a kernel spec."""
    kernelspec_manager.remove_kernel_spec(kernel_name)

set_debug_logging(home_path=HOME_PATH)

Creates a file with log of error in the current folder.

Parameters:

Name Type Description Default
home_path Path

path prefix to use for error logging, defaults to HOME_PATH.

HOME_PATH
Source code in ssb_project_cli/ssb_project/util.py
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def set_debug_logging(home_path: Path = HOME_PATH) -> None:
    """Creates a file with log of error in the current folder.

    Args:
        home_path: path prefix to use for error logging, defaults to HOME_PATH.
    """
    error_logs_path = f"{home_path}/ssb-project-cli/.error_logs/ssb-project-debug.log"
    log_dir = os.path.dirname(error_logs_path)
    if not os.path.exists(log_dir):
        os.makedirs(log_dir)
    logging.basicConfig(filename=error_logs_path, level=logging.DEBUG)

try_if_file_exists(f)

Convert a function that may throw a 'FileExistsError' into an Option value that returns None if an error is thrown.

Source code in ssb_project_cli/ssb_project/util.py
82
83
84
85
86
87
def try_if_file_exists(f: Callable[[], U]) -> Option[U]:
    """Convert a function that may throw a 'FileExistsError' into an Option value that returns None if an error is thrown."""
    try:
        return Some(value=f())
    except FileExistsError:
        return Nothing()