Utility functions

Utility functions used in the fastai library

doc(fastcore.xtras) shows an overview instead of listing every symbol.

File Functions

IO utilities include:

maybe_open: accept a path or an open file, closing only files it opens.
run: run a command and return stdout; raise IOError on failure.
atomic_save: write a temporary file, then rename it over the destination.
untar_dir: extract an archive, creating a containing directory when needed.
globtastic: recursively find files with include and exclude filters.

walk_join

def walk_join(
    root, name
):

source

walk

def walk(
    path:pathlib.Path | str, # path to start searching
    symlinks:bool=True, # follow symlinks?
    keep_file:<built-in function callable>=ret_true, # function that returns True for wanted files
    keep_folder:<built-in function callable>=ret_true, # function that returns True for folders to enter
    skip_folder:<built-in function callable>=ret_false, # function that returns True for folders to skip
    func:<built-in function callable>=walk_join, # function applied to each entry; default adds `/` to folders
    ret_folders:bool=False, # include folder entries inline with files?
    sort:bool=True, # sort entries alphabetically within each folder?
    maxdepth:int=None, # max depth to descend (1=just immediate contents; None=unlimited)
):

Generator: yields files and (optionally) folders as unified, inline-sorted entries

source

exttypes

def exttypes(
    types
):

Get exts for comma-separated or list typ; if not found in list, return list with just types. Supported: py, js, java, c, cpp, rb, r, ex, sh, web, doc, cfg

print(exttypes('py,doc'))
print(exttypes('zig,txt'))

['ipynb', 'py', 'md', 'rst']
['zig', 'txt']

source

globtastic

def globtastic(
    path:pathlib.Path | str='.', # path to start searching
    recursive:bool=True, # search subfolders
    maxdepth:int=None, # max depth to descend (1=just immediate contents; None=unlimited)
    symlinks:bool=True, # follow symlinks?
    file_glob:str=None, # Only include files matching glob
    file_re:str=None, # Only include files matching regex
    path_glob:str=None, # Only include files whose full paths match glob (wildcards match separators)
    path_re:str=None, # Only include files whose full paths match regex
    folder_re:str=None, # Only enter folders matching regex
    skip_file_glob:str=None, # Skip files matching glob
    skip_file_re:str=None, # Skip files matching regex
    skip_folder_re:str=None, # Skip folders matching regex,
    func:<built-in function callable>=walk_join, # function to apply to each matched file
    ret_folders:bool=False, # return folders, not just files
    sort:bool=True, # sort files by name within each folder
    types:str | list=None, # list or comma-separated str of ext types from: py, js, java, c, cpp, rb, r, ex, sh, web, doc, cfg
    exts:str | list=None, # list or comma-separated str of exts to include
)->fastcore.foundation.L: # Paths to matched files

A more powerful glob, including regex matches, symlink handling, and skip parameters

path is the directory to search. file_glob and file_re match filenames. path_glob and path_re match each file’s full path, including the search directory. These filters do not control directory traversal. path_glob uses fnmatch: * matches directory separators and ** has no special meaning.

globtastic('.', skip_folder_re='^[_.]', folder_re='core', file_glob='*.*py*', file_re='c')

['./01_basics.ipynb', './03c_aio.ipynb', './04_docments.ipynb', './05a_apisurface.ipynb', './06_script.ipynb', './14_funccall.ipynb', './fastcore/apisurface.py', './fastcore/basics.py', './fastcore/dispatch.py', './fastcore/docments.py', './fastcore/docscrape.py', './fastcore/funccall.py', './fastcore/script.py']

source

pglob

def pglob(
    path:pathlib.Path | str='.', # path to start searching
    func:<built-in function callable>=Path, # function to apply to each matched file
    *, recursive:bool=True, maxdepth:int=None, symlinks:bool=True, file_glob:str=None, file_re:str=None,
    path_glob:str=None, path_re:str=None, folder_re:str=None, skip_file_glob:str=None, skip_file_re:str=None,
    skip_folder_re:str=None, ret_folders:bool=False, sort:bool=True, types:str | list=None, exts:str | list=None
)->fastcore.foundation.L: # Paths to matched files

Shortcut for globtastic(..., call=Path)

pglob('..', skip_folder_re='^[_.]', types='doc', skip_file_re='^_')[:6]

[Path('../CHANGELOG.md'), Path('../CODE_OF_CONDUCT.md'), Path('../CONTRIBUTING.md'), Path('../README.md')]

source

maybe_open

def maybe_open(
    f, mode:str='r', **kwargs
):

Context manager: open f if it is a path (and close on exit)

This is useful for functions where you want to accept a path or file. maybe_open will not close your file handle if you pass one in.

def _f(fn):
    with maybe_open(fn) as f: return f.encoding

fname = '00_test.ipynb'
sys_encoding = 'cp1252' if sys.platform == 'win32' else 'utf-8'
test_eq(_f(fname).lower(), sys_encoding)
with open(fname) as fh: test_eq(_f(fh).lower(), sys_encoding)

For example, we can use this to reimplement imghdr.what from the Python standard library, which is written in Python 3.9 as:

from fastcore import imghdr

def what(file, h=None):
    f = None
    try:
        if h is None:
            if isinstance(file, (str,os.PathLike)):
                f = open(file, 'rb')
                h = f.read(32)
            else:
                location = file.tell()
                h = file.read(32)
                file.seek(location)
        for tf in imghdr.tests:
            res = tf(h, f)
            if res: return res
    finally:
        if f: f.close()
    return None

Here’s an example of the use of this function:

fname = 'images/puppy.jpg'
what(fname)

'jpeg'

With maybe_open, Self, and L.map_first, we can rewrite this in a much more concise and (in our opinion) clear way:

def what(file, h=None):
    if h is None:
        with maybe_open(file, 'rb') as f: h = f.peek(32)
    return L(imghdr.tests).map_first(~Self(h,file))

…and we can check that it still works:

test_eq(what(fname), 'jpeg')

…along with the version passing a file handle:

with open(fname,'rb') as f: test_eq(what(f), 'jpeg')

…along with the h parameter version:

with open(fname,'rb') as f: test_eq(what(None, h=f.read(32)), 'jpeg')

source

mkdir

def mkdir(
    path, exist_ok:bool=False, parents:bool=False, overwrite:bool=False, **kwargs
):

Creates and returns a directory defined by path, optionally removing previous existing directory if overwrite is True

with tempfile.TemporaryDirectory() as d:
    path = Path(os.path.join(d, 'new_dir'))
    new_dir = mkdir(path)
    assert new_dir.exists()
    test_eq(new_dir, path)
        
    # test overwrite
    with open(new_dir/'test.txt', 'w') as f: f.writelines('test')
    test_eq(len(list(walk(new_dir))), 1) # assert file is present
    new_dir = mkdir(new_dir, overwrite=True)
    test_eq(len(list(walk(new_dir))), 0) # assert file was deleted

source

image_size

def image_size(
    fn
):

Tuple of (w,h) for png, gif, or jpg; None otherwise

test_eq(image_size(fname), (1200,803))

from PIL import Image
from IPython.display import Image as IPImage

img = Image.new('RGB', (50, 50), color='red')
img

source

img_bytes

def img_bytes(
    img, fmt:str='PNG'
):

ib = img_bytes(img)
IPImage(ib)

source

detect_mime

def detect_mime(
    data
):

Get the MIME type for bytes data, covering common PDF, audio, video, and image types

detect_mime(ib)

'image/png'

test_is(detect_mime('this is a non-byte string'),None)

source

bunzip

def bunzip(
    fn
):

bunzip fn, raising exception if output already exists

f = Path('files/test.txt')
if f.exists(): f.unlink()
bunzip('files/test.txt.bz2')
t = f.open().readlines()
test_eq(len(t),1)
test_eq(t[0], 'test\n')
f.unlink()

source

loads

def loads(
    s, **kw
):

Same as json.loads, but handles None

source

loads_multi

def loads_multi(
    s:str
):

