More env var cleanup (#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 inspect
 import io
 import json
 import multiprocessing
-import os
 import platform
 import sys
 import traceback
@@ -96,7 +95,7 @@ from reflex.state import (
     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 @@ class App(MiddlewareMixin, LifespanMixin):
         # 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 @@ class App(MiddlewareMixin, LifespanMixin):
             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 @@ class App(MiddlewareMixin, LifespanMixin):
         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 @@ class App(MiddlewareMixin, LifespanMixin):
             )
         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 @@ except ModuleNotFoundError:
         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 os
 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,6 +310,141 @@ 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."""
 
@@ -307,83 +452,98 @@ class PathExistsFlag:
 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()

reflex/constants/__init__.py

@@ -2,18 +2,13 @@
 
 from .base import (
     COOKIES,
-    ENV_BACKEND_ONLY_ENV_VAR,
-    ENV_FRONTEND_ONLY_ENV_VAR,
-    ENV_MODE_ENV_VAR,
     IS_WINDOWS,
     LOCAL_STORAGE,
     POLLING_MAX_HTTP_BUFFER_SIZE,
     PYTEST_CURRENT_TEST,
     REFLEX_VAR_CLOSING_TAG,
     REFLEX_VAR_OPENING_TAG,
-    RELOAD_CONFIG,
     SESSION_STORAGE,
-    SKIP_COMPILE_ENV_VAR,
     ColorMode,
     Dirs,
     Env,
@@ -106,7 +101,6 @@ __ALL__ = [
     POLLING_MAX_HTTP_BUFFER_SIZE,
     PYTEST_CURRENT_TEST,
     Reflex,
-    RELOAD_CONFIG,
     RequirementsTxt,
     RouteArgType,
     RouteRegex,
@@ -116,7 +110,6 @@ __ALL__ = [
     ROUTER_DATA_INCLUDE,
     ROUTE_NOT_FOUND,
     SETTER_PREFIX,
-    SKIP_COMPILE_ENV_VAR,
     SocketEvent,
     StateManagerMode,
     Tailwind,

reflex/constants/base.py

@@ -112,7 +112,7 @@ class Templates(SimpleNamespace):
         from reflex.config import environment
 
         return (
-            environment.REFLEX_BUILD_FRONTEND
+            environment.REFLEX_BUILD_FRONTEND.get()
             + "/gen?reflex_init_token={reflex_init_token}"
         )
 
@@ -126,7 +126,7 @@ class Templates(SimpleNamespace):
         """
         from reflex.config import environment
 
-        return environment.REFLEX_BUILD_BACKEND + "/api/init/{reflex_init_token}"
+        return environment.REFLEX_BUILD_BACKEND.get() + "/api/init/{reflex_init_token}"
 
     @classproperty
     @classmethod
@@ -139,7 +139,8 @@ class Templates(SimpleNamespace):
         from reflex.config import environment
 
         return (
-            environment.REFLEX_BUILD_BACKEND + "/api/gen/{generation_hash}/refactored"
+            environment.REFLEX_BUILD_BACKEND.get()
+            + "/api/gen/{generation_hash}/refactored"
         )
 
     class Dirs(SimpleNamespace):
@@ -239,19 +240,9 @@ COOKIES = "cookies"
 LOCAL_STORAGE = "local_storage"
 SESSION_STORAGE = "session_storage"
 
-# If this env var is set to "yes", App.compile will be a no-op
-SKIP_COMPILE_ENV_VAR = "__REFLEX_SKIP_COMPILE"
-
-# This env var stores the execution mode of the app
-ENV_MODE_ENV_VAR = "REFLEX_ENV_MODE"
-
-ENV_BACKEND_ONLY_ENV_VAR = "REFLEX_BACKEND_ONLY"
-ENV_FRONTEND_ONLY_ENV_VAR = "REFLEX_FRONTEND_ONLY"
-
 # Testing variables.
 # Testing os env set by pytest when running a test case.
 PYTEST_CURRENT_TEST = "PYTEST_CURRENT_TEST"
-RELOAD_CONFIG = "__REFLEX_RELOAD_CONFIG"
 
 REFLEX_VAR_OPENING_TAG = "<reflex.Var>"
 REFLEX_VAR_CLOSING_TAG = "</reflex.Var>"

reflex/constants/installer.py

@@ -63,7 +63,7 @@ class Bun(SimpleNamespace):
         """
         from reflex.config import environment
 
-        return environment.REFLEX_DIR / "bun"
+        return environment.REFLEX_DIR.get() / "bun"
 
     @classproperty
     @classmethod
@@ -100,7 +100,7 @@ class Fnm(SimpleNamespace):
         """
         from reflex.config import environment
 
-        return environment.REFLEX_DIR / "fnm"
+        return environment.REFLEX_DIR.get() / "fnm"
 
     @classproperty
     @classmethod

reflex/custom_components/custom_components.py

@@ -609,14 +609,14 @@ def publish(
         help="The API token to use for authentication on python package repository. If token is provided, no username/password should be provided at the same time",
     ),
     username: Optional[str] = typer.Option(
-        environment.TWINE_USERNAME,
+        environment.TWINE_USERNAME.get(),
         "-u",
         "--username",
         show_default="TWINE_USERNAME environment variable value if set",
         help="The username to use for authentication on python package repository. Username and password must both be provided.",
     ),
     password: Optional[str] = typer.Option(
-        environment.TWINE_PASSWORD,
+        environment.TWINE_PASSWORD.get(),
         "-p",
         "--password",
         show_default="TWINE_PASSWORD environment variable value if set",

reflex/model.py

@@ -38,12 +38,12 @@ def get_engine(url: str | None = None) -> sqlalchemy.engine.Engine:
     url = url or conf.db_url
     if url is None:
         raise ValueError("No database url configured")
-    if not environment.ALEMBIC_CONFIG.exists():
+    if not environment.ALEMBIC_CONFIG.get().exists():
         console.warn(
             "Database is not initialized, run [bold]reflex db init[/bold] first."
         )
     # Print the SQL queries if the log level is INFO or lower.
-    echo_db_query = environment.SQLALCHEMY_ECHO
+    echo_db_query = environment.SQLALCHEMY_ECHO.get()
     # Needed for the admin dash on sqlite.
     connect_args = {"check_same_thread": False} if url.startswith("sqlite") else {}
     return sqlmodel.create_engine(url, echo=echo_db_query, connect_args=connect_args)
@@ -231,7 +231,7 @@ class Model(Base, sqlmodel.SQLModel):  # pyright: ignore [reportGeneralTypeIssue
         Returns:
             tuple of (config, script_directory)
         """
-        config = alembic.config.Config(environment.ALEMBIC_CONFIG)
+        config = alembic.config.Config(environment.ALEMBIC_CONFIG.get())
         return config, alembic.script.ScriptDirectory(
             config.get_main_option("script_location", default="version"),
         )
@@ -266,8 +266,8 @@ class Model(Base, sqlmodel.SQLModel):  # pyright: ignore [reportGeneralTypeIssue
     def alembic_init(cls):
         """Initialize alembic for the project."""
         alembic.command.init(
-            config=alembic.config.Config(environment.ALEMBIC_CONFIG),
-            directory=str(environment.ALEMBIC_CONFIG.parent / "alembic"),
+            config=alembic.config.Config(environment.ALEMBIC_CONFIG.get()),
+            directory=str(environment.ALEMBIC_CONFIG.get().parent / "alembic"),
         )
 
     @classmethod
@@ -287,7 +287,7 @@ class Model(Base, sqlmodel.SQLModel):  # pyright: ignore [reportGeneralTypeIssue
         Returns:
             True when changes have been detected.
         """
-        if not environment.ALEMBIC_CONFIG.exists():
+        if not environment.ALEMBIC_CONFIG.get().exists():
             return False
 
         config, script_directory = cls._alembic_config()
@@ -388,7 +388,7 @@ class Model(Base, sqlmodel.SQLModel):  # pyright: ignore [reportGeneralTypeIssue
             True - indicating the process was successful.
             None - indicating the process was skipped.
         """
-        if not environment.ALEMBIC_CONFIG.exists():
+        if not environment.ALEMBIC_CONFIG.get().exists():
             return
 
         with cls.get_db_engine().connect() as connection:

reflex/reflex.py

@@ -160,7 +160,7 @@ def _run(
     console.set_log_level(loglevel)
 
     # Set env mode in the environment
-    os.environ[constants.ENV_MODE_ENV_VAR] = env.value
+    environment.REFLEX_ENV_MODE.set(env)
 
     # Show system info
     exec.output_system_info()
@@ -277,13 +277,13 @@ def run(
         False,
         "--frontend-only",
         help="Execute only frontend.",
-        envvar=constants.ENV_FRONTEND_ONLY_ENV_VAR,
+        envvar=environment.REFLEX_FRONTEND_ONLY.name,
     ),
     backend: bool = typer.Option(
         False,
         "--backend-only",
         help="Execute only backend.",
-        envvar=constants.ENV_BACKEND_ONLY_ENV_VAR,
+        envvar=environment.REFLEX_BACKEND_ONLY.name,
     ),
     frontend_port: str = typer.Option(
         config.frontend_port, help="Specify a different frontend port."
@@ -302,8 +302,8 @@ def run(
     if frontend and backend:
         console.error("Cannot use both --frontend-only and --backend-only options.")
         raise typer.Exit(1)
-    os.environ[constants.ENV_BACKEND_ONLY_ENV_VAR] = str(backend).lower()
-    os.environ[constants.ENV_FRONTEND_ONLY_ENV_VAR] = str(frontend).lower()
+    environment.REFLEX_BACKEND_ONLY.set(backend)
+    environment.REFLEX_FRONTEND_ONLY.set(frontend)
 
     _run(env, frontend, backend, frontend_port, backend_port, backend_host, loglevel)
 
@@ -405,7 +405,7 @@ script_cli = typer.Typer()
 
 def _skip_compile():
     """Skip the compile step."""
-    os.environ[constants.SKIP_COMPILE_ENV_VAR] = "yes"
+    environment.REFLEX_SKIP_COMPILE.set(True)
 
 
 @db_cli.command(name="init")
@@ -420,7 +420,7 @@ def db_init():
         return
 
     # Check the alembic config.
-    if environment.ALEMBIC_CONFIG.exists():
+    if environment.ALEMBIC_CONFIG.get().exists():
         console.error(
             "Database is already initialized. Use "
             "[bold]reflex db makemigrations[/bold] to create schema change "

reflex/state.py

@@ -3377,7 +3377,7 @@ class StateManagerRedis(StateManager):
             )
         except ResponseError:
             # Some redis servers only allow out-of-band configuration, so ignore errors here.
-            if not environment.REFLEX_IGNORE_REDIS_CONFIG_ERROR:
+            if not environment.REFLEX_IGNORE_REDIS_CONFIG_ERROR.get():
                 raise
         async with self.redis.pubsub() as pubsub:
             await pubsub.psubscribe(lock_key_channel)

reflex/testing.py

@@ -43,6 +43,7 @@ import reflex.utils.exec
 import reflex.utils.format
 import reflex.utils.prerequisites
 import reflex.utils.processes
+from reflex.config import environment
 from reflex.state import (
     BaseState,
     StateManager,
@@ -250,6 +251,7 @@ class AppHarness:
 
     def _initialize_app(self):
         # disable telemetry reporting for tests
+
         os.environ["TELEMETRY_ENABLED"] = "false"
         self.app_path.mkdir(parents=True, exist_ok=True)
         if self.app_source is not None:
@@ -615,10 +617,10 @@ class AppHarness:
         if self.frontend_url is None:
             raise RuntimeError("Frontend is not running.")
         want_headless = False
-        if os.environ.get("APP_HARNESS_HEADLESS"):
+        if environment.APP_HARNESS_HEADLESS.get():
             want_headless = True
         if driver_clz is None:
-            requested_driver = os.environ.get("APP_HARNESS_DRIVER", "Chrome")
+            requested_driver = environment.APP_HARNESS_DRIVER.get()
             driver_clz = getattr(webdriver, requested_driver)
             if driver_options is None:
                 driver_options = getattr(webdriver, f"{requested_driver}Options")()
@@ -640,7 +642,7 @@ class AppHarness:
                 driver_options.add_argument("headless")
         if driver_options is None:
             raise RuntimeError(f"Could not determine options for {driver_clz}")
-        if args := os.environ.get("APP_HARNESS_DRIVER_ARGS"):
+        if args := environment.APP_HARNESS_DRIVER_ARGS.get():
             for arg in args.split(","):
                 driver_options.add_argument(arg)
         if driver_option_args is not None:
@@ -944,7 +946,7 @@ class AppHarnessProd(AppHarness):
     def _start_backend(self):
         if self.app_instance is None:
             raise RuntimeError("App was not initialized.")
-        os.environ[reflex.constants.SKIP_COMPILE_ENV_VAR] = "yes"
+        environment.REFLEX_SKIP_COMPILE.set(True)
         self.backend = uvicorn.Server(
             uvicorn.Config(
                 app=self.app_instance,
@@ -961,7 +963,7 @@ class AppHarnessProd(AppHarness):
         try:
             return super()._poll_for_servers(timeout)
         finally:
-            os.environ.pop(reflex.constants.SKIP_COMPILE_ENV_VAR, None)
+            environment.REFLEX_SKIP_COMPILE.set(None)
 
     def stop(self):
         """Stop the frontend python webserver."""

reflex/utils/exec.py

@@ -184,7 +184,7 @@ def should_use_granian():
     Returns:
         True if Granian should be used.
     """
-    return environment.REFLEX_USE_GRANIAN
+    return environment.REFLEX_USE_GRANIAN.get()
 
 
 def get_app_module():
@@ -369,7 +369,9 @@ def run_uvicorn_backend_prod(host, port, loglevel):
         command,
         run=True,
         show_logs=True,
-        env={constants.SKIP_COMPILE_ENV_VAR: "yes"},  # skip compile for prod backend
+        env={
+            environment.REFLEX_SKIP_COMPILE.name: "true"
+        },  # skip compile for prod backend
     )
 
 
@@ -405,7 +407,7 @@ def run_granian_backend_prod(host, port, loglevel):
             run=True,
             show_logs=True,
             env={
-                constants.SKIP_COMPILE_ENV_VAR: "yes"
+                environment.REFLEX_SKIP_COMPILE.name: "true"
             },  # skip compile for prod backend
         )
     except ImportError:
@@ -491,11 +493,8 @@ def is_prod_mode() -> bool:
     Returns:
         True if the app is running in production mode or False if running in dev mode.
     """
-    current_mode = os.environ.get(
-        constants.ENV_MODE_ENV_VAR,
-        constants.Env.DEV.value,
-    )
-    return current_mode == constants.Env.PROD.value
+    current_mode = environment.REFLEX_ENV_MODE.get()
+    return current_mode == constants.Env.PROD
 
 
 def is_frontend_only() -> bool:
@@ -504,7 +503,13 @@ def is_frontend_only() -> bool:
     Returns:
         True if the app is running in frontend-only mode.
     """
-    return os.environ.get(constants.ENV_FRONTEND_ONLY_ENV_VAR, "").lower() == "true"
+    console.deprecate(
+        "is_frontend_only() is deprecated and will be removed in a future release.",
+        reason="Use `environment.REFLEX_FRONTEND_ONLY.get()` instead.",
+        deprecation_version="0.6.5",
+        removal_version="0.7.0",
+    )
+    return environment.REFLEX_FRONTEND_ONLY.get()
 
 
 def is_backend_only() -> bool:
@@ -513,7 +518,13 @@ def is_backend_only() -> bool:
     Returns:
         True if the app is running in backend-only mode.
     """
-    return os.environ.get(constants.ENV_BACKEND_ONLY_ENV_VAR, "").lower() == "true"
+    console.deprecate(
+        "is_backend_only() is deprecated and will be removed in a future release.",
+        reason="Use `environment.REFLEX_BACKEND_ONLY.get()` instead.",
+        deprecation_version="0.6.5",
+        removal_version="0.7.0",
+    )
+    return environment.REFLEX_BACKEND_ONLY.get()
 
 
 def should_skip_compile() -> bool:
@@ -522,4 +533,10 @@ def should_skip_compile() -> bool:
     Returns:
         True if the app should skip compile.
     """
