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
|
import click
import os
import os.path as op
import pathlib
import shlex
import stat
import subprocess as sp
@click.command()
@click.pass_context
@click.version_option()
@click.option(
"--export/--no-export",
default=None,
help="Export generated environment variables [default: --export]",
)
@click.argument(
"envdir",
default=str(pathlib.Path.home() / ".envdir"),
metavar="DIR",
)
def main(context, export, envdir):
r"""Load environment variables from DIR (or ~/.envdir).
For each non-directory entry in DIR, this will output a brief shell script
to set an environment with the same name. If the entry is an executable
program, it will be run (with the current environment) and its output will
be used as the value of the environment variable. Otherwise, it will be
read, and its content will be used. In either case, a single trailing
newline will be removed, if present.
This process skips any file which can't be run or read, as appropriate, and
outputs a warning on stderr.
The intended use case for this is in shell profiles, in the form:
eval "$(envdir-helper)"
The generated output is compatible with sh, and thus with bash and zsh.
"""
def warn_on_skipped(path, reason):
"""Log any skipped paths to stderr."""
stderr = click.get_text_stream("stderr")
click.echo(f"{context.info_name}: skipping {path}: {reason}", file=stderr)
if export is None:
env_script = detect_env_script(
envdir, rc=no_export_env_script, default=export_env_script
)
elif export:
env_script = export_env_script
else:
env_script = no_export_env_script
for name, content in walk_entries(envdir, on_skipped=warn_on_skipped):
script = env_script(name, content)
click.echo(script)
def detect_env_script(path, rc, default): # pylint: disable=invalid-name
"""Detect which of two values to use based on whether `path` ends with
`"rc"`. If it does, returns `rc`; otherwise, returns `default`.
"""
if path.endswith("rc"):
return rc
return default
def export_env_script(name, content):
"""Given a name and contents, generate a shell script that will set and
export the corresponding environment variable, with that content."""
# use sh-friendly syntax here: don't assume `export` can have assignment
# side effects, so don't use `export FOO=BAR`. `FOO=BAR; export FOO` is
# portable to all posix shells.
qname = shlex.quote(name)
qcontent = shlex.quote(content)
return f"{qname}={qcontent}; export {qname}"
def no_export_env_script(name, content):
"""Given a name and contents, generate a shell script that will set and NOT
export the corresponding environment variable, with that content."""
qname = shlex.quote(name)
qcontent = shlex.quote(content)
return f"{qname}={qcontent}"
def walk_entries(envdir, on_skipped):
"""Yields a name, value pair for each environment file in envdir.
The path for any skipped items (generally, failing or unrunnable programs,
or unreadable files) will be passed to the `on_skipped` callback, along with
the exception that caused it to be skipped.
"""
for name in sorted(os.listdir(envdir)):
path = op.join(envdir, name)
try:
path_stat = os.stat(path)
if directory(path_stat):
continue
if executable(path_stat):
content = from_program(path)
else:
content = from_file(path)
except Exception as reason: # pylint: disable=broad-except
on_skipped(path, reason)
continue
if content.endswith("\n"):
content = content[:-1]
yield name, content
def directory(item):
"""True iff item is a stat_result representing a directory."""
return stat.S_ISDIR(item.st_mode)
def executable(item):
"""True iff item is a stat_result representing something executable.
This doesn't distinguish between dirs and files; both are considered
executable if any +x bit is set.
"""
mode = stat.S_IMODE(item.st_mode)
mask = stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH
return mode & mask != 0
def from_program(path):
"""Reads a program's complete stdout into a string."""
result = sp.run(
[path],
check=True,
stdout=sp.PIPE,
)
return result.stdout.decode("UTF-8")
def from_file(path):
"""Reads a file's complete content into a string."""
with open(path, "r") as file:
return file.read()
|