Generator of >=0 decoded json dicts, possibly with non-json ignored text at start and end

tst = """
# ignored
{ "a":1 }
hello
{
"b":2
}
"""

test_eq(list(loads_multi(tst)), [{'a': 1}, {'b': 2}])

source

dumps

def dumps(
    obj, **kw
):

Same as json.dumps

source

untar_dir

def untar_dir(
    fname, dest, rename:bool=False, overwrite:bool=False, uid:int=-1, gid:int=-1
):

untar file into dest, creating a directory if the root contains more than one item; recursively chown if uid/gid set

def test_untar(foldername, rename=False, **kwargs):
    with tempfile.TemporaryDirectory() as d:
        nm = os.path.join(d, 'a')
        shutil.make_archive(nm, 'gztar', **kwargs)
        with tempfile.TemporaryDirectory() as d2:
            d2 = Path(d2)
            untar_dir(nm+'.tar.gz', d2, rename=rename)
            test_eq(d2.ls(), [d2/foldername])

If the contents of fname contain just one file or directory, it is placed directly in dest:

# using `base_dir` in `make_archive` results in `images` directory included in file names
test_untar('images', base_dir='images')

With uid/gid, ownership is set on every extracted entry, including symlinks themselves (not their targets), so a dangling link inside the archive does not abort the extraction:

if hasattr(os, 'chown'):  # not on Windows
    with tempfile.TemporaryDirectory() as d:
        src = Path(d)/'links'
        src.mkdir()
        (src/'dangling').symlink_to('missing-target')
        shutil.make_archive(str(Path(d)/'a'), 'gztar', root_dir=d, base_dir='links')
        out = untar_dir(Path(d)/'a.tar.gz', Path(d)/'out', uid=os.getuid(), gid=os.getgid())
        assert (out/'dangling').is_symlink()

If rename then the directory created is named based on the archive, without extension:

test_untar('a', base_dir='images', rename=True)

If the contents of fname contain multiple files and directories, a new folder in dest is created with the same name as fname (but without extension):

# using `root_dir` in `make_archive` results in `images` directory *not* included in file names
test_untar('a', root_dir='images')

source

repo_details

def repo_details(
    url
):

Tuple of owner,name from ssh or https git repo url

test_eq(repo_details('https://github.com/fastai/fastai.git'), ['fastai', 'fastai'])
test_eq(repo_details('[email protected]:fastai/nbdev.git\n'), ['fastai', 'nbdev'])

source

shell

def shell(
    *args, **kwargs
):

Shortcut for subprocess.run(shell=True)

source

ssh

def ssh(
    host, args:str='', user:str='ubuntu', sock:NoneType=None
):

Run SSH command with given arguments

source

rsync_multi

def rsync_multi(
    ip, files, user:str='ubuntu', persist:str='5m'
):

Transfer multiple files with rename using persistent SSH connection

source

run

def run(
    cmd, *rest, same_in_win:bool=False, ignore_ex:bool=False, as_bytes:bool=False, stderr:bool=True,
    inp:NoneType=None
):

Pass cmd (splitting with shlex if string) to subprocess.run; return stdout; raise IOError if fails

You can pass a string (which will be split based on standard shell rules), a list, or pass args directly:

run('echo', same_in_win=True)
run('pip', '--version', same_in_win=True)
run(['pip', '--version'], same_in_win=True)

'pip 26.2.1 from /Users/jhoward/aai-ws/.venv/lib/python3.13/site-packages/pip (python 3.13)'

if sys.platform == 'win32':
    assert 'ipynb' in run('cmd /c dir /p')
    assert 'ipynb' in run(['cmd', '/c', 'dir', '/p'])
    assert 'ipynb' in run('cmd', '/c', 'dir',  '/p')
else:
    assert 'ipynb' in run('ls -ls')
    assert 'ipynb' in run(['ls', '-l'])
    assert 'ipynb' in run('ls', '-l')

Some commands fail in non-error situations, like grep. Use ignore_ex in those cases, which will return a tuple of stdout and returncode:

if sys.platform == 'win32': test_eq(run('cmd /c findstr asdfds 00_test.ipynb', ignore_ex=True)[0], 1)
else: test_eq(run('grep asdfds 00_test.ipynb', ignore_ex=True)[0], 1)

run automatically decodes returned bytes to a str. Use as_bytes to skip that:

if sys.platform == 'win32': test_eq(run('cmd /c echo hi'), 'hi')
else: test_eq(run('echo hi', as_bytes=True), b'hi\n')

source

open_file

def open_file(
    fn, mode:str='r', **kwargs
):

Open a file, with optional compression if gz or bz2 suffix

source

save_pickle

def save_pickle(
    fn, o
):

Save a pickle file, to a file name or opened file

source

load_pickle

def load_pickle(
    fn
):

Load a pickle file from a file name or opened file

for suf in '.pkl','.bz2','.gz':
    # delete=False is added for Windows
    # https://stackoverflow.com/questions/23212435/permission-denied-to-write-to-my-temporary-file
    with tempfile.NamedTemporaryFile(suffix=suf, delete=False) as f:
        fn = Path(f.name)
        save_pickle(fn, 't')
        t = load_pickle(fn)
    f.close()
    test_eq(t,'t')

source

parse_env

def parse_env(
    s:str=None, fn:Union[str, pathlib.Path]=None
)->dict:

Parse a shell-style environment string or file

testf = """# comment
   # another comment
 export FOO="bar#baz"
BAR=thing # comment "ok"
  baz='thong'
QUX=quux
export ZAP = "zip" # more comments
   FOOBAR = 42   # trailing space and comment"""

exp = dict(FOO='bar#baz', BAR='thing', baz='thong', QUX='quux', ZAP='zip', FOOBAR='42')

test_eq(parse_env(testf),  exp)

source

expand_wildcards

def expand_wildcards(
    code
):

Expand all wildcard imports in the given code string.

inp = """from math import *
from os import *
from random import *
def func(): return sin(pi) + path.join('a', 'b') + randint(1, 10)"""

exp = """from math import pi, sin
from os import path
from random import randint
def func(): return sin(pi) + path.join('a', 'b') + randint(1, 10)"""

test_eq(expand_wildcards(inp), exp)

inp = """from itertools import *
def func(): pass"""
test_eq(expand_wildcards(inp), inp)

inp = """def outer():
    from math import *
    def inner():
        from os import *
        return sin(pi) + path.join('a', 'b')"""

exp = """def outer():
    from math import pi, sin
    def inner():
        from os import path
        return sin(pi) + path.join('a', 'b')"""

test_eq(expand_wildcards(inp), exp)

source

Path.mkdir_perms

def mkdir_perms(
    mode:int=511, parents:bool=False, exist_ok:bool=False, uid:int=-1, gid:int=-1
):

Create directory like Path.mkdir but optionally set uid/gid on newly created dirs

with tempfile.TemporaryDirectory() as tmpdir:
    base = Path(tmpdir)
    p1 = base/'test1'
    p1.mkdir_perms()
    assert p1.exists() and p1.is_dir()
    
    p2 = base/'a'/'b'/'c'
    p2.mkdir_perms(parents=True)
    assert p2.exists() and (base/'a').exists()
    
    p1.mkdir_perms(exist_ok=True)
    assert p1.exists()
    
    with expect_fail(FileExistsError): p1.mkdir_perms()
    with expect_fail(FileNotFoundError): (base/'missing'/'child').mkdir_perms()

source

atomic_save

def atomic_save(
    fn, mode:str='wb', uid:int=-1, gid:int=-1, **kwargs
):

Context manager for writing a file atomically via a temp file that is renamed on close

atomic_save writes a temporary file in the destination directory, then renames it on success. Readers never see a partial write. If writing fails, the original file is unchanged:

with tempfile.TemporaryDirectory() as d:
    fn = Path(d)/'test.txt'
    fn.write_text('original')
    with atomic_save(fn) as f: f.write(b'new')
    test_eq(fn.read_text(), 'new')

source

load_mod

def load_mod(
    name, path
):

