Coverage for /opt/anaconda/lib/python3.7/site-packages/llama/io/registry.py : 70%
Hot-keys on this page
r m x p toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# (c) Stefan Countryman 2019
3"""
4Abstract implementations of LLAMA IO interfaces.
5"""
7from urllib.parse import urlparse
8from threading import Lock
10_LOCK = Lock()
11_REGISTRY = []
14def register(io_class):
15 """
16 Register a ``llama.classes.IO`` implementation for use by LLAMA.
18 Parameters
19 ----------
20 io_class : llama.classes.IO
21 The ``IO`` implementation to register.
23 Raises
24 ------
25 ValueError
26 If ``scheme`` is already registered.
27 """
28 if not io_class.schemes:
29 raise ValueError(f"Must specify a scheme for {io_class}")
30 _LOCK.acquire()
31 try:
32 if io_class in _REGISTRY:
33 raise ValueError(f"{io_class} already in registry.")
34 for scheme in io_class.schemes:
35 if any(scheme in c.schemes for c in _REGISTRY):
36 raise ValueError(f"scheme already registered: {scheme}")
37 _REGISTRY.append(io_class)
38 finally:
39 _LOCK.release()
42def get_io(scheme):
43 """
44 Get the ``llama.classes.IO`` implementation registered under ``scheme``.
45 """
46 _LOCK.acquire()
47 try:
48 for io_class in _REGISTRY:
49 if scheme in io_class.schemes:
50 return io_class
51 raise KeyError(f"{scheme} not a registered IO scheme.")
52 finally:
53 _LOCK.release()
56def get_schemes():
57 """
58 Return a list of available IO scheme names.
59 """
60 _LOCK.acquire()
61 try:
62 return [s for io_class in _REGISTRY for s in io_class.schemes]
63 finally:
64 _LOCK.release()
67def io_class_from_rundir(rundir):
68 """
69 Parse the IO strategy from the scheme specified by the run directory.
70 """
71 url = urlparse(str(rundir))
72 return get_io(url.scheme)