Initial commit

This commit is contained in:
Senad Uka
2019-11-17 12:44:16 +01:00
parent e41eae7301
commit a3ef27c7a0
4894 changed files with 1771218 additions and 0 deletions

View File

@@ -0,0 +1,43 @@
from typing import Any, Optional # noqa
from .main import load_dotenv, get_key, set_key, unset_key, find_dotenv, dotenv_values
def load_ipython_extension(ipython):
# type: (Any) -> None
from .ipython import load_ipython_extension
load_ipython_extension(ipython)
def get_cli_string(path=None, action=None, key=None, value=None, quote=None):
# type: (Optional[str], Optional[str], Optional[str], Optional[str], Optional[str]) -> str
"""Returns a string suitable for running as a shell script.
Useful for converting a arguments passed to a fabric task
to be passed to a `local` or `run` command.
"""
command = ['dotenv']
if quote:
command.append('-q %s' % quote)
if path:
command.append('-f %s' % path)
if action:
command.append(action)
if key:
command.append(key)
if value:
if ' ' in value:
command.append('"%s"' % value)
else:
command.append(value)
return ' '.join(command).strip()
__all__ = ['get_cli_string',
'load_dotenv',
'dotenv_values',
'get_key',
'set_key',
'unset_key',
'find_dotenv',
'load_ipython_extension']

View File

@@ -0,0 +1,141 @@
import os
import sys
from subprocess import Popen
from typing import Any, Dict, List # noqa
try:
import click
except ImportError:
sys.stderr.write('It seems python-dotenv is not installed with cli option. \n'
'Run pip install "python-dotenv[cli]" to fix this.')
sys.exit(1)
from .main import dotenv_values, get_key, set_key, unset_key
from .version import __version__
@click.group()
@click.option('-f', '--file', default=os.path.join(os.getcwd(), '.env'),
type=click.Path(exists=True),
help="Location of the .env file, defaults to .env file in current working directory.")
@click.option('-q', '--quote', default='always',
type=click.Choice(['always', 'never', 'auto']),
help="Whether to quote or not the variable values. Default mode is always. This does not affect parsing.")
@click.version_option(version=__version__)
@click.pass_context
def cli(ctx, file, quote):
# type: (click.Context, Any, Any) -> None
'''This script is used to set, get or unset values from a .env file.'''
ctx.obj = {}
ctx.obj['FILE'] = file
ctx.obj['QUOTE'] = quote
@cli.command()
@click.pass_context
def list(ctx):
# type: (click.Context) -> None
'''Display all the stored key/value.'''
file = ctx.obj['FILE']
dotenv_as_dict = dotenv_values(file)
for k, v in dotenv_as_dict.items():
click.echo('%s=%s' % (k, v))
@cli.command()
@click.pass_context
@click.argument('key', required=True)
@click.argument('value', required=True)
def set(ctx, key, value):
# type: (click.Context, Any, Any) -> None
'''Store the given key/value.'''
file = ctx.obj['FILE']
quote = ctx.obj['QUOTE']
success, key, value = set_key(file, key, value, quote)
if success:
click.echo('%s=%s' % (key, value))
else:
exit(1)
@cli.command()
@click.pass_context
@click.argument('key', required=True)
def get(ctx, key):
# type: (click.Context, Any) -> None
'''Retrieve the value for the given key.'''
file = ctx.obj['FILE']
stored_value = get_key(file, key)
if stored_value:
click.echo('%s=%s' % (key, stored_value))
else:
exit(1)
@cli.command()
@click.pass_context
@click.argument('key', required=True)
def unset(ctx, key):
# type: (click.Context, Any) -> None
'''Removes the given key.'''
file = ctx.obj['FILE']
quote = ctx.obj['QUOTE']
success, key = unset_key(file, key, quote)
if success:
click.echo("Successfully removed %s" % key)
else:
exit(1)
@cli.command(context_settings={'ignore_unknown_options': True})
@click.pass_context
@click.argument('commandline', nargs=-1, type=click.UNPROCESSED)
def run(ctx, commandline):
# type: (click.Context, List[str]) -> None
"""Run command with environment variables present."""
file = ctx.obj['FILE']
dotenv_as_dict = dotenv_values(file)
if not commandline:
click.echo('No command given.')
exit(1)
ret = run_command(commandline, dotenv_as_dict) # type: ignore
exit(ret)
def run_command(command, env):
# type: (List[str], Dict[str, str]) -> int
"""Run command in sub process.
Runs the command in a sub process with the variables from `env`
added in the current environment variables.
Parameters
----------
command: List[str]
The command and it's parameters
env: Dict
The additional environment variables
Returns
-------
int
The return code of the command
"""
# copy the current environment variables and add the vales from
# `env`
cmd_env = os.environ.copy()
cmd_env.update(env)
p = Popen(command,
universal_newlines=True,
bufsize=0,
shell=False,
env=cmd_env)
_, _ = p.communicate()
return p.returncode
if __name__ == "__main__":
cli()