Load module name from file path

We can get a module spec from a module name:

import importlib.util

spec = importlib.util.find_spec('fastcore.basics')
spec

ModuleSpec(name='fastcore.basics', loader=<_frozen_importlib_external.SourceFileLoader object>, origin='/Users/jhoward/aai-ws/fastcore/fastcore/basics.py')

…then we can load it, using the origin path:

m = load_mod('fastcore.basics', spec.origin)
test_eq(m.__name__, 'fastcore.basics')
assert hasattr(m, 'store_attr')

source

import_no_init

def import_no_init(
    name
):

Import dotted name without running any __init__.py

m = import_no_init('fastcore.basics')
test_eq(m.__name__, 'fastcore.basics')
assert hasattr(m, 'store_attr')

m = import_no_init('fastcore')
assert hasattr(m, '__file__')
with expect_fail(): import_no_init('nonexistent_xyz')

Config

Config reads and writes an ini file with one DEFAULT section. Access keys as attributes, items, or with .get(key, default). Use create= to supply initial contents for a missing file.

types= converts values on read. Path values resolve relative to the config file’s directory. Config.find(name) searches a directory and its parents for the file.

source

save_config_file

def save_config_file(
    file, d, **kwargs
):

Write settings dict to a new config file, or overwrite the existing one.

source

read_config_file

def read_config_file(
    file, **kwargs
):

Config files are saved and read using Python’s configparser.ConfigParser, inside the DEFAULT section.

_d = dict(user='fastai', lib_name='fastcore', some_path='test', some_bool=True, some_num=3)
try:
    save_config_file('tmp.ini', _d)
    res = read_config_file('tmp.ini')
finally: os.unlink('tmp.ini')
dict(res)

{'user': 'fastai',
 'lib_name': 'fastcore',
 'some_path': 'test',
 'some_bool': 'True',
 'some_num': '3'}

source

find_file_parents

def find_file_parents(
    fname, frompath:NoneType=None
):

Search cfg_path and its parents to find cfg_name

save_config_file('../tmp.ini', _d)
try: found = find_file_parents('tmp.ini')
finally: os.unlink('../tmp.ini')
found

Path('/Users/jhoward/aai-ws/fastcore')

source

Config

def Config(
    cfg_path, cfg_name, create:NoneType=None, save:bool=True, extra_files:NoneType=None, types:NoneType=None,
    **cfg_kwargs
):

Reading and writing ConfigParser ini files

Config is a convenient wrapper around ConfigParser ini files with a single section (DEFAULT).

Instantiate a Config from an ini file at cfg_path/cfg_name:

save_config_file('../tmp.ini', _d)
try: cfg = Config('..', 'tmp.ini')
finally: os.unlink('../tmp.ini')
cfg

{'user': 'fastai', 'lib_name': 'fastcore', 'some_path': 'test', 'some_bool': 'True', 'some_num': '3'}

You can create a new file if one doesn’t exist by providing a create dict:

try: cfg = Config('..', 'tmp.ini', create=_d)
finally: os.unlink('../tmp.ini')
cfg

{'user': 'fastai', 'lib_name': 'fastcore', 'some_path': 'test', 'some_bool': 'True', 'some_num': '3'}

If you additionally pass save=False, the Config will contain the items from create without writing a new file:

cfg = Config('..', 'tmp.ini', create=_d, save=False)
test_eq(cfg.user,'fastai')
assert not Path('../tmp.ini').exists()

Config passes extra keyword arguments to ConfigParser. Inline comments are disabled by default. Set inline_comment_prefixes to your comment markers, as in this # example:

# Create a complete example config file with comments
cfg_str = """\
[DEFAULT]
user = fastai # inline comment

# Library configuration
lib_name = fastcore

# Paths
some_path = test 

# Feature flags
some_bool = True

# Numeric settings
some_num = # missing value
"""

with open('../tmp.ini', 'w') as f:
    f.write(cfg_str)

# Now read it back to verify
try: cfg = Config('..', 'tmp.ini', inline_comment_prefixes=('#'))
finally: os.unlink('../tmp.ini')
test_eq(cfg.user,'fastai')
test_eq(cfg.some_num,'')

source

Config.get

def get(
    k, default:NoneType=None
):

Keys can be accessed as attributes, items, or with get and an optional default:

test_eq(cfg.user,'fastai')
test_eq(cfg['some_path'], 'test')
test_eq(cfg.get('foo','bar'),'bar')

Extra files can be read before cfg_path/cfg_name using extra_files, in the order they appear:

with tempfile.TemporaryDirectory() as d:
    a = Config(d, 'a.ini', {'a':0,'b':0})
    b = Config(d, 'b.ini', {'a':1,'c':0})
    c = Config(d, 'c.ini', {'a':2,'d':0}, extra_files=[a.config_file,b.config_file])
    test_eq(c.d, {'a':'2','b':'0','c':'0','d':'0'})

Pass a {key: type} mapping in types to convert values when reading them. Keys without a type return strings. For Path values, relative paths resolve from the config file’s directory. bool values use str2bool.

_types = dict(some_path=Path, some_bool=bool, some_num=int)
cfg = Config('..', 'tmp.ini', create=_d, save=False, types=_types)

test_eq(cfg.user,'fastai')
test_eq(cfg['some_path'].resolve(), (Path('..')/'test').resolve())
test_eq(cfg.get('some_num'), 3)

source

Config.find

def find(
    cfg_name, cfg_path:NoneType=None, **kwargs
):

Search cfg_path and its parents to find cfg_name

You can use Config.find to search a path and its parents for a config file, starting in the current path if no path is specified:

Collections

dict2obj converts nested dicts to AttrDict and lists to L, allowing attribute access such as d.b.c. obj2dict reverses the conversion:

source

Unset

def Unset(
    *args, **kwds
):

Create a collection of name/value pairs.

Example enumeration:

class Color(Enum): … RED = 1 … BLUE = 2 … GREEN = 3

Access them by:

attribute access:

Color.RED <Color.RED: 1>
value lookup:

Color(1) <Color.RED: 1>
name lookup:

Color[‘RED’] <Color.RED: 1>

Enumerations can be iterated over, and know how many members they have:

len(Color) 3

list(Color) [<Color.RED: 1>, <Color.BLUE: 2>, <Color.GREEN: 3>]

Methods can be added to enumerations, and members can have their own attributes – see the documentation for details.

source

dict2obj

def dict2obj(
    d:__main__.Unset=UNSET, list_func:fastcore.foundation._L_Meta=L, dict_func:type=AttrDict, **kwargs
):

Convert (possibly nested) dicts (or lists of dicts) to AttrDict

d = dict2obj({'a':1, 'b':{'c':2, 'd':[3,4]}})
test_eq(d.b.c, 2)
test_eq(d.b.d[1], 4)

This is a convenience to give you “dotted” access to (possibly nested) dictionaries, e.g:

d1 = dict(a=1, b=dict(c=2,d=3))
d2 = dict2obj(d1)
test_eq(d2.b.c, 2)
test_eq(d2.b['c'], 2)

kwargs can also be used:

d3 = dict2obj(a=1, b=dict(c=2,d=3))
test_eq(d3.b.c, 2)
test_eq(d3.b['c'], 2)

It can also be used on lists of dicts.

_list_of_dicts = [d1, d1]
ds = dict2obj(_list_of_dicts)
test_eq(ds[0].b.c, 2)

None values should be preserved:

d3 = dict2obj(a=1, b=dict(c=2,d=None))
test_eq(d3.b['d'], None)

source

obj2dict

def obj2dict(
    d
):

Convert (possibly nested) AttrDicts (or lists of AttrDicts) to dict

obj2dict can be used to reverse what is done by dict2obj:

test_eq(obj2dict(d2), d1)
test_eq(obj2dict(ds), _list_of_dicts)

source

repr_dict

def repr_dict(
    d
):

Print nested dicts and lists, such as returned by dict2obj

print(repr_dict(d2))

- a: 1
- b: 
  - c: 2
  - d: 3

