Support envfile task level and global option

- Includes original envfile parser that closely follows bash syntax

README.rst

@@ -34,7 +34,9 @@ Features
 
 ✅  Tasks can be defined as a sequence of other tasks
 
-✅  Can also be configured to execute tasks with any virtualenv (not just poetry)
+✅  Works with .env files
+
+✅  Can also be configured to execute tasks with any virtualenv or none (not just poetry)
 
 
 Installation
@@ -271,6 +273,20 @@ You can specify arbitrary environment variables to be set for a task by providin
 
 Notice this example uses deep keys which can be more convenient but aren't as well supported by some toml implementations.
 
+You can also specify an env file (with bashlike syntax) to load per task like so:
+
+  .. code-block:: bash
+
+    # .env
+    STAGE=dev
+    PASSWORD='!@#$%^&*('
+
+  .. code-block:: toml
+
+    [tool.poe.tasks]
+    serve.script = "myapp:run"
+    serve.envfile = ".env"
+
 Declaring CLI options (experimental)
 ------------------------------------
 
@@ -347,6 +363,19 @@ You can configure environment variables to be set for all poe tasks in the pypro
   VAR1 = "FOO"
   VAR2 = "BAR"
 
+You can also specify an env file (with bashlike syntax) to load for all tasks like so:
+
+  .. code-block:: bash
+
+    # .env
+    STAGE=dev
+    PASSWORD='!@#$%^&*('
+
+  .. code-block:: toml
+
+    [tool.poe]
+    envfile = ".env"
+
 Run poe from anywhere
 ---------------------
 

poethepoet/app.py

@@ -86,6 +86,7 @@ def run_task(self, context: Optional[RunContext] = None) -> Optional[int]:
         if context is None:
             context = RunContext(
                 config=self.config,
+                ui=self.ui,
                 env=os.environ,
                 dry=self.ui["dry_run"],
                 poe_active=os.environ.get("POE_ACTIVE"),
@@ -107,6 +108,7 @@ def run_task_graph(self, context: Optional[RunContext] = None) -> Optional[int]:
 
         context = RunContext(
             config=self.config,
+            ui=self.ui,
             env=os.environ,
             dry=self.ui["dry_run"],
             poe_active=os.environ.get("POE_ACTIVE"),

poethepoet/config.py

@@ -15,6 +15,7 @@ class PoeConfig:
         "default_array_task_type": str,
         "default_array_item_task_type": str,
         "env": dict,
+        "envfile": str,
         "executor": dict,
     }
 
@@ -51,6 +52,10 @@ def default_array_item_task_type(self) -> str:
     def global_env(self) -> Dict[str, str]:
         return self._poe.get("env", {})
 
+    @property
+    def global_envfile(self) -> Optional[str]:
+        return self._poe.get("envfile")
+
     @property
     def project(self) -> Any:
         return self._project

poethepoet/context.py

@@ -7,36 +7,44 @@
     Tuple,
     TYPE_CHECKING,
 )
+from .exceptions import ExecutionError
 from .executor import PoeExecutor
+from .envfile import load_env_file
 
 if TYPE_CHECKING:
     from .config import PoeConfig
+    from .ui import PoeUi
 
 
 class RunContext:
     config: "PoeConfig"
+    ui: "PoeUi"
     env: Dict[str, str]
     dry: bool
     poe_active: Optional[str]
     project_dir: Path
     multistage: bool = False
     exec_cache: Dict[str, Any]
     captured_stdout: Dict[Tuple[str, ...], str]
+    _envfile_cache: Dict[str, Dict[str, str]]
 
     def __init__(
         self,
         config: "PoeConfig",
+        ui: "PoeUi",
         env: MutableMapping[str, str],
         dry: bool,
         poe_active: Optional[str],
     ):
         self.config = config
+        self.ui = ui
         self.project_dir = Path(config.project_dir)
         self.env = {**env, "POE_ROOT": str(config.project_dir)}
         self.dry = dry
         self.poe_active = poe_active
         self.exec_cache = {}
         self.captured_stdout = {}
+        self._envfile_cache = {}
 
     @property
     def executor_type(self) -> Optional[str]:
@@ -60,3 +68,29 @@ def get_executor(
             executor_config=task_options.get("executor"),
             capture_stdout=task_options.get("capture_stdout", False),
         )