-    return os.environ.get(constants.SKIP_COMPILE_ENV_VAR) == "yes"
+    console.deprecate(
+        "should_skip_compile() is deprecated and will be removed in a future release.",
+        reason="Use `environment.REFLEX_SKIP_COMPILE.get()` instead.",
+        deprecation_version="0.6.5",
+        removal_version="0.7.0",
+    )
+    return environment.REFLEX_SKIP_COMPILE.get()

reflex/utils/net.py

@@ -12,7 +12,7 @@ def _httpx_verify_kwarg() -> bool:
     Returns:
         True if SSL verification is enabled, False otherwise
     """
-    return not environment.SSL_NO_VERIFY
+    return not environment.SSL_NO_VERIFY.get()
 
 
 def get(url: str, **kwargs) -> httpx.Response:

reflex/utils/path_ops.py

@@ -136,7 +136,7 @@ def use_system_node() -> bool:
     Returns:
         Whether the system node should be used.
     """
-    return environment.REFLEX_USE_SYSTEM_NODE
+    return environment.REFLEX_USE_SYSTEM_NODE.get()
 
 
 def use_system_bun() -> bool:
@@ -145,7 +145,7 @@ def use_system_bun() -> bool:
     Returns:
         Whether the system bun should be used.
     """
-    return environment.REFLEX_USE_SYSTEM_BUN
+    return environment.REFLEX_USE_SYSTEM_BUN.get()
 
 
 def get_node_bin_path() -> Path | None:

reflex/utils/prerequisites.py

@@ -69,7 +69,7 @@ def get_web_dir() -> Path:
     Returns:
         The working directory.
     """