source

is_listy

def is_listy(
    x
):

isinstance(x, (tuple,list,L,slice,Generator,set,frozenset))

assert is_listy((1,))
assert is_listy([1])
assert is_listy(L([1]))
assert is_listy(slice(2))
assert not is_listy(array([1]))

source

mapped

def mapped(
    f, it
):

map f over it, unless it’s not listy, in which case return f(it)

def _f(x,a=1): return x-a

test_eq(mapped(_f,1),0)
test_eq(mapped(_f,[1,2]),[0,1])
test_eq(mapped(_f,(1,)),(0,))

source

take_lines

def take_lines(
    s, pat, atstart:bool=True, atend:bool=False, drop:bool=False
):

Take contiguous lines matching pat from either end of s

take_lines selects contiguous lines matching pat from the start, end, or both. By default it returns the selected lines; drop=True returns everything else. Original line endings are preserved.

By default, matching lines are taken only from the start.

s = '# heading\n\n%%time\nx = 1\n'
pat = r'^(#|%%|$)'

test_eq(take_lines(s, pat), '# heading\n\n%%time\n')
test_eq(take_lines(s, pat, drop=True), 'x = 1\n')

s = '# head\nx = 1\n# tail\n'
test_eq(take_lines(s, r'^#', atend=True), '# head\n# tail\n')
take_lines(s, r'^#', atend=True, drop=True)

'x = 1\n'

Extensions to Pathlib.Path

These methods extend Pathlib.Path:

ls() lists directory contents as an L, optionally filtered by MIME prefix or extension.
read_json, write_json, and read_jsonl handle JSON files.
readlines reads lines; mk_write creates parent directories before writing.
delete, relpath, and normpath provide file deletion and path conversions.

Set Path.BASE_PATH to display paths relative to that directory.

source

Path.readlines

def readlines(
    hint:int=-1, encoding:str='utf8'
):

Read the content of self

source

Path.read_json

def read_json(
    encoding:NoneType=None, errors:NoneType=None
):

Same as read_text followed by loads

source

Path.read_jsonl

def read_jsonl(
    encoding:NoneType=None, errors:NoneType=None
):

Parse newline-delimited JSON, returning one object per line

JSON Lines uses \n to separate records. splitlines() also splits at U+0085, U+2028, and U+2029, which can occur inside valid JSON strings:

p = Path(tempfile.mkdtemp())/'log.jsonl'
p.write_text('{"id": 1, "msg": "durable execution,\u2028backend development"}\n{"id": 2, "msg": "plain"}\n')
test_eq(len(p.read_text().splitlines()), 3)

read_jsonl splits records only at \n:

recs = p.read_jsonl()
test_eq(len(recs), 2)
recs

[{'id': 1, 'msg': 'durable execution,\u2028backend development'},
 {'id': 2, 'msg': 'plain'}]

Parse errors include the file and line number:

with tempfile.TemporaryDirectory() as d:
    p = Path(d)/'bad.jsonl'
    p.write_text('{"id": 1}\n{"id": 2\n')
    with expect_fail(ValueError, contains=f'{p}:2'): p.read_jsonl()

source

Path.mk_write

def mk_write(
    data, encoding:NoneType=None, errors:NoneType=None, mode:int=511, uid:int=-1, gid:int=-1
):

Make all parent dirs of self, and write data

source

Path.write_json

def write_json(
    data, encoding:NoneType=None, errors:NoneType=None, mode:int=511, uid:int=-1, gid:int=-1, **kw
):

Same as dumpsfollowed by mk_write

source

Path.relpath

def relpath(
    start:NoneType=None
):

Same as os.path.relpath, but returns a Path, and resolves symlinks

p = Path('../fastcore/').resolve()
p

Path('/Users/jhoward/aai-ws/fastcore/fastcore')

p.relpath(Path.cwd())

Path('../fastcore')

source

Path.ls

def ls(
    n_max:NoneType=None, file_type:NoneType=None, file_exts:NoneType=None
):

Contents of path as a list

We add an ls() method to pathlib.Path which is simply defined as list(Path.iterdir()), mainly for convenience in REPL environments such as notebooks.

path = Path()
t = path.ls()
assert len(t)>0
t1 = path.ls(10)
test_eq(len(t1), 10)
t2 = path.ls(file_exts='.ipynb')
assert len(t)>len(t2)
t[0]

Path('llms.txt')

You can also pass an optional file_type MIME prefix and/or a list of file extensions.

lib_path = (path/'../fastcore')
txt_files=lib_path.ls(file_type='text')
assert len(txt_files) > 0 and txt_files[0].suffix=='.py'
ipy_files=path.ls(file_exts=['.ipynb'])
assert len(ipy_files) > 0 and ipy_files[0].suffix=='.ipynb'
txt_files[0],ipy_files[0]

(Path('../fastcore/shutil.py'), Path('000_tour.ipynb'))

source

Path.normpath

def normpath():

Normalize path, eliminating double slashes, etc.

normpath normalizes a path by collapsing redundant separators and up-level references (e.g., ..).

p = Path('foo//bar/../baz')
p.normpath()

Path('foo/baz')

source

Path.repr

def __repr__():

Return repr(self).

Set Path.BASE_PATH to display paths within that directory relative to it. Paths outside the directory keep their existing representation:

t = ipy_files[0].absolute()
try:
    Path.BASE_PATH = t.parent.parent
    test_eq(repr(t), f"Path('nbs/{t.name}')")
finally: Path.BASE_PATH = None

source

Path.delete

def delete():

Delete a file, symlink, or directory tree

Reindexing Collections

source

ReindexCollection

def ReindexCollection(
    coll, idxs:NoneType=None, cache:NoneType=None, tfm:function=noop
):

Reindexes collection coll with indices idxs and optional LRU cache of size cache

ReindexCollection changes the order in which you read a collection without rearranging its contents. fastai uses it to prepare data for language models.

Pass the index order in idxs, or change it later with reindex. Use descending indices to read the list in reverse order:

rc=ReindexCollection(['a', 'b', 'c', 'd', 'e'], idxs=[4,3,2,1,0])
list(rc)

['e', 'd', 'c', 'b', 'a']

Alternatively, you can use the reindex method:

source

ReindexCollection.reindex

def reindex(
    idxs
):

Replace self.idxs with idxs

rc=ReindexCollection(['a', 'b', 'c', 'd', 'e'])
rc.reindex([4,3,2,1,0])
list(rc)

['e', 'd', 'c', 'b', 'a']

You can optionally specify a LRU cache, which uses functools.lru_cache upon instantiation:

sz = 50
t = ReindexCollection(L.range(sz), cache=2)

#trigger a cache hit by indexing into the same element multiple times
t[0], t[0]
t._get.cache_info()

CacheInfo(hits=1, misses=1, maxsize=2, currsize=1)

You can optionally clear the LRU cache by calling the cache_clear method:

source

ReindexCollection.cache_clear

def cache_clear():

Clear LRU cache

sz = 50
t = ReindexCollection(L.range(sz), cache=2)

#trigger a cache hit by indexing into the same element multiple times
t[0], t[0]
t.cache_clear()
t._get.cache_info()

CacheInfo(hits=0, misses=0, maxsize=2, currsize=0)

source

ReindexCollection.shuffle

def shuffle():

Randomly shuffle indices

Without idxs, ReindexCollection starts in the collection’s original order. shuffle changes the index order:

rc=ReindexCollection(['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h'])
rc.shuffle()
list(rc)

['d', 'g', 'a', 'c', 'e', 'h', 'b', 'f']

sz = 50
t = ReindexCollection(L.range(sz), cache=2)
test_eq(list(t), range(sz))
test_eq(t[sz-1], sz-1)
test_eq(t._get.cache_info().hits, 1)
t.shuffle()
test_eq(t._get.cache_info().hits, 1)
test_ne(list(t), range(sz))
test_eq(set(t), set(range(sz)))
t.cache_clear()
test_eq(t._get.cache_info().hits, 0)
test_eq(t.count(0), 1)

