Upload 17 files

dotenv/__init__.py

@@ -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",
+]

dotenv/__main__.py

@@ -0,0 +1,6 @@
+"""Entry point for cli, enables execution with `python -m dotenv`"""
+
+from .cli import cli
+
+if __name__ == "__main__":
+    cli()

dotenv/cli.py

@@ -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)

dotenv/ipython.py

@@ -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)

dotenv/main.py

@@ -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)

dotenv/parser.py

@@ -0,0 +1,182 @@
+import codecs
+import re
+from typing import (
+    IO,
+    Iterator,
+    Match,
+    NamedTuple,
+    Optional,
+    Pattern,
+    Sequence,
+)
+
+
+def make_regex(string: str, extra_flags: int = 0) -> Pattern[str]:
+    return re.compile(string, re.UNICODE | extra_flags)
+
+
+_newline = make_regex(r"(\r\n|\n|\r)")
+_multiline_whitespace = make_regex(r"\s*", extra_flags=re.MULTILINE)
+_whitespace = make_regex(r"[^\S\r\n]*")
+_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]*)")
+_single_quoted_value = make_regex(r"'((?:\\'|[^'])*)'")
+_double_quoted_value = make_regex(r'"((?:\\"|[^"])*)"')
+_unquoted_value = make_regex(r"([^\r\n]*)")
+_comment = make_regex(r"(?:[^\S\r\n]*#[^\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"\\[\\']")
+
+
+class Original(NamedTuple):
+    string: str
+    line: int
+
+
+class Binding(NamedTuple):
+    key: Optional[str]
+    value: Optional[str]
+    original: Original
+    error: bool
+
+
+class Position:
+    def __init__(self, chars: int, line: int) -> None:
+        self.chars = chars
+        self.line = line
+
+    @classmethod
+    def start(cls) -> "Position":
+        return cls(chars=0, line=1)
+
+    def set(self, other: "Position") -> None:
+        self.chars = other.chars
+        self.line = other.line
+
+    def advance(self, string: str) -> None:
+        self.chars += len(string)
+        self.line += len(re.findall(_newline, string))
+
+
+class Error(Exception):
+    pass
+
+
+class Reader:
+    def __init__(self, stream: IO[str]) -> None:
+        self.string = stream.read()
+        self.position = Position.start()
+        self.mark = Position.start()
+
+    def has_next(self) -> bool:
+        return self.position.chars < len(self.string)
+
+    def set_mark(self) -> None:
+        self.mark.set(self.position)
+
+    def get_marked(self) -> Original:
+        return Original(
+            string=self.string[self.mark.chars : self.position.chars],
+            line=self.mark.line,
+        )
+
+    def peek(self, count: int) -> str:
+        return self.string[self.position.chars : self.position.chars + count]
+
+    def read(self, count: int) -> str:
+        result = self.string[self.position.chars : self.position.chars + count]
+        if len(result) < count:
+            raise Error("read: End of string")
+        self.position.advance(result)
+        return result
+
+    def read_regex(self, regex: Pattern[str]) -> Sequence[str]:
+        match = regex.match(self.string, self.position.chars)
+        if match is None:
+            raise Error("read_regex: Pattern not found")
+        self.position.advance(self.string[match.start() : match.end()])
+        return match.groups()
+
+
+def decode_escapes(regex: Pattern[str], string: str) -> str:
+    def decode_match(match: Match[str]) -> str:
+        return codecs.decode(match.group(0), "unicode-escape")  # type: ignore
+
+    return regex.sub(decode_match, string)
+
+
+def parse_key(reader: Reader) -> Optional[str]:
+    char = reader.peek(1)
+    if char == "#":
+        return None
+    elif char == "'":
+        (key,) = reader.read_regex(_single_quoted_key)
+    else:
+        (key,) = reader.read_regex(_unquoted_key)
+    return key
+
+
+def parse_unquoted_value(reader: Reader) -> str:
+    (part,) = reader.read_regex(_unquoted_value)
+    return re.sub(r"\s+#.*", "", part).rstrip()
+
+
+def parse_value(reader: Reader) -> str:
+    char = reader.peek(1)
+    if char == "'":
+        (value,) = reader.read_regex(_single_quoted_value)
+        return decode_escapes(_single_quote_escapes, value)
+    elif char == '"':
+        (value,) = reader.read_regex(_double_quoted_value)
+        return decode_escapes(_double_quote_escapes, value)
+    elif char in ("", "\n", "\r"):
+        return ""
+    else:
+        return parse_unquoted_value(reader)
+
+
+def parse_binding(reader: Reader) -> Binding:
+    reader.set_mark()
+    try:
+        reader.read_regex(_multiline_whitespace)
+        if not reader.has_next():
+            return Binding(
+                key=None,
+                value=None,
+                original=reader.get_marked(),
+                error=False,
+            )
+        reader.read_regex(_export)
+        key = parse_key(reader)
+        reader.read_regex(_whitespace)
+        if reader.peek(1) == "=":
+            reader.read_regex(_equal_sign)
+            value: Optional[str] = parse_value(reader)
+        else:
+            value = None
+        reader.read_regex(_comment)
+        reader.read_regex(_end_of_line)
+        return Binding(
+            key=key,
+            value=value,
+            original=reader.get_marked(),
+            error=False,
+        )
+    except Error:
+        reader.read_regex(_rest_of_line)
+        return Binding(
+            key=None,
+            value=None,
+            original=reader.get_marked(),
+            error=True,
+        )
+
+
+def parse_stream(stream: IO[str]) -> Iterator[Binding]:
+    reader = Reader(stream)
+    while reader.has_next():
+        yield parse_binding(reader)

dotenv/py.typed

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

dotenv/variables.py

@@ -0,0 +1,86 @@
+import re
+from abc import ABCMeta, abstractmethod
+from typing import Iterator, Mapping, Optional, Pattern
+
+_posix_variable: Pattern[str] = re.compile(
+    r"""
+    \$\{
+        (?P<name>[^\}:]*)
+        (?::-
+            (?P<default>[^\}]*)
+        )?
+    \}
+    """,
+    re.VERBOSE,
+)
+
+
+class Atom(metaclass=ABCMeta):
+    def __ne__(self, other: object) -> bool:
+        result = self.__eq__(other)
+        if result is NotImplemented:
+            return NotImplemented
+        return not result
+
+    @abstractmethod
+    def resolve(self, env: Mapping[str, Optional[str]]) -> str: ...
+
+
+class Literal(Atom):
+    def __init__(self, value: str) -> None:
+        self.value = value
+
+    def __repr__(self) -> str:
+        return f"Literal(value={self.value})"
+
+    def __eq__(self, other: object) -> bool:
+        if not isinstance(other, self.__class__):
+            return NotImplemented
+        return self.value == other.value
+
+    def __hash__(self) -> int:
+        return hash((self.__class__, self.value))
+
+    def resolve(self, env: Mapping[str, Optional[str]]) -> str:
+        return self.value
+
+
+class Variable(Atom):
+    def __init__(self, name: str, default: Optional[str]) -> None:
+        self.name = name
+        self.default = default
+
+    def __repr__(self) -> str:
+        return f"Variable(name={self.name}, default={self.default})"
+
+    def __eq__(self, other: object) -> bool:
+        if not isinstance(other, self.__class__):
+            return NotImplemented
+        return (self.name, self.default) == (other.name, other.default)
+
+    def __hash__(self) -> int:
+        return hash((self.__class__, self.name, self.default))
+
+    def resolve(self, env: Mapping[str, Optional[str]]) -> str:
+        default = self.default if self.default is not None else ""
+        result = env.get(self.name, default)
+        return result if result is not None else ""
+
+
+def parse_variables(value: str) -> Iterator[Atom]:
+    cursor = 0
+
+    for match in _posix_variable.finditer(value):
+        (start, end) = match.span()
+        name = match["name"]
+        default = match["default"]
+
+        if start > cursor:
+            yield Literal(value=value[cursor:start])
+
+        yield Variable(name=name, default=default)
+        cursor = end
+
+    length = len(value)
+    if cursor < length:
+        yield Literal(value=value[cursor:length])

dotenv/version.py

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