-    return environment.REFLEX_WEB_WORKDIR
+    return environment.REFLEX_WEB_WORKDIR.get()
 
 
 def _python_version_check():
@@ -260,7 +260,7 @@ def windows_npm_escape_hatch() -> bool:
     Returns:
         If the user has set REFLEX_USE_NPM.
     """
-    return environment.REFLEX_USE_NPM
+    return environment.REFLEX_USE_NPM.get()
 
 
 def get_app(reload: bool = False) -> ModuleType:
@@ -278,7 +278,7 @@ def get_app(reload: bool = False) -> ModuleType:
     from reflex.utils import telemetry
 
     try:
-        os.environ[constants.RELOAD_CONFIG] = str(reload)
+        environment.RELOAD_CONFIG.set(reload)
         config = get_config()
         if not config.app_name:
             raise RuntimeError(
@@ -1019,7 +1019,7 @@ def needs_reinit(frontend: bool = True) -> bool:
         return False
 
     # Make sure the .reflex directory exists.
-    if not environment.REFLEX_DIR.exists():
+    if not environment.REFLEX_DIR.get().exists():
         return True
 
     # Make sure the .web directory exists in frontend mode.
@@ -1124,7 +1124,7 @@ def ensure_reflex_installation_id() -> Optional[int]:
     """
     try:
         initialize_reflex_user_directory()
