Загрузить файлы в «venv/Lib/site-packages/dotenv»
This commit is contained in:
51
venv/Lib/site-packages/dotenv/__init__.py
Normal file
51
venv/Lib/site-packages/dotenv/__init__.py
Normal file
@@ -0,0 +1,51 @@
|
|||||||
|
from typing import Any, Optional
|
||||||
|
|
||||||
|
from .main import dotenv_values, find_dotenv, get_key, load_dotenv, set_key, unset_key
|
||||||
|
|
||||||
|
|
||||||
|
def load_ipython_extension(ipython: Any) -> None:
|
||||||
|
from .ipython import load_ipython_extension
|
||||||
|
|
||||||
|
load_ipython_extension(ipython)
|
||||||
|
|
||||||
|
|
||||||
|
def get_cli_string(
|
||||||
|
path: Optional[str] = None,
|
||||||
|
action: Optional[str] = None,
|
||||||
|
key: Optional[str] = None,
|
||||||
|
value: Optional[str] = None,
|
||||||
|
quote: Optional[str] = None,
|
||||||
|
):
|
||||||
|
"""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(f"-q {quote}")
|
||||||
|
if path:
|
||||||
|
command.append(f"-f {path}")
|
||||||
|
if action:
|
||||||
|
command.append(action)
|
||||||
|
if key:
|
||||||
|
command.append(key)
|
||||||
|
if value:
|
||||||
|
if " " in value:
|
||||||
|
command.append(f'"{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",
|
||||||
|
]
|
||||||
6
venv/Lib/site-packages/dotenv/__main__.py
Normal file
6
venv/Lib/site-packages/dotenv/__main__.py
Normal file
@@ -0,0 +1,6 @@
|
|||||||
|
"""Entry point for cli, enables execution with `python -m dotenv`"""
|
||||||
|
|
||||||
|
from .cli import cli
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
cli()
|
||||||
236
venv/Lib/site-packages/dotenv/cli.py
Normal file
236
venv/Lib/site-packages/dotenv/cli.py
Normal file
@@ -0,0 +1,236 @@
|
|||||||
|
import json
|
||||||
|
import os
|
||||||
|
import shlex
|
||||||
|
import sys
|
||||||
|
from contextlib import contextmanager
|
||||||
|
from typing import IO, Any, Dict, Iterator, List, Optional
|
||||||
|
|
||||||
|
if sys.platform == "win32":
|
||||||
|
from subprocess import Popen
|
||||||
|
|
||||||
|
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, set_key, unset_key
|
||||||
|
from .version import __version__
|
||||||
|
|
||||||
|
|
||||||
|
def enumerate_env() -> Optional[str]:
|
||||||
|
"""
|
||||||
|
Return a path for the ${pwd}/.env file.
|
||||||
|
|
||||||
|
If pwd does not exist, return None.
|
||||||
|
"""
|
||||||
|
try:
|
||||||
|
cwd = os.getcwd()
|
||||||
|
except FileNotFoundError:
|
||||||
|
return None
|
||||||
|
path = os.path.join(cwd, ".env")
|
||||||
|
return path
|
||||||
|
|
||||||
|
|
||||||
|
@click.group()
|
||||||
|
@click.option(
|
||||||
|
"-f",
|
||||||
|
"--file",
|
||||||
|
default=enumerate_env(),
|
||||||
|
type=click.Path(file_okay=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.option(
|
||||||
|
"-e",
|
||||||
|
"--export",
|
||||||
|
default=False,
|
||||||
|
type=click.BOOL,
|
||||||
|
help="Whether to write the dot file as an executable bash script.",
|
||||||
|
)
|
||||||
|
@click.version_option(version=__version__)
|
||||||
|
@click.pass_context
|
||||||
|
def cli(ctx: click.Context, file: Any, quote: Any, export: Any) -> None:
|
||||||
|
"""This script is used to set, get or unset values from a .env file."""
|
||||||
|
ctx.obj = {"QUOTE": quote, "EXPORT": export, "FILE": file}
|
||||||
|
|
||||||
|
|
||||||
|
@contextmanager
|
||||||
|
def stream_file(path: os.PathLike) -> Iterator[IO[str]]:
|
||||||
|
"""
|
||||||
|
Open a file and yield the corresponding (decoded) stream.
|
||||||
|
|
||||||
|
Exits with error code 2 if the file cannot be opened.
|
||||||
|
"""
|
||||||
|
|
||||||
|
try:
|
||||||
|
with open(path) as stream:
|
||||||
|
yield stream
|
||||||
|
except OSError as exc:
|
||||||
|
print(f"Error opening env file: {exc}", file=sys.stderr)
|
||||||
|
sys.exit(2)
|
||||||
|
|
||||||
|
|
||||||
|
@cli.command(name="list")
|
||||||
|
@click.pass_context
|
||||||
|
@click.option(
|
||||||
|
"--format",
|
||||||
|
"output_format",
|
||||||
|
default="simple",
|
||||||
|
type=click.Choice(["simple", "json", "shell", "export"]),
|
||||||
|
help="The format in which to display the list. Default format is simple, "
|
||||||
|
"which displays name=value without quotes.",
|
||||||
|
)
|
||||||
|
def list_values(ctx: click.Context, output_format: str) -> None:
|
||||||
|
"""Display all the stored key/value."""
|
||||||
|
file = ctx.obj["FILE"]
|
||||||
|
|
||||||
|
with stream_file(file) as stream:
|
||||||
|
values = dotenv_values(stream=stream)
|
||||||
|
|
||||||
|
if output_format == "json":
|
||||||
|
click.echo(json.dumps(values, indent=2, sort_keys=True))
|
||||||
|
else:
|
||||||
|
prefix = "export " if output_format == "export" else ""
|
||||||
|
for k in sorted(values):
|
||||||
|
v = values[k]
|
||||||
|
if v is not None:
|
||||||
|
if output_format in ("export", "shell"):
|
||||||
|
v = shlex.quote(v)
|
||||||
|
click.echo(f"{prefix}{k}={v}")
|
||||||
|
|
||||||
|
|
||||||
|
@cli.command(name="set")
|
||||||
|
@click.pass_context
|
||||||
|
@click.argument("key", required=True)
|
||||||
|
@click.argument("value", required=True)
|
||||||
|
def set_value(ctx: click.Context, key: Any, value: Any) -> None:
|
||||||
|
"""
|
||||||
|
Store the given key/value.
|
||||||
|
|
||||||
|
This doesn't follow symlinks, to avoid accidentally modifying a file at a
|
||||||
|
potentially untrusted path.
|
||||||
|
"""
|
||||||
|
|
||||||
|
file = ctx.obj["FILE"]
|
||||||
|
quote = ctx.obj["QUOTE"]
|
||||||
|
export = ctx.obj["EXPORT"]
|
||||||
|
success, key, value = set_key(file, key, value, quote, export)
|
||||||
|
if success:
|
||||||
|
click.echo(f"{key}={value}")
|
||||||
|
else:
|
||||||
|
sys.exit(1)
|
||||||
|
|
||||||
|
|
||||||
|
@cli.command()
|
||||||
|
@click.pass_context
|
||||||
|
@click.argument("key", required=True)
|
||||||
|
def get(ctx: click.Context, key: Any) -> None:
|
||||||
|
"""Retrieve the value for the given key."""
|
||||||
|
file = ctx.obj["FILE"]
|
||||||
|
|
||||||
|
with stream_file(file) as stream:
|
||||||
|
values = dotenv_values(stream=stream)
|
||||||
|
|
||||||
|
stored_value = values.get(key)
|
||||||
|
if stored_value:
|
||||||
|
click.echo(stored_value)
|
||||||
|
else:
|
||||||
|
sys.exit(1)
|
||||||
|
|
||||||
|
|
||||||
|
@cli.command()
|
||||||
|
@click.pass_context
|
||||||
|
@click.argument("key", required=True)
|
||||||
|
def unset(ctx: click.Context, key: Any) -> None:
|
||||||
|
"""
|
||||||
|
Removes the given key.
|
||||||
|
|
||||||
|
This doesn't follow symlinks, to avoid accidentally modifying a file at a
|
||||||
|
potentially untrusted path.
|
||||||
|
"""
|
||||||
|
file = ctx.obj["FILE"]
|
||||||
|
quote = ctx.obj["QUOTE"]
|
||||||
|
success, key = unset_key(file, key, quote)
|
||||||
|
if success:
|
||||||
|
click.echo(f"Successfully removed {key}")
|
||||||
|
else:
|
||||||
|
sys.exit(1)
|
||||||
|
|
||||||
|
|
||||||
|
@cli.command(
|
||||||
|
context_settings={
|
||||||
|
"allow_extra_args": True,
|
||||||
|
"allow_interspersed_args": False,
|
||||||
|
"ignore_unknown_options": True,
|
||||||
|
}
|
||||||
|
)
|
||||||
|
@click.pass_context
|
||||||
|
@click.option(
|
||||||
|
"--override/--no-override",
|
||||||
|
default=True,
|
||||||
|
help="Override variables from the environment file with those from the .env file.",
|
||||||
|
)
|
||||||
|
@click.argument("commandline", nargs=-1, type=click.UNPROCESSED)
|
||||||
|
def run(ctx: click.Context, override: bool, commandline: tuple[str, ...]) -> None:
|
||||||
|
"""Run command with environment variables present."""
|
||||||
|
file = ctx.obj["FILE"]
|
||||||
|
if not os.path.isfile(file):
|
||||||
|
raise click.BadParameter(
|
||||||
|
f"Invalid value for '-f' \"{file}\" does not exist.", ctx=ctx
|
||||||
|
)
|
||||||
|
dotenv_as_dict = {
|
||||||
|
k: v
|
||||||
|
for (k, v) in dotenv_values(file).items()
|
||||||
|
if v is not None and (override or k not in os.environ)
|
||||||
|
}
|
||||||
|
|
||||||
|
if not commandline:
|
||||||
|
click.echo("No command given.")
|
||||||
|
sys.exit(1)
|
||||||
|
|
||||||
|
run_command([*commandline, *ctx.args], dotenv_as_dict)
|
||||||
|
|
||||||
|
|
||||||
|
def run_command(command: List[str], env: Dict[str, str]) -> None:
|
||||||
|
"""Replace the current process with the specified command.
|
||||||
|
|
||||||
|
Replaces the current process with the specified command and 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
|
||||||
|
-------
|
||||||
|
None
|
||||||
|
This function does not return any value. It replaces the current process with the new one.
|
||||||
|
|
||||||
|
"""
|
||||||
|
# copy the current environment variables and add the vales from
|
||||||
|
# `env`
|
||||||
|
cmd_env = os.environ.copy()
|
||||||
|
cmd_env.update(env)
|
||||||
|
|
||||||
|
if sys.platform == "win32":
|
||||||
|
# execvpe on Windows returns control immediately
|
||||||
|
# rather than once the command has finished.
|
||||||
|
p = Popen(command, universal_newlines=True, bufsize=0, shell=False, env=cmd_env)
|
||||||
|
_, _ = p.communicate()
|
||||||
|
|
||||||
|
sys.exit(p.returncode)
|
||||||
|
else:
|
||||||
|
os.execvpe(command[0], args=command, env=cmd_env)
|
||||||
50
venv/Lib/site-packages/dotenv/ipython.py
Normal file
50
venv/Lib/site-packages/dotenv/ipython.py
Normal file
@@ -0,0 +1,50 @@
|
|||||||
|
from IPython.core.magic import Magics, line_magic, magics_class # type: ignore
|
||||||
|
from IPython.core.magic_arguments import (
|
||||||
|
argument,
|
||||||
|
magic_arguments,
|
||||||
|
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)
|
||||||
480
venv/Lib/site-packages/dotenv/main.py
Normal file
480
venv/Lib/site-packages/dotenv/main.py
Normal file
@@ -0,0 +1,480 @@
|
|||||||
|
import io
|
||||||
|
import logging
|
||||||
|
import os
|
||||||
|
import pathlib
|
||||||
|
import stat
|
||||||
|
import sys
|
||||||
|
import tempfile
|
||||||
|
from collections import OrderedDict
|
||||||
|
from contextlib import contextmanager
|
||||||
|
from typing import IO, Dict, Iterable, Iterator, Mapping, Optional, Tuple, Union
|
||||||
|
|
||||||
|
from .parser import Binding, parse_stream
|
||||||
|
from .variables import parse_variables
|
||||||
|
|
||||||
|
# A type alias for a string path to be used for the paths in this file.
|
||||||
|
# These paths may flow to `open()` and `os.replace()`.
|
||||||
|
StrPath = Union[str, "os.PathLike[str]"]
|
||||||
|
|
||||||
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
|
def _load_dotenv_disabled() -> bool:
|
||||||
|
"""
|
||||||
|
Determine if dotenv loading has been disabled.
|
||||||
|
"""
|
||||||
|
if "PYTHON_DOTENV_DISABLED" not in os.environ:
|
||||||
|
return False
|
||||||
|
value = os.environ["PYTHON_DOTENV_DISABLED"].casefold()
|
||||||
|
return value in {"1", "true", "t", "yes", "y"}
|
||||||
|
|
||||||
|
|
||||||
|
def with_warn_for_invalid_lines(mappings: Iterator[Binding]) -> Iterator[Binding]:
|
||||||
|
for mapping in mappings:
|
||||||
|
if mapping.error:
|
||||||
|
logger.warning(
|
||||||
|
"python-dotenv could not parse statement starting at line %s",
|
||||||
|
mapping.original.line,
|
||||||
|
)
|
||||||
|
yield mapping
|
||||||
|
|
||||||
|
|
||||||
|
class DotEnv:
|
||||||
|
def __init__(
|
||||||
|
self,
|
||||||
|
dotenv_path: Optional[StrPath],
|
||||||
|
stream: Optional[IO[str]] = None,
|
||||||
|
verbose: bool = False,
|
||||||
|
encoding: Optional[str] = None,
|
||||||
|
interpolate: bool = True,
|
||||||
|
override: bool = True,
|
||||||
|
) -> None:
|
||||||
|
self.dotenv_path: Optional[StrPath] = dotenv_path
|
||||||
|
self.stream: Optional[IO[str]] = stream
|
||||||
|
self._dict: Optional[Dict[str, Optional[str]]] = None
|
||||||
|
self.verbose: bool = verbose
|
||||||
|
self.encoding: Optional[str] = encoding
|
||||||
|
self.interpolate: bool = interpolate
|
||||||
|
self.override: bool = override
|
||||||
|
|
||||||
|
@contextmanager
|
||||||
|
def _get_stream(self) -> Iterator[IO[str]]:
|
||||||
|
if self.dotenv_path and _is_file_or_fifo(self.dotenv_path):
|
||||||
|
with open(self.dotenv_path, encoding=self.encoding) as stream:
|
||||||
|
yield stream
|
||||||
|
elif self.stream is not None:
|
||||||
|
yield self.stream
|
||||||
|
else:
|
||||||
|
if self.verbose:
|
||||||
|
logger.info(
|
||||||
|
"python-dotenv could not find configuration file %s.",
|
||||||
|
self.dotenv_path or ".env",
|
||||||
|
)
|
||||||
|
yield io.StringIO("")
|
||||||
|
|
||||||
|
def dict(self) -> Dict[str, Optional[str]]:
|
||||||
|
"""Return dotenv as dict"""
|
||||||
|
if self._dict:
|
||||||
|
return self._dict
|
||||||
|
|
||||||
|
raw_values = self.parse()
|
||||||
|
|
||||||
|
if self.interpolate:
|
||||||
|
self._dict = OrderedDict(
|
||||||
|
resolve_variables(raw_values, override=self.override)
|
||||||
|
)
|
||||||
|
else:
|
||||||
|
self._dict = OrderedDict(raw_values)
|
||||||
|
|
||||||
|
return self._dict
|
||||||
|
|
||||||
|
def parse(self) -> Iterator[Tuple[str, Optional[str]]]:
|
||||||
|
with self._get_stream() as stream:
|
||||||
|
for mapping in with_warn_for_invalid_lines(parse_stream(stream)):
|
||||||
|
if mapping.key is not None:
|
||||||
|
yield mapping.key, mapping.value
|
||||||
|
|
||||||
|
def set_as_environment_variables(self) -> bool:
|
||||||
|
"""
|
||||||
|
Load the current dotenv as system environment variable.
|
||||||
|
"""
|
||||||
|
if not self.dict():
|
||||||
|
return False
|
||||||
|
|
||||||
|
for k, v in self.dict().items():
|
||||||
|
if k in os.environ and not self.override:
|
||||||
|
continue
|
||||||
|
if v is not None:
|
||||||
|
os.environ[k] = v
|
||||||
|
|
||||||
|
return True
|
||||||
|
|
||||||
|
def get(self, key: str) -> Optional[str]:
|
||||||
|
""" """
|
||||||
|
data = self.dict()
|
||||||
|
|
||||||
|
if key in data:
|
||||||
|
return data[key]
|
||||||
|
|
||||||
|
if self.verbose:
|
||||||
|
logger.warning("Key %s not found in %s.", key, self.dotenv_path)
|
||||||
|
|
||||||
|
return None
|
||||||
|
|
||||||
|
|
||||||
|
def get_key(
|
||||||
|
dotenv_path: StrPath,
|
||||||
|
key_to_get: str,
|
||||||
|
encoding: Optional[str] = "utf-8",
|
||||||
|
) -> Optional[str]:
|
||||||
|
"""
|
||||||
|
Get the value of a given key from the given .env.
|
||||||
|
|
||||||
|
Returns `None` if the key isn't found or doesn't have a value.
|
||||||
|
"""
|
||||||
|
return DotEnv(dotenv_path, verbose=True, encoding=encoding).get(key_to_get)
|
||||||
|
|
||||||
|
|
||||||
|
@contextmanager
|
||||||
|
def rewrite(
|
||||||
|
path: StrPath,
|
||||||
|
encoding: Optional[str],
|
||||||
|
follow_symlinks: bool = False,
|
||||||
|
) -> Iterator[Tuple[IO[str], IO[str]]]:
|
||||||
|
if follow_symlinks:
|
||||||
|
path = os.path.realpath(path)
|
||||||
|
|
||||||
|
try:
|
||||||
|
source: IO[str] = open(path, encoding=encoding)
|
||||||
|
try:
|
||||||
|
path_stat = os.lstat(path)
|
||||||
|
original_mode: Optional[int] = (
|
||||||
|
stat.S_IMODE(path_stat.st_mode)
|
||||||
|
if stat.S_ISREG(path_stat.st_mode)
|
||||||
|
else None
|
||||||
|
)
|
||||||
|
except BaseException:
|
||||||
|
source.close()
|
||||||
|
raise
|
||||||
|
except FileNotFoundError:
|
||||||
|
source = io.StringIO("")
|
||||||
|
original_mode = None
|
||||||
|
|
||||||
|
with tempfile.NamedTemporaryFile(
|
||||||
|
mode="w",
|
||||||
|
encoding=encoding,
|
||||||
|
delete=False,
|
||||||
|
prefix=".tmp_",
|
||||||
|
dir=os.path.dirname(os.path.abspath(path)),
|
||||||
|
) as dest:
|
||||||
|
dest_path = pathlib.Path(dest.name)
|
||||||
|
error = None
|
||||||
|
|
||||||
|
try:
|
||||||
|
with source:
|
||||||
|
yield (source, dest)
|
||||||
|
except BaseException as err:
|
||||||
|
error = err
|
||||||
|
|
||||||
|
if error is None:
|
||||||
|
try:
|
||||||
|
if original_mode is not None:
|
||||||
|
os.chmod(dest_path, original_mode)
|
||||||
|
|
||||||
|
os.replace(dest_path, path)
|
||||||
|
except BaseException:
|
||||||
|
dest_path.unlink(missing_ok=True)
|
||||||
|
raise
|
||||||
|
else:
|
||||||
|
dest_path.unlink(missing_ok=True)
|
||||||
|
raise error from None
|
||||||
|
|
||||||
|
|
||||||
|
def set_key(
|
||||||
|
dotenv_path: StrPath,
|
||||||
|
key_to_set: str,
|
||||||
|
value_to_set: str,
|
||||||
|
quote_mode: str = "always",
|
||||||
|
export: bool = False,
|
||||||
|
encoding: Optional[str] = "utf-8",
|
||||||
|
follow_symlinks: bool = False,
|
||||||
|
) -> Tuple[Optional[bool], str, str]:
|
||||||
|
"""
|
||||||
|
Adds or Updates a key/value to the given .env
|
||||||
|
|
||||||
|
The target .env file is created if it doesn't exist.
|
||||||
|
|
||||||
|
This function doesn't follow symlinks by default, to avoid accidentally
|
||||||
|
modifying a file at a potentially untrusted path. If you don't need this
|
||||||
|
protection and need symlinks to be followed, use `follow_symlinks`.
|
||||||
|
"""
|
||||||
|
if quote_mode not in ("always", "auto", "never"):
|
||||||
|
raise ValueError(f"Unknown quote_mode: {quote_mode}")
|
||||||
|
|
||||||
|
quote = quote_mode == "always" or (
|
||||||
|
quote_mode == "auto" and not value_to_set.isalnum()
|
||||||
|
)
|
||||||
|
|
||||||
|
if quote:
|
||||||
|
value_out = "'{}'".format(value_to_set.replace("'", "\\'"))
|
||||||
|
else:
|
||||||
|
value_out = value_to_set
|
||||||
|
if export:
|
||||||
|
line_out = f"export {key_to_set}={value_out}\n"
|
||||||
|
else:
|
||||||
|
line_out = f"{key_to_set}={value_out}\n"
|
||||||
|
|
||||||
|
with rewrite(dotenv_path, encoding=encoding, follow_symlinks=follow_symlinks) as (
|
||||||
|
source,
|
||||||
|
dest,
|
||||||
|
):
|
||||||
|
replaced = False
|
||||||
|
missing_newline = False
|
||||||
|
for mapping in with_warn_for_invalid_lines(parse_stream(source)):
|
||||||
|
if mapping.key == key_to_set:
|
||||||
|
dest.write(line_out)
|
||||||
|
replaced = True
|
||||||
|
else:
|
||||||
|
dest.write(mapping.original.string)
|
||||||
|
missing_newline = not mapping.original.string.endswith("\n")
|
||||||
|
if not replaced:
|
||||||
|
if missing_newline:
|
||||||
|
dest.write("\n")
|
||||||
|
dest.write(line_out)
|
||||||
|
|
||||||
|
return True, key_to_set, value_to_set
|
||||||
|
|
||||||
|
|
||||||
|
def unset_key(
|
||||||
|
dotenv_path: StrPath,
|
||||||
|
key_to_unset: str,
|
||||||
|
quote_mode: str = "always",
|
||||||
|
encoding: Optional[str] = "utf-8",
|
||||||
|
follow_symlinks: bool = False,
|
||||||
|
) -> Tuple[Optional[bool], str]:
|
||||||
|
"""
|
||||||
|
Removes a given key from the given `.env` file.
|
||||||
|
|
||||||
|
If the .env path given doesn't exist, fails.
|
||||||
|
If the given key doesn't exist in the .env, fails.
|
||||||
|
|
||||||
|
This function doesn't follow symlinks by default, to avoid accidentally
|
||||||
|
modifying a file at a potentially untrusted path. If you don't need this
|
||||||
|
protection and need symlinks to be followed, use `follow_symlinks`.
|
||||||
|
"""
|
||||||
|
if not os.path.exists(dotenv_path):
|
||||||
|
logger.warning("Can't delete from %s - it doesn't exist.", dotenv_path)
|
||||||
|
return None, key_to_unset
|
||||||
|
|
||||||
|
removed = False
|
||||||
|
with rewrite(dotenv_path, encoding=encoding, follow_symlinks=follow_symlinks) as (
|
||||||
|
source,
|
||||||
|
dest,
|
||||||
|
):
|
||||||
|
for mapping in with_warn_for_invalid_lines(parse_stream(source)):
|
||||||
|
if mapping.key == key_to_unset:
|
||||||
|
removed = True
|
||||||
|
else:
|
||||||
|
dest.write(mapping.original.string)
|
||||||
|
|
||||||
|
if not removed:
|
||||||
|
logger.warning(
|
||||||
|
"Key %s not removed from %s - key doesn't exist.", key_to_unset, dotenv_path
|
||||||
|
)
|
||||||
|
return None, key_to_unset
|
||||||
|
|
||||||
|
return removed, key_to_unset
|
||||||
|
|
||||||
|
|
||||||
|
def resolve_variables(
|
||||||
|
values: Iterable[Tuple[str, Optional[str]]],
|
||||||
|
override: bool,
|
||||||
|
) -> Mapping[str, Optional[str]]:
|
||||||
|
new_values: Dict[str, Optional[str]] = {}
|
||||||
|
|
||||||
|
for name, value in values:
|
||||||
|
if value is None:
|
||||||
|
result = None
|
||||||
|
else:
|
||||||
|
atoms = parse_variables(value)
|
||||||
|
env: Dict[str, Optional[str]] = {}
|
||||||
|
if override:
|
||||||
|
env.update(os.environ) # type: ignore
|
||||||
|
env.update(new_values)
|
||||||
|
else:
|
||||||
|
env.update(new_values)
|
||||||
|
env.update(os.environ) # type: ignore
|
||||||
|
result = "".join(atom.resolve(env) for atom in atoms)
|
||||||
|
|
||||||
|
new_values[name] = result
|
||||||
|
|
||||||
|
return new_values
|
||||||
|
|
||||||
|
|
||||||
|
def _walk_to_root(path: str) -> Iterator[str]:
|
||||||
|
"""
|
||||||
|
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: str = ".env",
|
||||||
|
raise_error_if_not_found: bool = False,
|
||||||
|
usecwd: bool = False,
|
||||||
|
) -> str:
|
||||||
|
"""
|
||||||
|
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"""
|
||||||
|
if hasattr(sys, "ps1") or hasattr(sys, "ps2"):
|
||||||
|
return True
|
||||||
|
try:
|
||||||
|
main = __import__("__main__", None, None, fromlist=["__file__"])
|
||||||
|
except ModuleNotFoundError:
|
||||||
|
return False
|
||||||
|
return not hasattr(main, "__file__")
|
||||||
|
|
||||||
|
def _is_debugger():
|
||||||
|
return sys.gettrace() is not None
|
||||||
|
|
||||||
|
if usecwd or _is_interactive() or _is_debugger() or getattr(sys, "frozen", False):
|
||||||
|
# Should work without __file__, e.g. in REPL or IPython notebook.
|
||||||
|
path = os.getcwd()
|
||||||
|
else:
|
||||||
|
# will work for .py files
|
||||||
|
frame = sys._getframe()
|
||||||
|
current_file = __file__
|
||||||
|
|
||||||
|
while frame.f_code.co_filename == current_file or not os.path.exists(
|
||||||
|
frame.f_code.co_filename
|
||||||
|
):
|
||||||
|
assert frame.f_back is not None
|
||||||
|
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 _is_file_or_fifo(check_path):
|
||||||
|
return check_path
|
||||||
|
|
||||||
|
if raise_error_if_not_found:
|
||||||
|
raise IOError("File not found")
|
||||||
|
|
||||||
|
return ""
|
||||||
|
|
||||||
|
|
||||||
|
def load_dotenv(
|
||||||
|
dotenv_path: Optional[StrPath] = None,
|
||||||
|
stream: Optional[IO[str]] = None,
|
||||||
|
verbose: bool = False,
|
||||||
|
override: bool = False,
|
||||||
|
interpolate: bool = True,
|
||||||
|
encoding: Optional[str] = "utf-8",
|
||||||
|
) -> bool:
|
||||||
|
"""Parse a .env file and then load all the variables found as environment variables.
|
||||||
|
|
||||||
|
Parameters:
|
||||||
|
dotenv_path: Absolute or relative path to .env file.
|
||||||
|
stream: Text stream (such as `io.StringIO`) with .env content, used if
|
||||||
|
`dotenv_path` is `None`.
|
||||||
|
verbose: Whether to output a warning the .env file is missing.
|
||||||
|
override: Whether to override the system environment variables with the variables
|
||||||
|
from the `.env` file.
|
||||||
|
encoding: Encoding to be used to read the file.
|
||||||
|
Returns:
|
||||||
|
Bool: True if at least one environment variable is set else False
|
||||||
|
|
||||||
|
If both `dotenv_path` and `stream` are `None`, `find_dotenv()` is used to find the
|
||||||
|
.env file with it's default parameters. If you need to change the default parameters
|
||||||
|
of `find_dotenv()`, you can explicitly call `find_dotenv()` and pass the result
|
||||||
|
to this function as `dotenv_path`.
|
||||||
|
|
||||||
|
If the environment variable `PYTHON_DOTENV_DISABLED` is set to a truthy value,
|
||||||
|
.env loading is disabled.
|
||||||
|
"""
|
||||||
|
if _load_dotenv_disabled():
|
||||||
|
logger.debug(
|
||||||
|
"python-dotenv: .env loading disabled by PYTHON_DOTENV_DISABLED environment variable"
|
||||||
|
)
|
||||||
|
return False
|
||||||
|
|
||||||
|
if dotenv_path is None and stream is None:
|
||||||
|
dotenv_path = find_dotenv()
|
||||||
|
|
||||||
|
dotenv = DotEnv(
|
||||||
|
dotenv_path=dotenv_path,
|
||||||
|
stream=stream,
|
||||||
|
verbose=verbose,
|
||||||
|
interpolate=interpolate,
|
||||||
|
override=override,
|
||||||
|
encoding=encoding,
|
||||||
|
)
|
||||||
|
return dotenv.set_as_environment_variables()
|
||||||
|
|
||||||
|
|
||||||
|
def dotenv_values(
|
||||||
|
dotenv_path: Optional[StrPath] = None,
|
||||||
|
stream: Optional[IO[str]] = None,
|
||||||
|
verbose: bool = False,
|
||||||
|
interpolate: bool = True,
|
||||||
|
encoding: Optional[str] = "utf-8",
|
||||||
|
) -> Dict[str, Optional[str]]:
|
||||||
|
"""
|
||||||
|
Parse a .env file and return its content as a dict.
|
||||||
|
|
||||||
|
The returned dict will have `None` values for keys without values in the .env file.
|
||||||
|
For example, `foo=bar` results in `{"foo": "bar"}` whereas `foo` alone results in
|
||||||
|
`{"foo": None}`
|
||||||
|
|
||||||
|
Parameters:
|
||||||
|
dotenv_path: Absolute or relative path to the .env file.
|
||||||
|
stream: `StringIO` object with .env content, used if `dotenv_path` is `None`.
|
||||||
|
verbose: Whether to output a warning if the .env file is missing.
|
||||||
|
encoding: Encoding to be used to read the file.
|
||||||
|
|
||||||
|
If both `dotenv_path` and `stream` are `None`, `find_dotenv()` is used to find the
|
||||||
|
.env file.
|
||||||
|
"""
|
||||||
|
if dotenv_path is None and stream is None:
|
||||||
|
dotenv_path = find_dotenv()
|
||||||
|
|
||||||
|
return DotEnv(
|
||||||
|
dotenv_path=dotenv_path,
|
||||||
|
stream=stream,
|
||||||
|
verbose=verbose,
|
||||||
|
interpolate=interpolate,
|
||||||
|
override=True,
|
||||||
|
encoding=encoding,
|
||||||
|
).dict()
|
||||||
|
|
||||||
|
|
||||||
|
def _is_file_or_fifo(path: StrPath) -> bool:
|
||||||
|
"""
|
||||||
|
Return True if `path` exists and is either a regular file or a FIFO.
|
||||||
|
"""
|
||||||
|
if os.path.isfile(path):
|
||||||
|
return True
|
||||||
|
|
||||||
|
try:
|
||||||
|
st = os.stat(path)
|
||||||
|
except (FileNotFoundError, OSError):
|
||||||
|
return False
|
||||||
|
|
||||||
|
return stat.S_ISFIFO(st.st_mode)
|
||||||
Reference in New Issue
Block a user