View File

@@ -0,0 +1,34 @@
import sys
from typing import Text # noqa
PY2 = sys.version_info[0] == 2 # type: bool
if PY2:
from StringIO import StringIO # noqa
else:
from io import StringIO # noqa
def to_env(text):
# type: (Text) -> str
"""
Encode a string the same way whether it comes from the environment or a `.env` file.
"""
if PY2:
return text.encode(sys.getfilesystemencoding() or "utf-8")
else:
return text
def to_text(string):
# type: (str) -> Text
"""
Make a string Unicode if it isn't already.
This is useful for defining raw unicode strings because `ur"foo"` isn't valid in
Python 3.
"""
if PY2:
return string.decode("utf-8")
else:
return string

View File

@@ -0,0 +1,41 @@
from __future__ import print_function
from IPython.core.magic import Magics, line_magic, magics_class # type: ignore
from IPython.core.magic_arguments import (argument, magic_arguments, # type: ignore
parse_argstring) # type: ignore
from .main import find_dotenv, load_dotenv
@magics_class
class IPythonDotEnv(Magics):
@magic_arguments()
@argument(
'-o', '--override', action='store_true',
help="Indicate to override existing variables"
)
@argument(
'-v', '--verbose', action='store_true',
help="Indicate function calls to be verbose"
)
@argument('dotenv_path', nargs='?', type=str, default='.env',
help='Search in increasingly higher folders for the `dotenv_path`')
@line_magic
def dotenv(self, line):
args = parse_argstring(self.dotenv, line)
# Locate the .env file
dotenv_path = args.dotenv_path
try:
dotenv_path = find_dotenv(dotenv_path, True, True)
except IOError:
print("cannot find .env file")
return
# Load the .env file
load_dotenv(dotenv_path, verbose=args.verbose, override=args.override)
def load_ipython_extension(ipython):
"""Register the %dotenv magic."""
ipython.register_magics(IPythonDotEnv)

View File