-        installation_id_file = environment.REFLEX_DIR / "installation_id"
+        installation_id_file = environment.REFLEX_DIR.get() / "installation_id"
 
         installation_id = None
         if installation_id_file.exists():
@@ -1149,7 +1149,7 @@ def ensure_reflex_installation_id() -> Optional[int]:
 def initialize_reflex_user_directory():
     """Initialize the reflex user directory."""
     # Create the reflex directory.
-    path_ops.mkdir(environment.REFLEX_DIR)
+    path_ops.mkdir(environment.REFLEX_DIR.get())
 
 
 def initialize_frontend_dependencies():
@@ -1172,7 +1172,10 @@ def check_db_initialized() -> bool:
     Returns:
         True if alembic is initialized (or if database is not used).
     """
-    if get_config().db_url is not None and not environment.ALEMBIC_CONFIG.exists():
+    if (
+        get_config().db_url is not None
+        and not environment.ALEMBIC_CONFIG.get().exists()
+    ):
         console.error(
             "Database is not initialized. Run [bold]reflex db init[/bold] first."
         )
@@ -1182,7 +1185,7 @@ def check_db_initialized() -> bool:
 
 def check_schema_up_to_date():
     """Check if the sqlmodel metadata matches the current database schema."""
-    if get_config().db_url is None or not environment.ALEMBIC_CONFIG.exists():
+    if get_config().db_url is None or not environment.ALEMBIC_CONFIG.get().exists():
         return
     with model.Model.get_db_engine().connect() as connection:
         try:

reflex/utils/registry.py

@@ -55,4 +55,4 @@ def _get_npm_registry() -> str:
     Returns:
         str:
     """
