Proposal for environment context

esque/cli/environment.py

@@ -1,9 +1,44 @@
 import os
+from typing import List, Optional, Union
 
 # Make sure to only ever import the whole module instead of single variables
 # That way it is easier to mock the variables during testing if necessary.
 # BAD: from esque.cli.environment import ESQUE_CONF_PATH
 # GOOD: from esque.cli import environment
 
-ESQUE_CONF_PATH = os.environ.get("ESQUE_CONF_PATH")
-ESQUE_VERBOSE = os.environ.get("ESQUE_VERBOSE")
+
+ESQUE_CONF_PATH: str
+ESQUE_VERBOSE: str
+ESQUE_CONTEXT_ENABLED: bool
+ESQUE_BOOTSTRAP_SERVERS: List
+ESQUE_SECURITY_PROTOCOL: str
+ESQUE_SCHEMA_REGISTRY: str
+ESQUE_NUM_PARTITIONS: int
+ESQUE_REPLICATION_FACTOR: int
+ESQUE_SASL_MECHANISM: str
+ESQUE_SASL_USER: str
+ESQUE_SASL_PASSWORD: str
+ESQUE_SSL_CAFILE: str
+ESQUE_SSL_CERTFILE: str
+ESQUE_SSL_KEYFILE: str
+ESQUE_SSL_PASSWORD: str
+
+
+def __getattr__(name: str) -> Optional[Union[str, bool, int, List[str]]]:
+    env = {k: v for k, v in globals()["__annotations__"].items() if k.startswith("ESQUE_")}
+
+    if name in env.keys():
+        val = os.getenv(name)
+
+        if val is None:
+            return None
+
+        if env[name] == bool:
+            return val == "True"
+
+        if env[name] == List:
+            return list(val.split(","))
+
+        return env[name](os.getenv(name))
+
+    raise AttributeError(f"Environment Variable {name} not defined.")

esque/config/__init__.py

@@ -10,7 +10,7 @@
 from pykafka import SslConfig
 
 import esque.validation
-from esque.cli import environment
+from esque.cli import environment as env
 from esque.config.migration import check_config_version
 from esque.errors import (
     ConfigException,
@@ -21,6 +21,7 @@
     UnsupportedSaslMechanism,
 )
 from esque.helpers import SingletonMeta
+from esque.validation import SCHEMA_DIR, validate
 
 if TYPE_CHECKING:
     try:
@@ -49,8 +50,8 @@ def config_path() -> Path:
 
 # create functions we can mock during tests
 def _config_path() -> Path:
-    if environment.ESQUE_CONF_PATH:
-        return Path(environment.ESQUE_CONF_PATH)
+    if env.ESQUE_CONF_PATH:
+        return Path(env.ESQUE_CONF_PATH)
     legacy_path = config_dir() / "esque.cfg"
     current_path = config_dir() / "esque_config.yaml"
     if legacy_path.exists() and not current_path.exists():
@@ -84,12 +85,52 @@ def available_contexts(self) -> List[str]:
     def default_values(self) -> Dict[str, Any]:
         return self.current_context_dict.get("default_values", {})
 
+    @property
+    def env_context(self) -> Optional[Dict]:
+        if not env.ESQUE_CONTEXT_ENABLED:
+            return None
+        context = {
+            "bootstrap_servers": env.ESQUE_BOOTSTRAP_SERVERS,
+            "security_protocol": env.ESQUE_SECURITY_PROTOCOL,
+            "schema_registry": env.ESQUE_SCHEMA_REGISTRY,
+            "default_values": {
+                "num_partitions": env.ESQUE_NUM_PARTITIONS,
+                "replication_factor": env.ESQUE_REPLICATION_FACTOR,
+            },
+        }
+
+        security_protocol = env.ESQUE_SECURITY_PROTOCOL
+        if security_protocol.startswith("SASL"):
+
+            context["sasl_params"] = {
+                "mechanism": env.ESQUE_SASL_MECHANISM,
+                "user": env.ESQUE_SASL_USER,
+                "password": env.ESQUE_SASL_PASSWORD,
+            }
+
+        if "SSL" in security_protocol:
+            context["ssl_params"] = (
+                {
+                    "cafile": env.ESQUE_SSL_CAFILE,
+                    "certfile": env.ESQUE_SSL_CERTFILE,
+                    "keyfile": env.ESQUE_SSL_KEYFILE,
+                    "password": env.ESQUE_SSL_PASSWORD,
+                },
+            )
+
+        validate(context, SCHEMA_DIR / "esque_context.yaml")
+        return context
+
     @property
     def current_context(self) -> str:
+        if env.ESQUE_CONTEXT_ENABLED:
+            return "env"
         return self._cfg["current_context"]
 
     @property
     def current_context_dict(self) -> Dict[str, Any]:
+        if env.ESQUE_CONTEXT_ENABLED:
+            return self.env_context
         return self._cfg["contexts"][self.current_context]
 
     @property

esque/validation/schemas/esque_context.yaml

@@ -0,0 +1,24 @@
+context:
+  bootstrap_servers: list(include('url'))
+  security_protocol: s_enum('PLAINTEXT','SASL_PLAINTEXT','SSL','SASL_SSL', required=False, case_sensitive=False)
+  schema_registry: include('url', required=False)
+  default_values: include('defaults', required=False)
+  ssl_params: include('ssl_params', required=False)
+  sasl_params: include('sasl_params', required=False)
+
+url: regex('^(?:[a-zA-Z]+://)?(?:[a-zA-Z][a-zA-Z0-9._-]+|[0-9]{,3}(?:\\.[0-9]{,3}){3})(?::[0-9]+)?$')
+
+defaults:
+  num_partitions: int(required=False)
+  replication_factor: int(required=False)
+
+sasl_params:
+  mechanism: s_enum('PLAIN','SCRAM-SHA-256','SCRAM-SHA-512', case_sensitive=False)
+  user: str()
+  password: str()
+
+ssl_params:
+  cafile: str(required=False)
+  certfile: str(required=False)
+  keyfile: str(required=False)
+  password: str(required=False)