Adjust Fargate Agent defaults and env var parsing

CHANGELOG.md

@@ -16,6 +16,10 @@ These changes are available in the [master branch](https://github.com/PrefectHQ/
 
 - None
 
+### Fixes
+
+- Fix Fargate Agent access defaults and environment variable support - [#1687](https://github.com/PrefectHQ/prefect/pull/1687)
+
 ### Deprecations
 
 - None

src/prefect/agent/fargate/agent.py

@@ -1,3 +1,4 @@
+from ast import literal_eval
 import os
 from typing import Iterable
 
@@ -28,15 +29,15 @@ class FargateAgent(Agent):
             Agents when polling for work
         - aws_access_key_id (str, optional): AWS access key id for connecting the boto3
             client. Defaults to the value set in the environment variable
-            `AWS_ACCESS_KEY_ID`.
+            `AWS_ACCESS_KEY_ID` or `None`
         - aws_secret_access_key (str, optional): AWS secret access key for connecting
             the boto3 client. Defaults to the value set in the environment variable
-            `AWS_SECRET_ACCESS_KEY`.
+            `AWS_SECRET_ACCESS_KEY` or `None`
         - aws_session_token (str, optional): AWS session key for connecting the boto3
             client. Defaults to the value set in the environment variable
-            `AWS_SESSION_TOKEN`.
+            `AWS_SESSION_TOKEN` or `None`
         - region_name (str, optional): AWS region name for connecting the boto3 client.
-            Defaults to the value set in the environment variable `REGION_NAME`.
+            Defaults to the value set in the environment variable `REGION_NAME` or `None`
         - **kwargs (dict, optional): additional keyword arguments to pass to boto3 for
             `register_task_definition` and `run_task`
     """
@@ -56,12 +57,12 @@ def __init__(  # type: ignore
         from boto3 import client as boto3_client
 
         # Config used for boto3 client initialization
-        aws_access_key_id = aws_access_key_id or os.getenv("AWS_ACCESS_KEY_ID", "")
+        aws_access_key_id = aws_access_key_id or os.getenv("AWS_ACCESS_KEY_ID")
         aws_secret_access_key = aws_secret_access_key or os.getenv(
-            "AWS_SECRET_ACCESS_KEY", ""
+            "AWS_SECRET_ACCESS_KEY"
         )
-        aws_session_token = aws_session_token or os.getenv("AWS_SESSION_TOKEN", "")
-        region_name = region_name or os.getenv("REGION_NAME", "")
+        aws_session_token = aws_session_token or os.getenv("AWS_SESSION_TOKEN")
+        region_name = region_name or os.getenv("REGION_NAME")
 
         # Parse accepted kwargs for definition and run
         self.task_definition_kwargs, self.task_run_kwargs = self._parse_kwargs(kwargs)
@@ -130,13 +131,25 @@ def _parse_kwargs(self, user_kwargs: dict) -> tuple:
         # Check environment if keys were not provided
         for key in definition_kwarg_list:
             if not task_definition_kwargs.get(key) and os.getenv(key):
-                task_definition_kwargs.update({key: os.getenv(key)})
                 self.logger.debug("{} from environment variable".format(key))
+                def_env_value = os.getenv(key)
+                try:
+                    # Parse env var if needed
+                    def_env_value = literal_eval(def_env_value)  # type: ignore
+                except ValueError:
+                    pass
+                task_definition_kwargs.update({key: def_env_value})
 
         for key in run_kwarg_list:
             if not task_run_kwargs.get(key) and os.getenv(key):
-                task_run_kwargs.update({key: os.getenv(key)})
                 self.logger.debug("{} from environment variable".format(key))
+                run_env_value = os.getenv(key)
+                try:
+                    # Parse env var if needed
+                    run_env_value = literal_eval(run_env_value)  # type: ignore
+                except ValueError:
+                    pass
+                task_run_kwargs.update({key: run_env_value})
 
         return task_definition_kwargs, task_run_kwargs
 

tests/agent/test_fargate_agent.py

@@ -35,10 +35,21 @@ def test_fargate_agent_config_options_default(monkeypatch, runner_token):
     assert agent.boto3_client
 
 
-def test_k8s_agent_config_options(monkeypatch, runner_token):
+def test_fargate_agent_config_options(monkeypatch, runner_token):
     boto3_client = MagicMock()
     monkeypatch.setattr("boto3.client", boto3_client)
 
+    # Client args
+    monkeypatch.setenv("AWS_ACCESS_KEY_ID", "")
+    monkeypatch.setenv("AWS_SECRET_ACCESS_KEY", "")
+    monkeypatch.setenv("AWS_SESSION_TOKEN", "")
+    monkeypatch.setenv("REGION_NAME", "")
+
+    monkeypatch.delenv("AWS_ACCESS_KEY_ID")
+    monkeypatch.delenv("AWS_SECRET_ACCESS_KEY")
+    monkeypatch.delenv("AWS_SESSION_TOKEN")
+    monkeypatch.delenv("REGION_NAME")
+
     with set_temporary_config({"cloud.agent.auth_token": "TEST_TOKEN"}):
         agent = FargateAgent(name="test", labels=["test"])
         assert agent
@@ -48,6 +59,14 @@ def test_k8s_agent_config_options(monkeypatch, runner_token):
         assert agent.logger
         assert agent.boto3_client
 
+        boto3_client.assert_called_with(
+            "ecs",
+            aws_access_key_id=None,
+            aws_secret_access_key=None,
+            aws_session_token=None,
+            region_name=None,
+        )
+
 
 def test_parse_task_definition_kwargs(monkeypatch, runner_token):
     boto3_client = MagicMock()
@@ -329,6 +348,45 @@ def test_fargate_agent_config_env_vars(monkeypatch, runner_token):
     )
 
 
+def test_fargate_agent_config_env_vars_lists_dicts(monkeypatch, runner_token):
+    boto3_client = MagicMock()
+    monkeypatch.setattr("boto3.client", boto3_client)
+
+    def_kwarg_dict = {
+        "placementConstraints": ["test"],
+        "proxyConfiguration": {"test": "test"},
+    }
+
+    run_kwarg_dict = {
+        "placementConstraints": ["test"],
+        "networkConfiguration": {"test": "test"},
+    }
+
+    # Client args
+    monkeypatch.setenv("AWS_ACCESS_KEY_ID", "id")
+    monkeypatch.setenv("AWS_SECRET_ACCESS_KEY", "secret")
+    monkeypatch.setenv("AWS_SESSION_TOKEN", "token")
+    monkeypatch.setenv("REGION_NAME", "region")
+
+    # Def / run args
+    monkeypatch.setenv("placementConstraints", "['test']")
+    monkeypatch.setenv("proxyConfiguration", "{'test': 'test'}")
+    monkeypatch.setenv("networkConfiguration", "{'test': 'test'}")
+
+    agent = FargateAgent(subnets=["subnet"])
+    assert agent
+    assert agent.task_definition_kwargs == def_kwarg_dict
+    assert agent.task_run_kwargs == run_kwarg_dict
+
+    boto3_client.assert_called_with(
+        "ecs",
+        aws_access_key_id="id",
+        aws_secret_access_key="secret",
+        aws_session_token="token",
+        region_name="region",
+    )
+
+
 def test_deploy_flows(monkeypatch, runner_token):
     boto3_client = MagicMock()