-    return environment.NPM_CONFIG_REGISTRY or get_best_registry()
+    return environment.NPM_CONFIG_REGISTRY.get() or get_best_registry()

reflex/utils/telemetry.py

@@ -8,6 +8,8 @@ import multiprocessing
 import platform
 import warnings
 
+from reflex.config import environment
+
 try:
     from datetime import UTC, datetime
 except ImportError:
@@ -20,7 +22,6 @@ import psutil
 
 from reflex import constants
 from reflex.utils import console
-from reflex.utils.exec import should_skip_compile
 from reflex.utils.prerequisites import ensure_reflex_installation_id, get_project_hash
 
 POSTHOG_API_URL: str = "https://app.posthog.com/capture/"
@@ -94,7 +95,7 @@ def _raise_on_missing_project_hash() -> bool:
         False when compilation should be skipped (i.e. no .web directory is required).
         Otherwise return True.
     """
-    return not should_skip_compile()
+    return not environment.REFLEX_SKIP_COMPILE.get()
 
 
 def _prepare_event(event: str, **kwargs) -> dict:

tests/integration/conftest.py

@@ -6,6 +6,7 @@ from pathlib import Path
 
 import pytest
 
+from reflex.config import environment
 from reflex.testing import AppHarness, AppHarnessProd
 
 DISPLAY = None
@@ -21,7 +22,7 @@ def xvfb():
     Yields:
         the pyvirtualdisplay object that the browser will be open on
     """
-    if os.environ.get("GITHUB_ACTIONS") and not os.environ.get("APP_HARNESS_HEADLESS"):
+    if os.environ.get("GITHUB_ACTIONS") and not environment.APP_HARNESS_HEADLESS.get():
         from pyvirtualdisplay.smartdisplay import (  # pyright: ignore [reportMissingImports]
             SmartDisplay,
         )
