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
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
|
"""Non-core imports in this module are lazily loaded, in order to improve CLI speed """
from pathlib import Path
import time
from typing import TYPE_CHECKING, List, Optional, Union
from jupyter_cache.readers import NbReadError
if TYPE_CHECKING:
from jupyter_cache.base import JupyterCacheAbstract
from jupyter_cache.cache.db import NbCacheRecord, NbProjectRecord
def to_relative_paths(
paths: List[Union[str, Path]],
folder: Union[str, Path],
check_existence: bool = False,
) -> List[Path]:
"""Make paths relative to a reference folder.
:param paths: list of paths
:param folder: The folder that all paths should be in (or subfolder).
:param check_existence: check the paths exist
:raises IOError: path is not relative or failed existence check
"""
rel_paths = []
folder = Path(folder).absolute()
for path in paths:
path = Path(path).absolute()
if check_existence and not path.exists():
raise OSError(f"Path does not exist: {path}")
if check_existence and not path.is_file():
raise OSError(f"Path is not a file: {path}")
try:
rel_path = path.relative_to(folder)
except ValueError:
raise OSError(f"Path '{path}' is not in folder '{folder}''")
rel_paths.append(rel_path)
return rel_paths
class Timer:
"""Context manager for timing runtime."""
def __init__(self):
self._last_time = time.perf_counter()
self._split_time = 0.0
@property
def last_split(self):
return self._split_time
def reset(self):
"""Reset timer."""
self._last_time = time.perf_counter()
self._split_time = 0.0
def split(self):
"""Record a split time."""
self._split_time = time.perf_counter() - self._last_time
def __enter__(self):
"""Reset timer."""
self.reset()
return self
def __exit__(self, *exc_info):
"""Record a split time."""
self.split()
def shorten_path(file_path: Union[str, Path], length: Optional[int]) -> Path:
"""Split the path into separate parts,
select the last 'length' elements and join them again
"""
if length is None:
return Path(file_path)
return Path(*Path(file_path).parts[-length:])
def tabulate_cache_records(
records: List["NbCacheRecord"], hashkeys=False, path_length=None
) -> str:
"""Tabulate cache records.
:param records: list of ``NbCacheRecord``
:param hashkeys: include a hashkey column
:param path_length: truncate URI paths to x components
"""
import tabulate
return tabulate.tabulate(
[
r.format_dict(hashkey=hashkeys, path_length=path_length, show_data=False)
for r in sorted(records, key=lambda r: r.accessed, reverse=True)
],
headers="keys",
)
def tabulate_project_records(
records: List["NbProjectRecord"],
path_length: Optional[int] = None,
cache: Optional["JupyterCacheAbstract"] = None,
assets=False,
) -> str:
"""Tabulate cache records.
:param records: list of ``NbProjectRecord``
:param path_length: truncate URI paths to x components
:param cache: If the cache is given,
we use it to add a column of matched cached pk (if available)
:param assets: Show the number of assets
"""
import tabulate
rows = []
for record in records:
cache_record = None
read_error = None
if cache is not None:
try:
cache_record = cache.get_cached_project_nb(record.uri)
except NbReadError as exc:
read_error = f"{exc.__class__.__name__}: {exc}"
rows.append(
record.format_dict(
cache_record=cache_record,
path_length=path_length,
assets=assets,
read_error=read_error,
)
)
return tabulate.tabulate(rows, headers="keys")
|