@@ -0,0 +1,283 @@
# -*- coding: utf-8 -*-
from __future__ import absolute_import, print_function, unicode_literals
import io
import os
import re
import shutil
import sys
import tempfile
from typing import (Dict, Iterator, List, Match, Optional, # noqa
Pattern, Union, TYPE_CHECKING, Text, IO, Tuple)
import warnings
from collections import OrderedDict
from contextlib import contextmanager
from .compat import StringIO, PY2, to_env
from .parser import parse_stream
if TYPE_CHECKING: # pragma: no cover
if sys.version_info >= (3, 6):
_PathLike = os.PathLike
else:
_PathLike = Text
if sys.version_info >= (3, 0):
_StringIO = StringIO
else:
_StringIO = StringIO[Text]
__posix_variable = re.compile(r'\$\{[^\}]*\}') # type: Pattern[Text]
class DotEnv():
def __init__(self, dotenv_path, verbose=False, encoding=None):
# type: (Union[Text, _PathLike, _StringIO], bool, Union[None, Text]) -> None
self.dotenv_path = dotenv_path # type: Union[Text,_PathLike, _StringIO]
self._dict = None # type: Optional[Dict[Text, Text]]
self.verbose = verbose # type: bool
self.encoding = encoding # type: Union[None, Text]
@contextmanager
def _get_stream(self):
# type: () -> Iterator[IO[Text]]
if isinstance(self.dotenv_path, StringIO):
yield self.dotenv_path
elif os.path.isfile(self.dotenv_path):
with io.open(self.dotenv_path, encoding=self.encoding) as stream:
yield stream
else:
if self.verbose:
warnings.warn("File doesn't exist {}".format(self.dotenv_path)) # type: ignore
yield StringIO('')
def dict(self):
# type: () -> Dict[Text, Text]
"""Return dotenv as dict"""
if self._dict:
return self._dict
values = OrderedDict(self.parse())
self._dict = resolve_nested_variables(values)
return self._dict
def parse(self):
# type: () -> Iterator[Tuple[Text, Text]]
with self._get_stream() as stream:
for mapping in parse_stream(stream):
if mapping.key is not None and mapping.value is not None:
yield mapping.key, mapping.value
def set_as_environment_variables(self, override=False):
# type: (bool) -> bool
"""
Load the current dotenv as system environemt variable.
"""
for k, v in self.dict().items():
if k in os.environ and not override:
continue
os.environ[to_env(k)] = to_env(v)
return True
def get(self, key):
# type: (Text) -> Optional[Text]
"""
"""
data = self.dict()
if key in data:
return data[key]
if self.verbose:
warnings.warn("key %s not found in %s." % (key, self.dotenv_path)) # type: ignore
return None
def get_key(dotenv_path, key_to_get):
# type: (Union[Text, _PathLike], Text) -> Optional[Text]
"""
Gets the value of a given key from the given .env
If the .env path given doesn't exist, fails
"""
return DotEnv(dotenv_path, verbose=True).get(key_to_get)
@contextmanager
def rewrite(path):
# type: (_PathLike) -> Iterator[Tuple[IO[Text], IO[Text]]]
try:
with tempfile.NamedTemporaryFile(mode="w+", delete=False) as dest:
with io.open(path) as source:
yield (source, dest) # type: ignore
except BaseException:
if os.path.isfile(dest.name):
os.unlink(dest.name)
raise
else:
shutil.move(dest.name, path)
def set_key(dotenv_path, key_to_set, value_to_set, quote_mode="always"):
# type: (_PathLike, Text, Text, Text) -> Tuple[Optional[bool], Text, Text]
"""
Adds or Updates a key/value to the given .env
If the .env path given doesn't exist, fails instead of risking creating
an orphan .env somewhere in the filesystem
"""
value_to_set = value_to_set.strip("'").strip('"')
if not os.path.exists(dotenv_path):
warnings.warn("can't write to %s - it doesn't exist." % dotenv_path) # type: ignore
return None, key_to_set, value_to_set
if " " in value_to_set:
quote_mode = "always"
line_template = '{}="{}"\n' if quote_mode == "always" else '{}={}\n'
line_out = line_template.format(key_to_set, value_to_set)
with rewrite(dotenv_path) as (source, dest):
replaced = False
for mapping in parse_stream(source):
if mapping.key == key_to_set:
dest.write(line_out)
replaced = True
else:
dest.write(mapping.original)
if not replaced:
dest.write(line_out)
return True, key_to_set, value_to_set
def unset_key(dotenv_path, key_to_unset, quote_mode="always"):
# type: (_PathLike, Text, Text) -> Tuple[Optional[bool], Text]
"""
Removes a given key from the given .env
If the .env path given doesn't exist, fails
If the given key doesn't exist in the .env, fails
"""
if not os.path.exists(dotenv_path):
warnings.warn("can't delete from %s - it doesn't exist." % dotenv_path) # type: ignore
return None, key_to_unset
removed = False
with rewrite(dotenv_path) as (source, dest):
for mapping in parse_stream(source):
if mapping.key == key_to_unset:
removed = True
else:
dest.write(mapping.original)
if not removed:
warnings.warn("key %s not removed from %s - key doesn't exist." % (key_to_unset, dotenv_path)) # type: ignore
return None, key_to_unset
return removed, key_to_unset
def resolve_nested_variables(values):
# type: (Dict[Text, Text]) -> Dict[Text, Text]
def _replacement(name):
# type: (Text) -> Text
"""
get appropriate value for a variable name.
first search in environ, if not found,
then look into the dotenv variables
"""
ret = os.getenv(name, new_values.get(name, ""))
return ret
def _re_sub_callback(match_object):
# type: (Match[Text]) -> Text
"""
From a match object gets the variable name and returns
the correct replacement
"""
return _replacement(match_object.group()[2:-1])
new_values = {}
for k, v in values.items():
new_values[k] = __posix_variable.sub(_re_sub_callback, v)
return new_values
def _walk_to_root(path):
# type: (Text) -> Iterator[Text]
"""
Yield directories starting from the given directory up to the root
"""
if not os.path.exists(path):
raise IOError('Starting path not found')
if os.path.isfile(path):
path = os.path.dirname(path)
last_dir = None
current_dir = os.path.abspath(path)
while last_dir != current_dir:
yield current_dir
parent_dir = os.path.abspath(os.path.join(current_dir, os.path.pardir))
last_dir, current_dir = current_dir, parent_dir
def find_dotenv(filename='.env', raise_error_if_not_found=False, usecwd=False):
# type: (Text, bool, bool) -> Text
"""
Search in increasingly higher folders for the given file
Returns path to the file if found, or an empty string otherwise
"""
def _is_interactive():
""" Decide whether this is running in a REPL or IPython notebook """
main = __import__('__main__', None, None, fromlist=['__file__'])
return not hasattr(main, '__file__')
if usecwd or _is_interactive():
# Should work without __file__, e.g. in REPL or IPython notebook.
path = os.getcwd()
else:
# will work for .py files
frame = sys._getframe()
# find first frame that is outside of this file
if PY2 and not __file__.endswith('.py'):
# in Python2 __file__ extension could be .pyc or .pyo (this doesn't account
# for edge case of Python compiled for non-standard extension)
current_file = __file__.rsplit('.', 1)[0] + '.py'
else:
current_file = __file__
while frame.f_code.co_filename == current_file:
frame = frame.f_back
frame_filename = frame.f_code.co_filename
path = os.path.dirname(os.path.abspath(frame_filename))
for dirname in _walk_to_root(path):
check_path = os.path.join(dirname, filename)
if os.path.isfile(check_path):
return check_path
if raise_error_if_not_found:
raise IOError('File not found')
return ''
def load_dotenv(dotenv_path=None, stream=None, verbose=False, override=False, **kwargs):
# type: (Union[Text, _PathLike, None], Optional[_StringIO], bool, bool, Union[None, Text]) -> bool
f = dotenv_path or stream or find_dotenv()
return DotEnv(f, verbose=verbose, **kwargs).set_as_environment_variables(override=override)
def dotenv_values(dotenv_path=None, stream=None, verbose=False, **kwargs):
# type: (Union[Text, _PathLike, None], Optional[_StringIO], bool, Union[None, Text]) -> Dict[Text, Text]
f = dotenv_path or stream or find_dotenv()
return DotEnv(f, verbose=verbose, **kwargs).dict()