+
+    def get_env_file(self, envfile_path_str: str) -> Dict[str, str]:
+        if envfile_path_str in self._envfile_cache:
+            return self._envfile_cache[envfile_path_str]
+
+        result = {}
+
+        envfile_path = self.project_dir.joinpath(envfile_path_str)
+        if envfile_path.is_file():
+            try:
+                with envfile_path.open() as envfile:
+                    result = load_env_file(envfile)
+            except ValueError as error:
+                message = error.args[0]
+                raise ExecutionError(
+                    f"Syntax error in referenced envfile: {envfile_path_str!r}; {message}"
+                ) from error
+
+        else:
+            self.ui.print_msg(
+                f"Warning: Poe failed to locate envfile at {envfile_path_str!r}",
+                verbosity=1,
+            )
+
+        self._envfile_cache[envfile_path_str] = result
+        return result

poethepoet/envfile.py

@@ -0,0 +1,175 @@
+from enum import Enum
+import re
+from typing import Dict, List, Optional, TextIO
+
+
+class ParserException(ValueError):
+    def __init__(self, message: str, position: int):
+        super().__init__(message)
+        self.message = message
+        self.position = position
+
+
+class ParserState(Enum):
+    # Scanning for a new assignment
+    SCAN_VAR_NAME = 0
+    # In a value with no quoting
+    SCAN_VALUE = 1
+    # Inside single quotes
+    IN_SINGLE_QUOTE = 2
+    # Inside double quotes
+    IN_DOUBLE_QUOTE = 3
+
+
+def load_env_file(envfile: TextIO) -> Dict[str, str]:
+    """
+    Parses variable assignments from the given string. Expects a subset of bash syntax.
+    """
+
+    content_lines = envfile.readlines()
+    try:
+        return parse_env_file("".join(content_lines))
+    except ParserException as error:
+        line_num, position = _get_line_number(content_lines, error.position)
+        raise ValueError(f"{error.message} at line {line_num} position {position}.")
+
+
+VARNAME_PATTERN = r"^[\s\t;]*(?:export[\s\t]+)?([a-zA-Z_][a-zA-Z_0-9]*)"
+ASSIGNMENT_PATTERN = f"{VARNAME_PATTERN}="
+COMMENT_SUFFIX_PATTERN = r"^[\s\t;]*\#.*?\n"
+WHITESPACE_PATTERN = r"^[\s\t;]*"
+UNQUOTED_VALUE_PATTERN = r"^(.*?)(?:(\t|\s|;|'|\"|\\+))"
+SINGLE_QUOTE_VALUE_PATTERN = r"^((?:.|\n)*?)'"
+DOUBLE_QUOTE_VALUE_PATTERN = r"^((?:.|\n)*?)(\"|\\+)"
+
+
+def parse_env_file(content: str):
+    content = content + "\n"
+    result = {}
+    cursor = 0
+    state = ParserState.SCAN_VAR_NAME
+    var_name: Optional[str] = ""
+    var_content = []
+
+    while cursor < len(content):
+        if state == ParserState.SCAN_VAR_NAME:
+            # scan for new variable assignment
+            match = re.search(ASSIGNMENT_PATTERN, content[cursor:], re.MULTILINE)
+
+            if match is None:
+                comment_match = re.match(COMMENT_SUFFIX_PATTERN, content[cursor:])
+                if comment_match:
+                    cursor += comment_match.end()
+                    continue
+
+                if (
+                    re.match(WHITESPACE_PATTERN, content[cursor:], re.MULTILINE).end()  # type: ignore
+                    == len(content) - cursor
+                ):
+                    # The rest of the input is whitespace or semicolons
+                    break
+
+                # skip any immediate whitespace
+                cursor += re.match(  # type: ignore
+                    r"[\s\t\n]*", content[cursor:]
+                ).span()[1]
+
+                var_name_match = re.match(VARNAME_PATTERN, content[cursor:])
+                if var_name_match:
+                    cursor += var_name_match.span()[1]
+                    raise ParserException(
+                        f"Expected assignment operator", cursor,
+                    )
+
+                raise ParserException(f"Expected variable assignment", cursor)
+
+            var_name = match.group(1)
+            cursor += match.end()
+            state = ParserState.SCAN_VALUE
+
+        if state == ParserState.SCAN_VALUE:
+            # collect up until the first quote, whitespace, or group of backslashes
+            match = re.search(UNQUOTED_VALUE_PATTERN, content[cursor:], re.MULTILINE)
+            assert match
+            new_var_content, match_terminator = match.groups()
+            var_content.append(new_var_content)
+            cursor += len(new_var_content)
+
+            if match_terminator.isspace() or match_terminator == ";":
+                assert var_name
+                result[var_name] = "".join(var_content)
+                var_name = None
+                var_content = []
+                state = ParserState.SCAN_VAR_NAME
+                continue
+
+            if match_terminator == "'":
+                cursor += 1
+                state = ParserState.IN_SINGLE_QUOTE
+
+            elif match_terminator == '"':
+                cursor += 1
+                state = ParserState.IN_DOUBLE_QUOTE
+                continue
+
+            else:
+                # We found one or more backslashes
+                num_backslashes = len(match_terminator)
+                # Keep the excess (escaped) backslashes
+                var_content.append("\\" * (num_backslashes // 2))
+                cursor += num_backslashes
+
+                if num_backslashes % 2 > 0:
+                    # Odd number of backslashes, means the next char is escaped
+                    next_char = content[cursor]
+                    var_content.append(next_char)
+                    cursor += 1
+                    continue
+
+        if state == ParserState.IN_SINGLE_QUOTE:
+            # collect characters up until a single quote
+            match = re.search(
+                SINGLE_QUOTE_VALUE_PATTERN, content[cursor:], re.MULTILINE
+            )
+            if match is None:
+                raise ParserException(f"Unmatched single quote", cursor - 1)
+            var_content.append(match.group(1))
+            cursor += match.end()
+            state = ParserState.SCAN_VALUE
+            continue
+
+        if state == ParserState.IN_DOUBLE_QUOTE:
+            # collect characters up until a run of backslashes or double quote
+            match = re.search(
+                DOUBLE_QUOTE_VALUE_PATTERN, content[cursor:], re.MULTILINE
+            )
+            if match is None:
+                raise ParserException(f"Unmatched double quote", cursor - 1)
+            new_var_content, backslashes_or_dquote = match.groups()
+            var_content.append(new_var_content)
+            cursor += match.end()
+
+            if backslashes_or_dquote == '"':
+                state = ParserState.SCAN_VALUE
+                continue
+
+            # Keep the excess (escaped) backslashes
+            var_content.append("\\" * (len(backslashes_or_dquote) // 2))
+
+            if len(backslashes_or_dquote) % 2 == 0:
+                # whatever follows is escaped
+                next_char = content[cursor]
+                var_content.append(next_char)
+                cursor += 1
+
+    return result
+
+
+def _get_line_number(lines: List[str], position: int):
+    line_num = 1
+    for line in lines:
+        if len(line) > position:
+            break
+        line_num += 1
+        position -= len(line)
+    return line_num, position

poethepoet/helpers.py

@@ -2,4 +2,4 @@
 
 
 def is_valid_env_var(var_name: str) -> bool:
-    return bool(re.match("[a-zA-Z_][a-zA-Z0-9_]*", var_name))
+    return bool(re.match("^[a-zA-Z_][a-zA-Z0-9_]*$", var_name))

poethepoet/task/base.py

@@ -70,6 +70,7 @@ class PoeTask(metaclass=MetaPoeTask):
         "capture_stdout": (str),
         "deps": list,
         "env": dict,
+        "envfile": str,
         "executor": dict,
         "help": str,
         "uses": dict,
@@ -198,9 +199,23 @@ def run(
         """
         Run this task
         """
+        return self._handle_run(context, extra_args, self._build_env(env, context))
 
-        # Get env vars from glboal options
-        env = dict(env or {}, **self._config.global_env)
+    def _build_env(
+        self, env: Optional[MutableMapping[str, str]], context: "RunContext",
+    ):
+        env = dict(env or {})
+
+        # Get env vars from envfile referenced in global options
+        if self._config.global_envfile is not None:
+            env.update(context.get_env_file(self._config.global_envfile))
+
+        # Get env vars from global options
+        env.update(self._config.global_env)
+
+        # Get env vars from envfile referenced in task options
+        if self.options.get("envfile"):
+            env.update(context.get_env_file(self.options["envfile"]))
 
         # Get env vars from task options
         if self.options.get("env"):
@@ -209,7 +224,7 @@ def run(
         # Get env vars from dependencies
         env.update(self.get_dep_values(context))
 
-        return self._handle_run(context, extra_args, env)
+        return env
 
     def parse_named_args(self, extra_args: Sequence[str]) -> Optional[Dict[str, str]]:
         args_def = self.options.get("args")

tests/fixtures/envfile/credentials.env

@@ -0,0 +1,3 @@
+USER=admin
+PASSWORD=12345
+HOST=dev.example.com

tests/fixtures/envfile/prod.env

@@ -0,0 +1,2 @@
+HOST=prod.example.com
+PATH_SUFFIX=/app