`SaveReturn` and `save_iter` Variants

A generator can return a final value as well as yielding values. Normal iteration discards that return value:

def example_generator():
    total = 0
    for i in range(3):
        total += i
        yield i
    return total  # This gets lost!

# The return value (3) is lost
values = list(example_generator())  # [0, 1, 2]

source

SaveReturn

def SaveReturn(
    its
):

Wrap an iterator such that the generator function’s return value is stored in .value

Exported source

class SaveReturn:
    "Wrap an iterator such that the generator function's return value is stored in `.value`"
    def __init__(self, its): self.its = its
    def __iter__(self):
        self.value = yield from self.its
        return self.value

SaveReturn wraps a non-async generator and captures its return value. It uses yield from, whose result is the value returned by the generator:

def sum_range(n):
    total = 0
    for i in range(n):
        total += i
        yield i
    return total  # This value is returned by yield from

sr = SaveReturn(sum_range(5))
values = list(sr)  # This will consume the generator and get the return value
print(f"Values: {values}")
sr.value

Values: [0, 1, 2, 3, 4]

In order to provide an accurate signature for save_iter, we need a version of wraps that removes leading parameters:

source

trim_wraps

def trim_wraps(
    f, n:int=1
):

Like wraps, but removes the first n parameters from the signature

trim_wraps is a decorator factory that works like functools.wraps, but removes the first n parameters from the wrapped function’s signature. This is useful when creating wrapper functions that consume some parameters internally and shouldn’t expose them in the public API.

def adder(base, x, y): return base + x + y

def make_adder(base_value):
    @trim_wraps(adder)
    def _(x, y): return adder(base_value, x, y)
    return _

add_10 = make_adder(10)
print(f"{add_10.__name__}{inspect.signature(add_10)}")

adder(x, y)

source

save_iter

def save_iter(
    g
):

Decorator that allows a generator function to store values in the returned iterator object

save_iter passes the returned iterator to the generator as its first argument. You can store multiple attributes on that object at any point during iteration. Here o.value holds the final sum:

@save_iter
def sum_range(o, n):  # Note: 'o' parameter added
    total = 0
    for i in range(n):
        total += i
        yield i
    o.value = total  # Store directly on the iterator object

Call sum_range with only n. save_iter supplies o:

print(sum_range.__signature__)

(n)

sr = sum_range(5)
print(f"Values: {list(sr)}")
print(f"Sum stored: {sr.value}")

Values: [0, 1, 2, 3, 4]
Sum stored: 10

source

asave_iter

def asave_iter(
    g
):

Like save_iter, but for async iterators

asave_iter works like save_iter for async generators. These cannot return a value or use yield from, so SaveReturn does not apply.

@asave_iter
async def asum_range(self, n):
    total = 0
    for i in range(n):
        total += i
        yield i
    self.value = total

asr = asum_range(5)
print(f"Values: {[o async for o in asr]}")
print(f"Sum stored: {asr.value}")

Values: [0, 1, 2, 3, 4]
Sum stored: 10

Other Helpers

Other utilities include:

exec_eval: run a code string and return its last expression, like a notebook cell.
fenced: wrap text in a Markdown fence longer than any fence character run in the body.
str_diff: return a unified diff, or '' for identical text.
unqid and friendly_name: generate random Python identifiers or memorable word-based names.
flexicache: cache results with configurable expiration policies.

source

frontmatter

def frontmatter(
    txt:str, strvals:bool=False
)->tuple:

Tuple of (dict, body) from frontmatter in txt; missing frontmatter returns ({}, txt), malformed YAML raises

source

strloader

def strloader():

yaml.BaseLoader extended to resolve true/True/false/False, and nothing else, to bool

The closing fence can end the text without a trailing newline.

strvals=True keeps scalar values as strings while preserving YAML’s lists and mappings. This is useful when reading amounts and dates from legal forms.

Unquoted true, True, false, and False become booleans. Keeping 'false' as a string would make a flag test treat it as true. Quoted values and yes/no/on/off remain strings:

test_eq(frontmatter('---\ntitle: "Hi there"\ntags: [a,b]\n---\nBody'), ({'title':'Hi there', 'tags':['a','b']}, 'Body'))
test_eq(frontmatter('---\nbroken: [a,'), ({}, '---\nbroken: [a,'))
test_eq(frontmatter('No frontmatter here'), ({}, 'No frontmatter here'))
test_eq(frontmatter('---\njust text no closing'), ({}, '---\njust text no closing'))
test_eq(frontmatter('---\n---\nBody'), ({}, '---\n---\nBody'))
test_fail(lambda: frontmatter('---\nbad: "unclosed\n---\nB'))  # malformed raises; absent stays ({},txt)
test_eq(frontmatter('---\nt: v\n...\nBody'), ({'t':'v'}, 'Body'))
test_eq(frontmatter('---\nt: v\n---'), ({'t':'v'}, ''))
test_eq(frontmatter('---\nn: 1000\ngs:\n  - {d: X, n: 1}\n---\nB', strvals=True), ({'n':'1000','gs':[{'d':'X','n':'1'}]}, 'B'))
test_eq(frontmatter('---\na: true\nb: False\nc: TRUE\nq: "true"\nnm: Norway\nn: no\n---\nB', strvals=True), (dict(a=True, b=False, c='TRUE', q='true', nm='Norway', n='no'), 'B'))

source

clean_cli_output

def clean_cli_output(
    txt:str, strip:bool=True
):

Render CLI output as it appears on screen: alternate screen, \r overwrites, \b backspaces, and ANSI escapes

# Alternate screen → empty
test_eq(clean_cli_output('hello\x1b[?1049hworld'), '')

# Carriage return keeps last segment
test_eq(clean_cli_output('downloading...\rprogress 50%\rdone!'), 'done!')

# Multi-line with \r
test_eq(clean_cli_output('line1\nfoo\rbar\nline3'), 'line1\nbar\nline3')

# Trailing \r overwrites nothing: the screen still shows the text
test_eq(clean_cli_output('abc\r'), 'abc')
test_eq(clean_cli_output('download\rdone\r'), 'done')

# Backspace: a following char overwrites the previous one; trailing \b just moves the cursor
test_eq(clean_cli_output('hellp\bo world'), 'hello world')
test_eq(clean_cli_output('ab\b'), 'ab')
test_eq(clean_cli_output('\bx'), 'x')

# \b never crosses lines
test_eq(clean_cli_output('a\n\bb'), 'a\nb')

# ANSI stripping
test_eq(clean_cli_output('\x1b[31mred\x1b[0m text'), 'red text')

# OSC sequences
test_eq(clean_cli_output('\x1b]0;title\x07hello'), 'hello')

# strip_ansi=False preserves escapes
test_eq(clean_cli_output('\x1b[31mred\x1b[0m', strip=False), '\x1b[31mred\x1b[0m')

# Plain text unchanged
test_eq(clean_cli_output('just plain text\nline two'), 'just plain text\nline two')

# PTY \r\n line endings normalized to \n
test_eq(clean_cli_output('hello\r\n'), 'hello\n')
test_eq(clean_cli_output('line1\r\nline2\r\n'), 'line1\nline2\n')

# After returning from alternate screen, remaining output should be kept
test_eq(clean_cli_output('before\x1b[?1049hscreen stuff\x1b[?1049lafter'), 'after')

source

unqid

def unqid(
    seeded:bool=False
):

Generate a unique id suitable for use as a Python identifier

unqid generates a random unique identifier that is safe to use as a Python variable name (starts with _, uses only alphanumeric characters and underscores). It’s based on UUID4, encoded in URL-safe base64.

If seeded=True, uses random.getrandbits which respects random.seed(), making it reproducible. Otherwise uses uuid4() which is always random.

unqid()

'_sWGFSh8ETTq4er7bX4DUag'

With seeding for reproducibility:

random.seed(42)
a = unqid(seeded=True)
random.seed(42)
b = unqid(seeded=True)
test_eq(a, b)

Without seeding - always unique:

test_ne(unqid(), unqid())

source

rtoken_hex

def rtoken_hex(
    nbytes:int=16, # Number of bytes to generate
)->str: # hex string of length nbytes*2

Generate a random hex string using Python’s random module.

This is the same as secrets.token_hex, but is reproducible/seedable.

import secrets

secrets.token_hex(4),rtoken_hex(4)

('f8d349de', '8c7d7247')

source

friendly_name

def friendly_name(
    levels:int=3, suffix:int=4
):

Generate a random human-readable name with customizable word levels and suffix length

friendly_name generates random, human-readable names by combining adjectives, nouns, verbs, and adverbs with a random alphanumeric suffix. This is useful for creating memorable identifiers for temporary files, test data, or user-friendly resource names.

friendly_name()  # Default: 3 word levels + 4-char suffix

'objective-forest-builds-0y6d'

Names are hyphen-separated and follow the pattern adjective-noun-verb-adverb, randomly chosen from lists of size 102, 116, 110, and 30, respectively. The levels param selects how many of the names to include:

friendly_name(2)  # 2 words + 4-char suffix

'lavender-hummingbird-divu'

suffix sets the length of the random alphanumeric ending. Each suffix item is taken from the 36 options of lowercase letters plus digits.

friendly_name(4, 6)  # All 4 word types + 6-char suffix

'elated-koala-begins-softly-zpqk51'

source

n_friendly_names

def n_friendly_names(
    levels:int=3, suffix:int=4
):

Number of possible combos for `friendly_names

The number of combinations if all levels are included is:

print(f'{n_friendly_names(4):,}')

65,581,614,489,600

The default settings give:

print(f'{n_friendly_names():,}')

2,186,053,816,320

source

exec_eval

def exec_eval(
    code, # Code to exec/eval
    g:NoneType=None, # Globals namespace dict
    l:NoneType=None, # Locals namespace dict
):

Evaluate code in g (defaults to globals()) and l (defaults to locals())

This is a combination of eval and exec, which behaves like ipython and Jupyter. If the last line is an expression, it is evaluated and the result is returned:

exec_eval('''
def f(x): return x+1
f(1)
''')

By default, the code uses the caller’s globals and locals. For instance, here f is available since it’s been added to our symbol table:

exec_eval('print(f(2))')

Pass a dict as the g param in order to use an arbitrary namespace:

exec_eval('print(f)', {'f': 'Hi I am f.'})

Hi I am f.

This function helps us identify the first declared raw function of a dispatched function:

from plum import Function

def f1(x): return "Any"
def f2(x:int): return "Int"

df = Function(f1).dispatch(f1).dispatch(f2)

test_eq(_unwrapped_type_dispatch_func(df), f1)

source

get_source_link

def get_source_link(
    func
):

Return link to func in source code

get_source_link allows you get a link to source code related to an object. For nbdev related projects such as fastcore, we can get the full link to a GitHub repo. For nbdev projects, be sure to properly set the git_url in settings.ini (derived from lib_name and branch on top of the prefix you will need to adapt) so that those links are correct.

For example, below we get the link to fastcore.test.test_eq:

from fastcore.test import test_eq

assert 'fastcore/test.py' in get_source_link(test_eq)
assert get_source_link(test_eq).startswith('https://github.com/AnswerDotAI/fastcore')
get_source_link(test_eq)

'https://github.com/AnswerDotAI/fastcorefastcore/test.py#L76'

source

sparkline

def sparkline(
    data, mn:NoneType=None, mx:NoneType=None, empty_zero:bool=False
):

Sparkline for data, with Nones (and zero, if empty_zero) shown as empty column

data = [9,6,None,1,4,0,8,15,10]
print(f'without "empty_zero": {sparkline(data, empty_zero=False)}')
print(f'   with "empty_zero": {sparkline(data, empty_zero=True )}')

without "empty_zero": ▅▂ ▁▂▁▃▇▅
   with "empty_zero": ▅▂ ▁▂ ▃▇▅

You can set a maximum and minimum for the y-axis of the sparkline with the arguments mn and mx respectively:

sparkline([1,2,3,400], mn=0, mx=3)

'▂▅▇▇'

source

modify_exception

def modify_exception(
    e:Exception, # An exception
    msg:str=None, # A custom message
    replace:bool=False, # Whether to replace e.args with [msg]
)->Exception:

Modifies e with a custom message attached

msg = "This is my custom message!"

with expect_fail(Exception, ''): (_ for _ in ()).throw(modify_exception(Exception(), None))
with expect_fail(Exception, msg): (_ for _ in ()).throw(modify_exception(Exception(), msg))
with expect_fail(Exception, "The first message This is my custom message!"): (_ for _ in ()).throw(modify_exception(Exception("The first message"), msg))
with expect_fail(Exception, "This is my custom message!"): (_ for _ in ()).throw(modify_exception(Exception("The first message"), msg, True))

source

round_multiple

def round_multiple(
    x, mult, round_down:bool=False
):

Round x to nearest multiple of mult

test_eq(round_multiple(63,32), 64)
test_eq(round_multiple(50,32), 64)
test_eq(round_multiple(40,32), 32)
test_eq(round_multiple( 0,32),  0)
test_eq(round_multiple(63,32, round_down=True), 32)
test_eq(round_multiple((63,40),32), (64,32))

source

set_num_threads

def set_num_threads(
    nt
):

Get numpy (and others) to use nt threads

This sets the number of threads consistently for many tools, by:

Set the following environment variables equal to nt: OPENBLAS_NUM_THREADS,NUMEXPR_NUM_THREADS,OMP_NUM_THREADS,MKL_NUM_THREADS
Sets nt threads for numpy and pytorch.

source

join_path_file

def join_path_file(
    file, path, ext:str=''
):

Return path/file if file is a string or a Path, file otherwise

path = Path.cwd()/'_tmp'/'tst'
f = join_path_file('tst.txt', path)
assert path.exists()
test_eq(f, path/'tst.txt')
with open(f, 'w') as f_: assert join_path_file(f_, path) == f_
shutil.rmtree(Path.cwd()/'_tmp')

source

autostart

def autostart(
    g
):

Decorator that automatically starts a generator

source

EventTimer

def EventTimer(
    store:int=5, span:int=60
):

An event timer with history of store items of time span

Add events with add, and get number of events and their frequency (freq).

# Random wait function for testing
def _randwait(): yield from (sleep(random.random()/200) for _ in range(100))

c = EventTimer(store=5, span=0.03)
for o in _randwait(): c.add(1)
print(f'Num Events: {c.events}, Freq/sec: {c.freq:.01f}')
print('Most recent: ', sparkline(c.hist), *L(c.hist).map('{:.01f}'))

Num Events: 9, Freq/sec: 334.2
Most recent:  ▃▂▇▁▆ 263.0 253.9 283.5 243.7 273.8

source

stringfmt_names

def stringfmt_names(
    s:str
)->list:

Unique brace-delimited names in s

s = '/pulls/{pull_number}/reviews/{review_id}'
test_eq(stringfmt_names(s), ['pull_number','review_id'])

source

PartialFormatter

def PartialFormatter():

A string.Formatter that doesn’t error on missing fields, and tracks missing fields and unused args

source

partial_format

def partial_format(
    s:str, **kwargs
):

string format s, ignoring missing field errors, returning missing and extra fields

The result is a tuple of (formatted_string,missing_fields,extra_fields), e.g:

res,missing,xtra = partial_format(s, pull_number=1, foo=2)
test_eq(res, '/pulls/1/reviews/{review_id}')
test_eq(missing, ['review_id'])
test_eq(xtra, {'foo':2})

source

truncstr

def truncstr(
    s:str, maxlen:int, suf:str='…', space:str='', sizevar:str=None
)->str:

Truncate s to length maxlen, adding suffix suf if truncated; a callable suf gets the number of characters cut

A callable suf receives the number of characters removed, including those displaced by the suffix itself:

w = 'abacadabra'
test_eq(truncstr(w, 10), w)
test_eq(truncstr(w, 5), 'abac…')
test_eq(truncstr(w, 5, suf=''), 'abaca')
test_eq(truncstr(w, 11, space='_'), w+"_")
test_eq(truncstr(w, 10, space='_'), w[:-1]+'…')
test_eq(truncstr(w, 5, suf='!!'), 'aba!!')
test_eq(truncstr(w, 8, suf=lambda n: f'…[{n}]'), 'abac…[6]')
test_eq(truncstr('x'*5000, 20, suf=lambda n: f'…[{humanize(n)}]'), 'x'*15+'…[5k]')

Set sizevar='_n_' to substitute the original string length for {_n_} in truncstr’s suffix. Here (11) records the length before truncation:

test_eq(truncstr('hello world', 8, suf='…({_n_})', sizevar='_n_'), 'hel…(11)')

trunc_ctr keeps the head and tail of a long string, elides the middle, and marks the elision with its humanized size. Use it when both ends of a document matter, such as a summary at the top and the discussion at the end.

source

trunc_ctr

def trunc_ctr(
    s:str, mx:int=1000
)->str:

Truncate the middle of s so ~mx chars remain, marking the elision with its humanized size

d = 'summary ' + 'x'*30 + ' replies'
test_eq(trunc_ctr(d, 100), d)
test_eq(trunc_ctr(d, None), d)
trunc_ctr(d, 12)

'summar…[34 chars]…eplies'

Wrap a large tool result in TruncatedString to shorten its bare display without discarding text. Its repr uses trunc_ctr. Slicing, searching and print still use the complete string.

source

TruncatedString

def TruncatedString(
    *args, **kwargs
):

A str whose repr shows contents verbatim, middle-truncated to ~mx chars

t = TruncatedString(d, 12)
test_eq(t, d)
test_eq(len(t), len(d))
test_eq(t[:7], 'summary')
t

summar…[34 chars]…eplies

source

fenced

def fenced(
    text:str, info:str='', ch:str='`'
)->str:

