-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathbase.py
144 lines (103 loc) · 3.75 KB
/
base.py
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
from abc import abstractmethod
from contextvars import ContextVar, Token
from pathlib import Path
from types import TracebackType
from typing import TYPE_CHECKING, Callable, Dict, Iterator, Optional, Tuple, Type, Union
from returns.functions import compose
from .utils import init_mix_ins
__all__ = [
"File",
"Project",
"Dir",
]
PathResolver = Callable[[Path], Path]
class _FileContainerMixIn:
def __init__(self) -> None:
self.__store: Dict[str, Union[File, Dir]] = dict()
def add(self, item: Union["File", "Dir"]) -> None:
assert item.name not in self.__store
self.__store[item.name] = item
def walk(self) -> Iterator[Tuple[PathResolver, "File"]]:
item: Union[File, Dir]
for item in self.__store.values():
def path_resolver(path: Path) -> Path:
return path / item.name
if isinstance(item, File):
yield path_resolver, item
else:
for subpath_resolver, subitem in item.walk():
yield compose(path_resolver, subpath_resolver), subitem
def subpaths(self) -> Iterator[str]:
return (str(path_resolver(Path("."))) for path_resolver, _ in self.walk())
if TYPE_CHECKING:
_ContextToken = Token[Optional[_FileContainerMixIn]]
else:
_ContextToken = Token
__context: ContextVar[Optional[_FileContainerMixIn]] = ContextVar(
"__context", default=None
)
def _context_get() -> _FileContainerMixIn:
project = __context.get()
assert project is not None
return project
def _context_set_root(value: _FileContainerMixIn) -> _ContextToken:
project = __context.get()
assert project is None
return __context.set(value)
def _context_set(value: _FileContainerMixIn) -> _ContextToken:
project = __context.get()
assert project is not None
return __context.set(value)
def _context_reset(token: _ContextToken) -> None:
project = __context.get()
assert project is not None
__context.reset(token)
class _ChildMixIn:
def __init__(self) -> None:
self.parent = _context_get()
assert isinstance(self, File) or isinstance(self, Dir)
self.parent.add(self)
class File(_ChildMixIn):
def __init__(self, name: str) -> None:
self.name = name
init_mix_ins(self, File)
@abstractmethod
def synth_content(self) -> str:
...
class _ContextMixIn(_FileContainerMixIn):
def __init__(self) -> None:
self.__context_token: Optional[_ContextToken] = None
init_mix_ins(self, _ContextMixIn)
def __enter__(self) -> None:
assert self.__context_token is None
if isinstance(self, Project):
self.__context_token = _context_set_root(self)
else:
self.__context_token = _context_set(self)
def __exit__(
self,
exc_type: Optional[Type[BaseException]],
exc_value: Optional[BaseException],
traceback: Optional[TracebackType],
) -> None:
assert self.__context_token is not None
_context_reset(self.__context_token)
self.__context_token = None
class Project(_ContextMixIn):
def synth(self, root: Optional[Path] = None) -> None:
if root is None:
root = Path.cwd()
root.mkdir(parents=True, exist_ok=True)
for path_resolver, f in self.walk():
path = path_resolver(root)
path.parent.mkdir(parents=True, exist_ok=True)
# ensure writable
if path.is_file():
path.chmod(0o644)
# write and mark read only
path.write_text(f.synth_content())
path.chmod(0o444)
class Dir(_ContextMixIn, _ChildMixIn):
def __init__(self, name: str) -> None:
self.name = name
init_mix_ins(self, Dir)