@@ -42,7 +43,7 @@ def pytest_exception_interact(node, call, report):
         call: The pytest call describing when/where the test was invoked.
         report: The pytest log report object.
     """
-    screenshot_dir = os.environ.get("SCREENSHOT_DIR")
+    screenshot_dir = environment.SCREENSHOT_DIR.get()
     if DISPLAY is None or screenshot_dir is None:
         return
 

tests/units/test_config.py

@@ -8,6 +8,8 @@ import pytest
 import reflex as rx
 import reflex.config
 from reflex.config import (
+    EnvVar,
+    env_var,
     environment,
     interpret_boolean_env,
     interpret_enum_env,
@@ -214,7 +216,7 @@ def test_replace_defaults(
 
 
 def reflex_dir_constant() -> Path:
-    return environment.REFLEX_DIR
+    return environment.REFLEX_DIR.get()
 
 
 def test_reflex_dir_env_var(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
@@ -227,6 +229,7 @@ def test_reflex_dir_env_var(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) ->
     monkeypatch.setenv("REFLEX_DIR", str(tmp_path))
 
     mp_ctx = multiprocessing.get_context(method="spawn")
+    assert reflex_dir_constant() == tmp_path
     with mp_ctx.Pool(processes=1) as pool:
         assert pool.apply(reflex_dir_constant) == tmp_path
 
@@ -242,3 +245,38 @@ def test_interpret_int_env() -> None:
 @pytest.mark.parametrize("value, expected", [("true", True), ("false", False)])
 def test_interpret_bool_env(value: str, expected: bool) -> None:
     assert interpret_boolean_env(value, "TELEMETRY_ENABLED") == expected
+
+
+def test_env_var():
+    class TestEnv:
+        BLUBB: EnvVar[str] = env_var("default")
+        INTERNAL: EnvVar[str] = env_var("default", internal=True)
+        BOOLEAN: EnvVar[bool] = env_var(False)
+
+    assert TestEnv.BLUBB.get() == "default"
+    assert TestEnv.BLUBB.name == "BLUBB"
+    TestEnv.BLUBB.set("new")
+    assert os.environ.get("BLUBB") == "new"
+    assert TestEnv.BLUBB.get() == "new"
+    TestEnv.BLUBB.set(None)
+    assert "BLUBB" not in os.environ
+
+    assert TestEnv.INTERNAL.get() == "default"
+    assert TestEnv.INTERNAL.name == "__INTERNAL"
+    TestEnv.INTERNAL.set("new")
+    assert os.environ.get("__INTERNAL") == "new"
+    assert TestEnv.INTERNAL.get() == "new"
+    assert TestEnv.INTERNAL.getenv() == "new"
+    TestEnv.INTERNAL.set(None)
+    assert "__INTERNAL" not in os.environ
+
+    assert TestEnv.BOOLEAN.get() is False
+    assert TestEnv.BOOLEAN.name == "BOOLEAN"
+    TestEnv.BOOLEAN.set(True)
+    assert os.environ.get("BOOLEAN") == "True"
+    assert TestEnv.BOOLEAN.get() is True
+    TestEnv.BOOLEAN.set(False)
+    assert os.environ.get("BOOLEAN") == "False"
+    assert TestEnv.BOOLEAN.get() is False
+    TestEnv.BOOLEAN.set(None)
+    assert "BOOLEAN" not in os.environ

tests/units/utils/test_utils.py

@@ -10,6 +10,7 @@ from packaging import version
 
 from reflex import constants
 from reflex.base import Base
+from reflex.config import environment
 from reflex.event import EventHandler
 from reflex.state import BaseState
 from reflex.utils import (
@@ -593,3 +594,11 @@ def test_style_prop_with_event_handler_value(callable):
         rx.box(
             style=style,  # type: ignore
         )
+
+
+def test_is_prod_mode() -> None:
+    """Test that the prod mode is correctly determined."""
+    environment.REFLEX_ENV_MODE.set(constants.Env.PROD)
+    assert utils_exec.is_prod_mode()
+    environment.REFLEX_ENV_MODE.set(None)
+    assert not utils_exec.is_prod_mode()