More env var cleanup (reflex-dev#4248)

* fix and test bug in config env loading

* streamline env var interpretation with @adhami3310

* improve error messages, fix invalid value for TELEMETRY_ENABLED

* just a small hint

* ruffing

* fix typo from review

* refactor - ruff broke the imports..

* cleanup imports

* more

* add internal and enum env var support

* ruff cleanup

* more global imports

* revert telemetry, it lives in rx.Config

* minor fixes/cleanup

* i missed some refs

* fix darglint

* reload config is internal

* fix EnvVar name

* add test for EnvVar + minor typing improvement

* bool tests

* was this broken?

* retain old behavior

* migrate APP_HARNESS_HEADLESS to new env var system

* migrate more APP_HARNESS env vars to new config system

* migrate SCREENSHOT_DIR to new env var system

* refactor EnvVar.get to be a method

* readd deleted functions and deprecate them

* improve EnvVar api, cleanup RELOAD_CONFIG question

* move is_prod_mode back to where it was

reflex/app.py

@@ -12,7 +12,6 @@
 import io
 import json
 import multiprocessing
-import os
 import platform
 import sys
 import traceback
@@ -96,7 +95,7 @@
     code_uses_state_contexts,
 )
 from reflex.utils import codespaces, console, exceptions, format, prerequisites, types
-from reflex.utils.exec import is_prod_mode, is_testing_env, should_skip_compile
+from reflex.utils.exec import is_prod_mode, is_testing_env
 from reflex.utils.imports import ImportVar
 
 if TYPE_CHECKING:
@@ -507,7 +506,7 @@ def add_page(
         # Check if the route given is valid
         verify_route_validity(route)
 
-        if route in self.unevaluated_pages and os.getenv(constants.RELOAD_CONFIG):
+        if route in self.unevaluated_pages and environment.RELOAD_CONFIG.is_set():
             # when the app is reloaded(typically for app harness tests), we should maintain
             # the latest render function of a route.This applies typically to decorated pages
             # since they are only added when app._compile is called.
@@ -724,7 +723,7 @@ def _should_compile(self) -> bool:
             Whether the app should be compiled.
         """
         # Check the environment variable.
-        if should_skip_compile():
+        if environment.REFLEX_SKIP_COMPILE.get():
             return False
 
         nocompile = prerequisites.get_web_dir() / constants.NOCOMPILE_FILE
@@ -947,7 +946,7 @@ def get_compilation_time() -> str:
         executor = None
         if (
             platform.system() in ("Linux", "Darwin")
-            and (number_of_processes := environment.REFLEX_COMPILE_PROCESSES)
+            and (number_of_processes := environment.REFLEX_COMPILE_PROCESSES.get())
             is not None
         ):
             executor = concurrent.futures.ProcessPoolExecutor(
@@ -956,7 +955,7 @@ def get_compilation_time() -> str:
             )
         else:
             executor = concurrent.futures.ThreadPoolExecutor(
-                max_workers=environment.REFLEX_COMPILE_THREADS
+                max_workers=environment.REFLEX_COMPILE_THREADS.get()
             )
 
         for route, component in zip(self.pages, page_components):

reflex/base.py

@@ -16,9 +16,6 @@
         from pydantic.fields import ModelField  # type: ignore
 
 
-from reflex import constants
-
-
 def validate_field_name(bases: List[Type["BaseModel"]], field_name: str) -> None:
     """Ensure that the field's name does not shadow an existing attribute of the model.
 
@@ -31,7 +28,8 @@ def validate_field_name(bases: List[Type["BaseModel"]], field_name: str) -> None
     """
     from reflex.utils.exceptions import VarNameError
 
-    reload = os.getenv(constants.RELOAD_CONFIG) == "True"
+    # can't use reflex.config.environment here cause of circular import
+    reload = os.getenv("__RELOAD_CONFIG", "").lower() == "true"
     for base in bases:
         try:
             if not reload and getattr(base, field_name, None):

reflex/compiler/compiler.py

@@ -527,7 +527,7 @@ def remove_tailwind_from_postcss() -> tuple[str, str]:
 
 def purge_web_pages_dir():
     """Empty out .web/pages directory."""
-    if not is_prod_mode() and environment.REFLEX_PERSIST_WEB_DIR:
+    if not is_prod_mode() and environment.REFLEX_PERSIST_WEB_DIR.get():
         # Skip purging the web directory in dev mode if REFLEX_PERSIST_WEB_DIR is set.
         return
 

reflex/components/core/upload.py

@@ -132,7 +132,7 @@ def get_upload_dir() -> Path:
     """
     Upload.is_used = True
 
-    uploaded_files_dir = environment.REFLEX_UPLOADED_FILES_DIR
+    uploaded_files_dir = environment.REFLEX_UPLOADED_FILES_DIR.get()
     uploaded_files_dir.mkdir(parents=True, exist_ok=True)
     return uploaded_files_dir
 

reflex/config.py

@@ -10,7 +10,17 @@
 import sys
 import urllib.parse
 from pathlib import Path
-from typing import Any, Dict, List, Optional, Set
+from typing import (
+    TYPE_CHECKING,
+    Any,
+    Dict,
+    Generic,
+    List,
+    Optional,
+    Set,
+    TypeVar,
+    get_args,
+)
 
 from typing_extensions import Annotated, get_type_hints
 
@@ -300,90 +310,240 @@ def interpret_env_var_value(
         )
 
 
+T = TypeVar("T")
+
+
+class EnvVar(Generic[T]):
+    """Environment variable."""
+
+    name: str
+    default: Any
+    type_: T
+
+    def __init__(self, name: str, default: Any, type_: T) -> None:
+        """Initialize the environment variable.
+
+        Args:
+            name: The environment variable name.
+            default: The default value.
+            type_: The type of the value.
+        """
+        self.name = name
+        self.default = default
+        self.type_ = type_
+
+    def interpret(self, value: str) -> T:
+        """Interpret the environment variable value.
+
+        Args:
+            value: The environment variable value.
+
+        Returns:
+            The interpreted value.
+        """
+        return interpret_env_var_value(value, self.type_, self.name)
+
+    def getenv(self) -> Optional[T]:
+        """Get the interpreted environment variable value.
+
+        Returns:
+            The environment variable value.
+        """
+        env_value = os.getenv(self.name, None)
+        if env_value is not None:
+            return self.interpret(env_value)
+        return None
+
+    def is_set(self) -> bool:
+        """Check if the environment variable is set.
+
+        Returns:
+            True if the environment variable is set.
+        """
+        return self.name in os.environ
+
+    def get(self) -> T:
+        """Get the interpreted environment variable value or the default value if not set.
+
+        Returns:
+            The interpreted value.
+        """
+        env_value = self.getenv()
+        if env_value is not None:
+            return env_value
+        return self.default
+
+    def set(self, value: T | None) -> None:
+        """Set the environment variable. None unsets the variable.
+
+        Args:
+            value: The value to set.
+        """
+        if value is None:
+            _ = os.environ.pop(self.name, None)
+        else:
+            if isinstance(value, enum.Enum):
+                value = value.value
+            os.environ[self.name] = str(value)
+
+
+class env_var:  # type: ignore
+    """Descriptor for environment variables."""
+
+    name: str
+    default: Any
+    internal: bool = False
+
+    def __init__(self, default: Any, internal: bool = False) -> None:
+        """Initialize the descriptor.
+
+        Args:
+            default: The default value.
+            internal: Whether the environment variable is reflex internal.
+        """
+        self.default = default
+        self.internal = internal
+
+    def __set_name__(self, owner, name):
+        """Set the name of the descriptor.
+
+        Args:
+            owner: The owner class.
+            name: The name of the descriptor.
+        """
+        self.name = name
+
+    def __get__(self, instance, owner):
+        """Get the EnvVar instance.
+
+        Args:
+            instance: The instance.
+            owner: The owner class.
+
+        Returns:
+            The EnvVar instance.
+        """
+        type_ = get_args(get_type_hints(owner)[self.name])[0]
+        env_name = self.name
+        if self.internal:
+            env_name = f"__{env_name}"
+        return EnvVar(name=env_name, default=self.default, type_=type_)
+
+
+if TYPE_CHECKING:
+
+    def env_var(default, internal=False) -> EnvVar:
+        """Typing helper for the env_var descriptor.
+
+        Args:
+            default: The default value.
+            internal: Whether the environment variable is reflex internal.
+
+        Returns:
+            The EnvVar instance.
+        """
+        return default
+
+
 class PathExistsFlag:
     """Flag to indicate that a path must exist."""
 
 
 ExistingPath = Annotated[Path, PathExistsFlag]
 
 
-@dataclasses.dataclass(init=False)
 class EnvironmentVariables:
     """Environment variables class to instantiate environment variables."""
 
     # Whether to use npm over bun to install frontend packages.
-    REFLEX_USE_NPM: bool = False
+    REFLEX_USE_NPM: EnvVar[bool] = env_var(False)
 
     # The npm registry to use.
-    NPM_CONFIG_REGISTRY: Optional[str] = None
+    NPM_CONFIG_REGISTRY: EnvVar[Optional[str]] = env_var(None)
 
     # Whether to use Granian for the backend. Otherwise, use Uvicorn.
-    REFLEX_USE_GRANIAN: bool = False
+    REFLEX_USE_GRANIAN: EnvVar[bool] = env_var(False)
 
     # The username to use for authentication on python package repository. Username and password must both be provided.
-    TWINE_USERNAME: Optional[str] = None
+    TWINE_USERNAME: EnvVar[Optional[str]] = env_var(None)
 
     # The password to use for authentication on python package repository. Username and password must both be provided.
-    TWINE_PASSWORD: Optional[str] = None
+    TWINE_PASSWORD: EnvVar[Optional[str]] = env_var(None)
 
     # Whether to use the system installed bun. If set to false, bun will be bundled with the app.
-    REFLEX_USE_SYSTEM_BUN: bool = False
+    REFLEX_USE_SYSTEM_BUN: EnvVar[bool] = env_var(False)
 
     # Whether to use the system installed node and npm. If set to false, node and npm will be bundled with the app.
-    REFLEX_USE_SYSTEM_NODE: bool = False
+    REFLEX_USE_SYSTEM_NODE: EnvVar[bool] = env_var(False)
 
     # The working directory for the next.js commands.
-    REFLEX_WEB_WORKDIR: Path = Path(constants.Dirs.WEB)
+    REFLEX_WEB_WORKDIR: EnvVar[Path] = env_var(Path(constants.Dirs.WEB))
 
     # Path to the alembic config file
-    ALEMBIC_CONFIG: ExistingPath = Path(constants.ALEMBIC_CONFIG)
+    ALEMBIC_CONFIG: EnvVar[ExistingPath] = env_var(Path(constants.ALEMBIC_CONFIG))
 
     # Disable SSL verification for HTTPX requests.
-    SSL_NO_VERIFY: bool = False
+    SSL_NO_VERIFY: EnvVar[bool] = env_var(False)
 
     # The directory to store uploaded files.
-    REFLEX_UPLOADED_FILES_DIR: Path = Path(constants.Dirs.UPLOADED_FILES)
+    REFLEX_UPLOADED_FILES_DIR: EnvVar[Path] = env_var(
+        Path(constants.Dirs.UPLOADED_FILES)
+    )
 
-    # Whether to use seperate processes to compile the frontend and how many. If not set, defaults to thread executor.
-    REFLEX_COMPILE_PROCESSES: Optional[int] = None
+    # Whether to use separate processes to compile the frontend and how many. If not set, defaults to thread executor.
+    REFLEX_COMPILE_PROCESSES: EnvVar[Optional[int]] = env_var(None)
 
-    # Whether to use seperate threads to compile the frontend and how many. Defaults to `min(32, os.cpu_count() + 4)`.
-    REFLEX_COMPILE_THREADS: Optional[int] = None
+    # Whether to use separate threads to compile the frontend and how many. Defaults to `min(32, os.cpu_count() + 4)`.
+    REFLEX_COMPILE_THREADS: EnvVar[Optional[int]] = env_var(None)
 
     # The directory to store reflex dependencies.
-    REFLEX_DIR: Path = Path(constants.Reflex.DIR)
+    REFLEX_DIR: EnvVar[Path] = env_var(Path(constants.Reflex.DIR))
 
     # Whether to print the SQL queries if the log level is INFO or lower.
-    SQLALCHEMY_ECHO: bool = False
+    SQLALCHEMY_ECHO: EnvVar[bool] = env_var(False)
 
     # Whether to ignore the redis config error. Some redis servers only allow out-of-band configuration.
-    REFLEX_IGNORE_REDIS_CONFIG_ERROR: bool = False
+    REFLEX_IGNORE_REDIS_CONFIG_ERROR: EnvVar[bool] = env_var(False)
 
     # Whether to skip purging the web directory in dev mode.
-    REFLEX_PERSIST_WEB_DIR: bool = False
+    REFLEX_PERSIST_WEB_DIR: EnvVar[bool] = env_var(False)
 
     # The reflex.build frontend host.
-    REFLEX_BUILD_FRONTEND: str = constants.Templates.REFLEX_BUILD_FRONTEND
+    REFLEX_BUILD_FRONTEND: EnvVar[str] = env_var(
+        constants.Templates.REFLEX_BUILD_FRONTEND
+    )
 
     # The reflex.build backend host.
-    REFLEX_BUILD_BACKEND: str = constants.Templates.REFLEX_BUILD_BACKEND
+    REFLEX_BUILD_BACKEND: EnvVar[str] = env_var(
+        constants.Templates.REFLEX_BUILD_BACKEND
+    )
 
-    def __init__(self):
-        """Initialize the environment variables."""
-        type_hints = get_type_hints(type(self))
+    # This env var stores the execution mode of the app
+    REFLEX_ENV_MODE: EnvVar[constants.Env] = env_var(constants.Env.DEV)
 
-        for field in dataclasses.fields(self):
-            raw_value = os.getenv(field.name, None)
+    # Whether to run the backend only. Exclusive with REFLEX_FRONTEND_ONLY.
+    REFLEX_BACKEND_ONLY: EnvVar[bool] = env_var(False)
 
-            field.type = type_hints.get(field.name) or field.type
+    # Whether to run the frontend only. Exclusive with REFLEX_BACKEND_ONLY.
+    REFLEX_FRONTEND_ONLY: EnvVar[bool] = env_var(False)
 
-            value = (
-                interpret_env_var_value(raw_value, field.type, field.name)
-                if raw_value is not None
-                else get_default_value_for_field(field)
-            )
+    # Reflex internal env to reload the config.
+    RELOAD_CONFIG: EnvVar[bool] = env_var(False, internal=True)
+
+    # If this env var is set to "yes", App.compile will be a no-op
+    REFLEX_SKIP_COMPILE: EnvVar[bool] = env_var(False, internal=True)
+
+    # Whether to run app harness tests in headless mode.
+    APP_HARNESS_HEADLESS: EnvVar[bool] = env_var(False)
+
+    # Which app harness driver to use.
+    APP_HARNESS_DRIVER: EnvVar[str] = env_var("Chrome")
+
+    # Arguments to pass to the app harness driver.
+    APP_HARNESS_DRIVER_ARGS: EnvVar[str] = env_var("")
 
-            setattr(self, field.name, value)
+    # Where to save screenshots when tests fail.
+    SCREENSHOT_DIR: EnvVar[Optional[Path]] = env_var(None)
 
 
 environment = EnvironmentVariables()