Wrap text in a fence of ch, one char longer than any run of ch inside it (min 3)

When generated text goes inside a markdown code fence, the fence must be longer than any run of the fence character in the body, or the block ends early. fenced picks a safe fence: one char longer than the longest such run anywhere in the text, at least the standard three.

print(fenced('x=1', 'python'))
test_eq(fenced('x=1', 'python'), '```python\nx=1\n```')
s = 'a ````four```` backtick run'
test_eq(fenced(s), '`'*5 + '\n' + s + '\n' + '`'*5)

```python
x=1
```

Fence character runs count anywhere in the text, including mid-line. info is appended verbatim; include a leading space for forms such as ::: output. Trailing newlines are stripped. An empty body leaves one blank line between the fences.

test_eq(fenced('text', 'output', ch=':'), ':::output\ntext\n:::')
test_eq(fenced('x\n\n'), '```\nx\n```')
test_eq(fenced(''), '```\n\n```')

source

fenced_blocks

def fenced_blocks(
    text:str, # Markdown text to scan
    ch:str='`', # Fence character
)->list: # `(info, body, start, end)` per block; `start`/`end` are char offsets of the whole block

Top-level fenced blocks in text, fence-nesting-aware: the inverse of fenced

Only top-level blocks are returned. A closing fence must be at least as long as its opener and contain no other text. Shorter fences remain in the body. Unclosed blocks are omitted.

s = "before\n\n```json {.tool}\n{\"a\": 1}\n```\n\nafter\n"
blks = fenced_blocks(s)
test_eq(len(blks), 1)
info,body,start,end = blks[0]
test_eq(info, 'json {.tool}')
test_eq(body, '{"a": 1}\n')
test_eq(s[start:end], '```json {.tool}\n{"a": 1}\n```\n')
outer = fenced(s, 'markdown')                          # embed the whole thing in a longer fence
test_eq(fenced_blocks(outer)[0][0], 'markdown')        # only the top-level block reported
test_eq(fenced_blocks('```\nopen\n'), [])              # unclosed: no block
test_eq(fenced_blocks('::::a\nx\n::::\n', ch=':')[0][0], 'a')

str_diff returns a unified diff as a plain string, or '' when the texts match.

source

str_diff

def str_diff(
    a:str, # Original text
    b:str, # New text
    n:int=1, # Context lines around changes
    names:tuple=None, # Optional `(a_name,b_name)` labels; `---`/`+++` header omitted if None
)->str: # Unified diff, `''` when equal

Unified diff of a and b

test_eq(str_diff('a\nb\nc', 'a\nB\nc'), '@@ -1,3 +1,3 @@\n a\n-b\n+B\n c')
test_eq(str_diff('same', 'same'), '')
d = str_diff('x=1\n', 'x=2\n', names=('old.py','new.py'))
assert d.startswith('--- old.py\n+++ new.py\n')
assert '-x=1' in d and '+x=2' in d

source

utc2local

def utc2local(
    dt:datetime.datetime
)->datetime.datetime:

Convert dt from UTC to local time

dt = datetime(2000,1,1,12)
print(f'{dt} UTC is {utc2local(dt)} local time')

2000-01-01 12:00:00 UTC is 2000-01-01 22:00:00+10:00 local time

source

local2utc

def local2utc(
    dt:datetime.datetime
)->datetime.datetime:

Convert dt from local to UTC time

print(f'{dt} local is {local2utc(dt)} UTC time')

2000-01-01 12:00:00 local is 2000-01-01 02:00:00+00:00 UTC time

source

trace

def trace(
    f
):

Add set_trace to an existing function f

You can add a breakpoint to an existing function, e.g:

Path.cwd = trace(Path.cwd)
Path.cwd()

Now, when the function is called it will drop you into the debugger. Note, you must issue the s command when you begin to step into the function that is being traced.

source

modified_env

def modified_env(
    *delete, **replace
):

Context manager temporarily modifying os.environ by deleting delete and replacing replace

# USER isn't in Cloud Linux Environments
env_test = 'USERNAME' if sys.platform == "win32" else 'SHELL'
oldusr = os.environ[env_test]

replace_param = {env_test: 'a'}
with modified_env('PATH', **replace_param):
    test_eq(os.environ[env_test], 'a')
    assert 'PATH' not in os.environ

assert 'PATH' in os.environ
test_eq(os.environ[env_test], oldusr)

source

ContextManagers

def ContextManagers(
    mgrs
):

Wrapper for contextlib.ExitStack which enters a collection of context managers

source

shufflish

def shufflish(
    x, pct:float=0.04
):

Randomly relocate items of x up to pct of len(x) from their starting location

source

console_help

def console_help(
    libname:str
):

Show help for all console scripts from libname

source

hl_md

def hl_md(
    s, lang:str='html', show:bool=True
):

Syntax highlight s using lang.

When we display code in a notebook, it’s nice to highlight it, so we create a function to simplify that:

hl_md('<test><xml foo="bar">a child</xml></test>')

<test><xml foo="bar">a child</xml></test>

source

type2str

def type2str(
    typ:type
)->str:

Stringify typ

test_eq(type2str(Optional[float]), 'Union[float, None]')

source

dataclass_src

def dataclass_src(
    cls
):

DC = make_dataclass('DC', [('x', int), ('y', Optional[float], None), ('z', float, None)])
print(dataclass_src(DC))

@dataclass
class DC:
    x: int
    y: Union[float, None] = None
    z: float = None

source

nullable_dc

def nullable_dc(
    cls
):

Like dataclass, but default of UNSET added to fields without defaults

@nullable_dc
class Person: name: str; age: int; city: str = "Unknown"
Person(name="Bob")

Person(name='Bob', age=UNSET, city='Unknown')

source

make_nullable

def make_nullable(
    clas
):

@dataclass
class Person: name: str; age: int; city: str = "Unknown"

make_nullable(Person)
Person("Bob", city='NY')

Person(name='Bob', age=UNSET, city='NY')

Person(name="Bob")

Person(name='Bob', age=UNSET, city='Unknown')

