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
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
|
import os
import sys
from importlib.machinery import ModuleSpec
from importlib.util import module_from_spec, spec_from_file_location
from pathlib import Path
from types import ModuleType
from typing import Any, Optional, Tuple
from . import Config
from .exceptions import CollectionNotFound
from .util import debug
class Loader:
"""
Abstract class defining how to find/import a session's base `.Collection`.
.. versionadded:: 1.0
"""
def __init__(self, config: Optional["Config"] = None) -> None:
"""
Set up a new loader with some `.Config`.
:param config:
An explicit `.Config` to use; it is referenced for loading-related
config options. Defaults to an anonymous ``Config()`` if none is
given.
"""
if config is None:
config = Config()
self.config = config
def find(self, name: str) -> Optional[ModuleSpec]:
"""
Implementation-specific finder method seeking collection ``name``.
Must return a ModuleSpec valid for use by `importlib`, which is
typically a name string followed by the contents of the 3-tuple
returned by `importlib.module_from_spec` (``name``, ``loader``,
``origin``.)
For a sample implementation, see `.FilesystemLoader`.
.. versionadded:: 1.0
"""
raise NotImplementedError
def load(self, name: Optional[str] = None) -> Tuple[ModuleType, str]:
"""
Load and return collection module identified by ``name``.
This method requires a working implementation of `.find` in order to
function.
In addition to importing the named module, it will add the module's
parent directory to the front of `sys.path` to provide normal Python
import behavior (i.e. so the loaded module may load local-to-it modules
or packages.)
:returns:
Two-tuple of ``(module, directory)`` where ``module`` is the
collection-containing Python module object, and ``directory`` is
the string path to the directory the module was found in.
.. versionadded:: 1.0
"""
if name is None:
name = self.config.tasks.collection_name
spec = self.find(name)
if spec and spec.loader and spec.origin:
# Typically either tasks.py or tasks/__init__.py
source_file = Path(spec.origin)
# Will be 'the dir tasks.py is in', or 'tasks/', in both cases this
# is what wants to be in sys.path for "from . import sibling"
enclosing_dir = source_file.parent
# Will be "the directory above the spot that 'import tasks' found",
# namely the parent of "your task tree", i.e. "where project level
# config files are looked for". So, same as enclosing_dir for
# tasks.py, but one more level up for tasks/__init__.py...
module_parent = enclosing_dir
if spec.parent: # it's a package, so we have to go up again
module_parent = module_parent.parent
# Get the enclosing dir on the path
enclosing_str = str(enclosing_dir)
if enclosing_str not in sys.path:
sys.path.insert(0, enclosing_str)
# Actual import
module = module_from_spec(spec)
sys.modules[spec.name] = module # so 'from . import xxx' works
spec.loader.exec_module(module)
# Return the module and the folder it was found in
return module, str(module_parent)
msg = "ImportError loading {!r}, raising ImportError"
debug(msg.format(name))
raise ImportError
class FilesystemLoader(Loader):
"""
Loads Python files from the filesystem (e.g. ``tasks.py``.)
Searches recursively towards filesystem root from a given start point.
.. versionadded:: 1.0
"""
# TODO: could introduce config obj here for transmission to Collection
# TODO: otherwise Loader has to know about specific bits to transmit, such
# as auto-dashes, and has to grow one of those for every bit Collection
# ever needs to know
def __init__(self, start: Optional[str] = None, **kwargs: Any) -> None:
super().__init__(**kwargs)
if start is None:
start = self.config.tasks.search_root
self._start = start
@property
def start(self) -> str:
# Lazily determine default CWD if configured value is falsey
return self._start or os.getcwd()
def find(self, name: str) -> Optional[ModuleSpec]:
debug("FilesystemLoader find starting at {!r}".format(self.start))
spec = None
module = "{}.py".format(name)
paths = self.start.split(os.sep)
try:
# walk the path upwards to check for dynamic import
for x in reversed(range(len(paths) + 1)):
path = os.sep.join(paths[0:x])
if module in os.listdir(path):
spec = spec_from_file_location(
name, os.path.join(path, module)
)
break
elif name in os.listdir(path) and os.path.exists(
os.path.join(path, name, "__init__.py")
):
basepath = os.path.join(path, name)
spec = spec_from_file_location(
name,
os.path.join(basepath, "__init__.py"),
submodule_search_locations=[basepath],
)
break
if spec:
debug("Found module: {!r}".format(spec))
return spec
except (FileNotFoundError, ModuleNotFoundError):
msg = "ImportError loading {!r}, raising CollectionNotFound"
debug(msg.format(name))
raise CollectionNotFound(name=name, start=self.start)
return None
|