View File

@@ -0,0 +1,147 @@
import codecs
import re
from typing import (IO, Iterator, Match, NamedTuple, Optional, Pattern, # noqa
Sequence, Text)
from .compat import to_text
def make_regex(string, extra_flags=0):
# type: (str, int) -> Pattern[Text]
return re.compile(to_text(string), re.UNICODE | extra_flags)
_whitespace = make_regex(r"\s*", extra_flags=re.MULTILINE)
_export = make_regex(r"(?:export[^\S\r\n]+)?")
_single_quoted_key = make_regex(r"'([^']+)'")
_unquoted_key = make_regex(r"([^=\#\s]+)")
_equal_sign = make_regex(r"[^\S\r\n]*=[^\S\r\n]*")
_single_quoted_value = make_regex(r"'((?:\\'|[^'])*)'")
_double_quoted_value = make_regex(r'"((?:\\"|[^"])*)"')
_unquoted_value_part = make_regex(r"([^ \r\n]*)")
_comment = make_regex(r"(?:\s*#[^\r\n]*)?")
_end_of_line = make_regex(r"[^\S\r\n]*(?:\r\n|\n|\r)?")
_rest_of_line = make_regex(r"[^\r\n]*(?:\r|\n|\r\n)?")
_double_quote_escapes = make_regex(r"\\[\\'\"abfnrtv]")
_single_quote_escapes = make_regex(r"\\[\\']")
Binding = NamedTuple("Binding", [("key", Optional[Text]),
("value", Optional[Text]),
("original", Text)])
class Error(Exception):
pass
class Reader:
def __init__(self, stream):
# type: (IO[Text]) -> None
self.string = stream.read()
self.position = 0
self.mark = 0
def has_next(self):
# type: () -> bool
return self.position < len(self.string)
def set_mark(self):
# type: () -> None
self.mark = self.position
def get_marked(self):
# type: () -> Text
return self.string[self.mark:self.position]
def peek(self, count):
# type: (int) -> Text
return self.string[self.position:self.position + count]
def read(self, count):
# type: (int) -> Text
result = self.string[self.position:self.position + count]
if len(result) < count:
raise Error("read: End of string")
self.position += count
return result
def read_regex(self, regex):
# type: (Pattern[Text]) -> Sequence[Text]
match = regex.match(self.string, self.position)
if match is None:
raise Error("read_regex: Pattern not found")
self.position = match.end()
return match.groups()
def decode_escapes(regex, string):
# type: (Pattern[Text], Text) -> Text
def decode_match(match):
# type: (Match[Text]) -> Text
return codecs.decode(match.group(0), 'unicode-escape') # type: ignore
return regex.sub(decode_match, string)
def parse_key(reader):
# type: (Reader) -> Text
char = reader.peek(1)
if char == "'":
(key,) = reader.read_regex(_single_quoted_key)
else:
(key,) = reader.read_regex(_unquoted_key)
return key
def parse_unquoted_value(reader):
# type: (Reader) -> Text
value = u""
while True:
(part,) = reader.read_regex(_unquoted_value_part)
value += part
after = reader.peek(2)
if len(after) < 2 or after[0] in u"\r\n" or after[1] in u" #\r\n":
return value
value += reader.read(2)
def parse_value(reader):
# type: (Reader) -> Text
char = reader.peek(1)
if char == u"'":
(value,) = reader.read_regex(_single_quoted_value)
return decode_escapes(_single_quote_escapes, value)
elif char == u'"':
(value,) = reader.read_regex(_double_quoted_value)
return decode_escapes(_double_quote_escapes, value)
elif char in (u"", u"\n", u"\r"):
return u""
else:
return parse_unquoted_value(reader)
def parse_binding(reader):
# type: (Reader) -> Binding
reader.set_mark()
try:
reader.read_regex(_whitespace)
reader.read_regex(_export)
key = parse_key(reader)
reader.read_regex(_equal_sign)
value = parse_value(reader)
reader.read_regex(_comment)
reader.read_regex(_end_of_line)
return Binding(key=key, value=value, original=reader.get_marked())
except Error:
reader.read_regex(_rest_of_line)
return Binding(key=None, value=None, original=reader.get_marked())
def parse_stream(stream):
# type:(IO[Text]) -> Iterator[Binding]
reader = Reader(stream)
while reader.has_next():
try:
yield parse_binding(reader)
except Error:
return

View File

@@ -0,0 +1 @@
# Marker file for PEP 561

View File

@@ -0,0 +1 @@
__version__ = "0.10.3"