Person("Bob", 34)

Person(name='Bob', age=34, city='Unknown')

source

flexiclass

def flexiclass(
    cls, # The class to convert
)-><function dataclass at 0x7f211338c680>:

Convert cls into a dataclass like make_nullable. Converts in place and also returns the result.

This can be used as a decorator…

@flexiclass
class Person: name: str; age: int; city: str = "Unknown"

bob = Person(name="Bob")
bob

Person(name='Bob', age=UNSET, city='Unknown')

…or can update the behavior of an existing class (or dataclass):

class Person: name: str; age: int; city: str = "Unknown"

flexiclass(Person)
bob = Person(name="Bob")
bob

Person(name='Bob', age=UNSET, city='Unknown')

Action occurs in-place:

class Person: name: str; age: int; city: str = "Unknown"

flexiclass(Person)
is_dataclass(Person)

True

source

asdict

def asdict(
    o
)->dict:

Convert o to a dict, supporting dataclasses, namedtuples, iterables, and __dict__ attrs.

Any UNSET values are not included.

asdict(bob)

{'name': 'Bob', 'city': 'Unknown'}

Set the optional __flds__ parameter to customise the field list, and the optional __skip__ parameter to skip some names.

class CustomObj:
    def __init__(self): self.a,self.b,self.c,self.d = 1,2,3,4
    __flds__ = ['a','b','c','d']
    __skip__ = ['b']

obj = CustomObj()
test_eq(asdict(obj), {'a': 1, 'c': 3, 'd': 4})

An object may be iterable without yielding key-value pairs; since dict can’t consume it, asdict falls back to its instance attributes.

class IterObj:
    def __init__(self): self.a,self.b = 1,2
    def __iter__(self): return iter([self.a,self.b])

test_eq(asdict(IterObj()), dict(a=1,b=2))

To customise dict conversion behavior for a class, implement the _asdict method (this is used in the Python stdlib for named tuples).

source

vars_pub

def vars_pub(
    x
):

Get public non-skipped vars

The vars_pub function returns a list of public (non-underscore-prefixed) variable names from an object, excluding any names listed in the object’s optional __skip__ attribute.

class TestObj:
    def __init__(self): self.pub_attr,self._priv_attr,self.another_pub,self.skip_me = 1,2,3,4
    __skip__ = ['skip_me']

obj = TestObj()
test_eq(vars_pub(obj), ['pub_attr', 'another_pub'])

Without __skip__, all pub vars are returned

class SimpleObj:
    def __init__(self): self.a,self._b,self.c = 1,2,3

simple = SimpleObj()
test_eq(vars_pub(simple), ['a', 'c'])

source

is_typeddict

def is_typeddict(
    cls:type
)->bool:

Check if cls is a TypedDict

class MyDict(TypedDict): name:str

assert is_typeddict(MyDict)
assert not is_typeddict({'a':1})

source

is_namedtuple

def is_namedtuple(
    cls
):

True if cls is a namedtuple type

assert is_namedtuple(namedtuple('tst', ['a']))
assert not is_namedtuple(tuple)

source

CachedIter

def CachedIter(
    o
):

Cache the result returned by an iterator

def f():
    yield 1
    return 2

r = CachedIter(f())
for o in r: print(o)
r.value

flexicache

source

flexicache

def flexicache(
    *funcs, maxsize:int=128
):

Like lru_cache, but customisable with policy funcs

flexicache adds expiration policies to an LRU cache. time_policy expires results after an interval; mtime_policy expires them when a file changes.

When caching a new result, flexicache calls each policy with None. Return the state to save for that cache entry. On later lookups, flexicache passes the saved state to the policy. Return a truthy value to expire the result, or a falsy value to reuse it. After recomputing an expired result, flexicache calls each policy with None to save new state.

Set a policy’s needs_args=True to receive (state, args, kwargs) instead. This lets mtime_policy(arg=...) watch a file named in the cached call’s arguments.

source

time_policy

def time_policy(
    seconds
):

A flexicache policy that expires cached items after seconds have passed

source

mtime_policy

def mtime_policy(
    filepath:NoneType=None, # File to watch, fixed at policy creation
    arg:NoneType=None, # Or: positional index or keyword name of a call argument naming the file to watch
):

A flexicache policy that expires cached items after the watched file’s modified-time changes

@flexicache(time_policy(10), mtime_policy('000_tour.ipynb'))
def cached_func(x, y): return x+y

cached_func(1,2)

@flexicache(time_policy(10), mtime_policy('000_tour.ipynb'))
async def cached_func(x, y): return x+y

print(await cached_func(1,2))
await cached_func(1,2)

fp = Path('flexi-test.txt')
fp.write_text('a')
@flexicache(mtime_policy(arg=0))
def slurp(p): return p.read_text()

test_eq(slurp(fp), 'a')
fp.write_text('b')
os.utime(fp, (time()+1,)*2)
test_eq(slurp(fp), 'b')
fp.write_text('c')
os.utime(fp, (time()-1,)*2)
test_eq(slurp(fp), 'c')

@flexicache(mtime_policy(arg='p'))
def slurp2(p=None): return p.read_text()

test_eq(slurp2(p=fp), 'c')
fp.unlink()

source

timed_cache

def timed_cache(
    seconds:int=60, maxsize:int=128
):

Like lru_cache, but also with time-based eviction

# demonstrate that flexicache is LRU
@flexicache(maxsize=2)
def cached_func(x): return time()

time_1 = cached_func(1)
test_eq(time_1, cached_func(1))

time_2 = cached_func(2)
test_eq(time_1, cached_func(1))
test_eq(time_2, cached_func(2))

time_3 = cached_func(3) # Removes 1

test_eq(time_2, cached_func(2)) # cache remains
test_eq(time_3, cached_func(3)) # cache remains
test_ne(time_1, cached_func(1)) # NEQ, removes 2
test_ne(time_2, cached_func(2))  # NEQ, removes 3
test_eq(cached_func(1), cached_func(1))

This function is a small convenience wrapper for using flexicache with time_policy.

@timed_cache(seconds=0.05, maxsize=2)
def cached_func(x): return x * 2, time()

# basic caching
result1, time1 = cached_func(2)
test_eq(result1, 4)
sleep(0.001)
result2, time2 = cached_func(2)
test_eq(result2, 4)
test_eq(time1, time2)

# caching different values
result3, _ = cached_func(3)
test_eq(result3, 6)

# maxsize
_, time4 = cached_func(4)
_, time2_new = cached_func(2)
test_close(time2, time2_new, eps=0.1)
_, time3_new = cached_func(3)
test_ne(time3_new, time())

# time expiration
sleep(0.05)
_, time4_new = cached_func(4)
test_ne(time4_new, time())

File Functions

walk_join

walk

exttypes

globtastic

pglob

maybe_open

mkdir

image_size

img_bytes

detect_mime

bunzip

loads

loads_multi

dumps

untar_dir

repo_details

shell

ssh

rsync_multi

run

open_file

save_pickle

load_pickle

parse_env

expand_wildcards

Path.mkdir_perms

atomic_save

load_mod

import_no_init

Config

save_config_file

read_config_file

find_file_parents

Config

Config.get

Config.find

Collections

Unset

dict2obj

obj2dict

repr_dict

is_listy

mapped

take_lines

Extensions to Pathlib.Path

Path.readlines

Path.read_json

Path.read_jsonl

Path.mk_write

Path.write_json

Path.relpath

Path.ls

Path.normpath

Path.__repr__

Path.delete

Reindexing Collections

ReindexCollection

ReindexCollection.reindex

ReindexCollection.cache_clear

ReindexCollection.shuffle

SaveReturn and save_iter Variants

SaveReturn

trim_wraps

save_iter

asave_iter

Other Helpers

frontmatter

strloader

clean_cli_output

unqid

rtoken_hex

friendly_name

n_friendly_names

exec_eval

get_source_link

sparkline

modify_exception

round_multiple

set_num_threads

Path.repr

`SaveReturn` and `save_iter` Variants