move all environment variables to the same place (#4192)

* move all environment variables to the same place

* reorder things around

* move more variables to environment

* remove cyclical imports

* forgot default value for field

* for some reason type hints aren't being interpreted

* put the field type *before* not after

* make it get

* move a bit more

* add more fields

* move reflex dir

* add return

* put things somewhere else

* add custom error

reflex/app.py

@@ -64,7 +64,7 @@ from reflex.components.core.client_side_routing import (
 )
 from reflex.components.core.upload import Upload, get_upload_dir
 from reflex.components.radix import themes
-from reflex.config import get_config
+from reflex.config import environment, get_config
 from reflex.event import Event, EventHandler, EventSpec, window_alert
 from reflex.model import Model, get_db_status
 from reflex.page import (
@@ -957,15 +957,16 @@ class App(MiddlewareMixin, LifespanMixin, Base):
         executor = None
         if (
             platform.system() in ("Linux", "Darwin")
-            and os.environ.get("REFLEX_COMPILE_PROCESSES") is not None
+            and (number_of_processes := environment.REFLEX_COMPILE_PROCESSES)
+            is not None
         ):
             executor = concurrent.futures.ProcessPoolExecutor(
-                max_workers=int(os.environ.get("REFLEX_COMPILE_PROCESSES", 0)) or None,
+                max_workers=number_of_processes,
                 mp_context=multiprocessing.get_context("fork"),
             )
         else:
             executor = concurrent.futures.ThreadPoolExecutor(
-                max_workers=int(os.environ.get("REFLEX_COMPILE_THREADS", 0)) or None,
+                max_workers=environment.REFLEX_COMPILE_THREADS
             )
 
         with executor:

reflex/compiler/compiler.py

@@ -2,7 +2,6 @@
 
 from __future__ import annotations
 
-import os
 from datetime import datetime
 from pathlib import Path
 from typing import Dict, Iterable, Optional, Type, Union
@@ -16,7 +15,7 @@ from reflex.components.component import (
     CustomComponent,
     StatefulComponent,
 )
-from reflex.config import get_config
+from reflex.config import environment, get_config
 from reflex.state import BaseState
 from reflex.style import SYSTEM_COLOR_MODE
 from reflex.utils.exec import is_prod_mode
@@ -527,7 +526,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 os.environ.get("REFLEX_PERSIST_WEB_DIR"):
+    if not is_prod_mode() and environment.REFLEX_PERSIST_WEB_DIR:
         # Skip purging the web directory in dev mode if REFLEX_PERSIST_WEB_DIR is set.
         return
 

reflex/components/core/upload.py

@@ -2,13 +2,13 @@
 
 from __future__ import annotations
 
-import os
 from pathlib import Path
 from typing import Any, Callable, ClassVar, Dict, List, Optional, Tuple
 
 from reflex.components.component import Component, ComponentNamespace, MemoizationLeaf
 from reflex.components.el.elements.forms import Input
 from reflex.components.radix.themes.layout.box import Box
+from reflex.config import environment
 from reflex.constants import Dirs
 from reflex.event import (
     CallableEventSpec,
@@ -125,9 +125,7 @@ def get_upload_dir() -> Path:
     """
     Upload.is_used = True
 
-    uploaded_files_dir = Path(
-        os.environ.get("REFLEX_UPLOADED_FILES_DIR", "./uploaded_files")
-    )
+    uploaded_files_dir = environment.REFLEX_UPLOADED_FILES_DIR
     uploaded_files_dir.mkdir(parents=True, exist_ok=True)
     return uploaded_files_dir
 

reflex/config.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import dataclasses
 import importlib
 import os
 import sys
@@ -9,7 +10,10 @@ import urllib.parse
 from pathlib import Path
 from typing import Any, Dict, List, Optional, Set, Union
 
-from reflex.utils.exceptions import ConfigError
+from typing_extensions import get_type_hints
+
+from reflex.utils.exceptions import ConfigError, EnvironmentVarValueError
+from reflex.utils.types import value_inside_optional
 
 try:
     import pydantic.v1 as pydantic
@@ -131,6 +135,198 @@ class DBConfig(Base):
         return f"{self.engine}://{path}/{self.database}"
 
 
+def get_default_value_for_field(field: dataclasses.Field) -> Any:
+    """Get the default value for a field.
+
+    Args:
+        field: The field.
+
+    Returns:
+        The default value.
+
+    Raises:
+        ValueError: If no default value is found.
+    """
+    if field.default != dataclasses.MISSING:
+        return field.default
+    elif field.default_factory != dataclasses.MISSING:
+        return field.default_factory()
+    else:
+        raise ValueError(
+            f"Missing value for environment variable {field.name} and no default value found"
+        )
+
+
+def interpret_boolean_env(value: str) -> bool:
+    """Interpret a boolean environment variable value.
+
+    Args:
+        value: The environment variable value.
+
+    Returns:
+        The interpreted value.
+
+    Raises:
+        EnvironmentVarValueError: If the value is invalid.
+    """
+    true_values = ["true", "1", "yes", "y"]
+    false_values = ["false", "0", "no", "n"]
+
+    if value.lower() in true_values:
+        return True
+    elif value.lower() in false_values:
+        return False
+    raise EnvironmentVarValueError(f"Invalid boolean value: {value}")
+
+
+def interpret_int_env(value: str) -> int:
+    """Interpret an integer environment variable value.
+
+    Args:
+        value: The environment variable value.
+
+    Returns:
+        The interpreted value.
+
+    Raises:
+        EnvironmentVarValueError: If the value is invalid.
+    """
+    try:
+        return int(value)
+    except ValueError as ve:
+        raise EnvironmentVarValueError(f"Invalid integer value: {value}") from ve
+
+
+def interpret_path_env(value: str) -> Path:
+    """Interpret a path environment variable value.
+
+    Args:
+        value: The environment variable value.
+
+    Returns:
+        The interpreted value.
+
+    Raises:
+        EnvironmentVarValueError: If the path does not exist.
+    """
+    path = Path(value)
+    if not path.exists():
+        raise EnvironmentVarValueError(f"Path does not exist: {path}")
+    return path
+
+
+def interpret_env_var_value(value: str, field: dataclasses.Field) -> Any:
+    """Interpret an environment variable value based on the field type.
+
+    Args:
+        value: The environment variable value.
+        field: The field.
+
+    Returns:
+        The interpreted value.
+
+    Raises:
+        ValueError: If the value is invalid.
+    """
+    field_type = value_inside_optional(field.type)
+
+    if field_type is bool:
+        return interpret_boolean_env(value)
+    elif field_type is str:
+        return value
+    elif field_type is int:
+        return interpret_int_env(value)
+    elif field_type is Path:
+        return interpret_path_env(value)
+
+    else:
+        raise ValueError(
+            f"Invalid type for environment variable {field.name}: {field_type}. This is probably an issue in Reflex."
+        )
+
+
+@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
+
+    # The npm registry to use.
+    NPM_CONFIG_REGISTRY: Optional[str] = None
+
+    # Whether to use Granian for the backend. Otherwise, use Uvicorn.
+    REFLEX_USE_GRANIAN: bool = False
+
+    # The username to use for authentication on python package repository. Username and password must both be provided.
+    TWINE_USERNAME: Optional[str] = None
+
+    # The password to use for authentication on python package repository. Username and password must both be provided.
+    TWINE_PASSWORD: Optional[str] = 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
+
+    # 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
+
+    # The working directory for the next.js commands.
+    REFLEX_WEB_WORKDIR: Path = Path(constants.Dirs.WEB)
+
+    # Path to the alembic config file
+    ALEMBIC_CONFIG: Path = Path(constants.ALEMBIC_CONFIG)
+
+    # Disable SSL verification for HTTPX requests.
+    SSL_NO_VERIFY: bool = False
+
+    # The directory to store uploaded files.
+    REFLEX_UPLOADED_FILES_DIR: Path = 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 seperate threads to compile the frontend and how many. Defaults to `min(32, os.cpu_count() + 4)`.
+    REFLEX_COMPILE_THREADS: Optional[int] = None
+
+    # The directory to store reflex dependencies.
+    REFLEX_DIR: Path = Path(constants.Reflex.DIR)
+
+    # Whether to print the SQL queries if the log level is INFO or lower.
+    SQLALCHEMY_ECHO: bool = False
+
+    # Whether to ignore the redis config error. Some redis servers only allow out-of-band configuration.
+    REFLEX_IGNORE_REDIS_CONFIG_ERROR: bool = False
+
+    # Whether to skip purging the web directory in dev mode.
+    REFLEX_PERSIST_WEB_DIR: bool = False
+
+    # The reflex.build frontend host.
+    REFLEX_BUILD_FRONTEND: str = constants.Templates.REFLEX_BUILD_FRONTEND
+
+    # The reflex.build backend host.
+    REFLEX_BUILD_BACKEND: str = constants.Templates.REFLEX_BUILD_BACKEND
+
+    def __init__(self):
+        """Initialize the environment variables."""
+        type_hints = get_type_hints(type(self))
+
+        for field in dataclasses.fields(self):
+            raw_value = os.getenv(field.name, None)
+
+            field.type = type_hints.get(field.name) or field.type
+
+            value = (
+                interpret_env_var_value(raw_value, field)
+                if raw_value is not None
+                else get_default_value_for_field(field)
+            )
+
+            setattr(self, field.name, value)
+
+
+environment = EnvironmentVariables()
+
+
 class Config(Base):
     """The config defines runtime settings for the app.
 

reflex/constants/base.py

@@ -2,7 +2,6 @@
 
 from __future__ import annotations
 
-import os
 import platform
 from enum import Enum
 from importlib import metadata
@@ -11,6 +10,8 @@ from types import SimpleNamespace
 
 from platformdirs import PlatformDirs
 
+from .utils import classproperty
+
 IS_WINDOWS = platform.system() == "Windows"
 
 
@@ -20,6 +21,8 @@ class Dirs(SimpleNamespace):
     # The frontend directories in a project.
     # The web folder where the NextJS app is compiled to.
     WEB = ".web"
+    # The directory where uploaded files are stored.
+    UPLOADED_FILES = "uploaded_files"
     # The name of the assets directory.
     APP_ASSETS = "assets"
     # The name of the assets directory for external ressource (a subfolder of APP_ASSETS).
@@ -64,21 +67,13 @@ class Reflex(SimpleNamespace):
 
     # Files and directories used to init a new project.
     # The directory to store reflex dependencies.
-    # Get directory value from enviroment variables if it exists.
-    _dir = os.environ.get("REFLEX_DIR", "")
+    # on windows, we use C:/Users/<username>/AppData/Local/reflex.
+    # on macOS, we use ~/Library/Application Support/reflex.
+    # on linux, we use ~/.local/share/reflex.
+    # If user sets REFLEX_DIR envroment variable use that instead.
+    DIR = PlatformDirs(MODULE_NAME, False).user_data_path
 
-    DIR = Path(
-        _dir
-        or (
-            # on windows, we use C:/Users/<username>/AppData/Local/reflex.
-            # on macOS, we use ~/Library/Application Support/reflex.
-            # on linux, we use ~/.local/share/reflex.
-            # If user sets REFLEX_DIR envroment variable use that instead.
-            PlatformDirs(MODULE_NAME, False).user_data_dir
-        )
-    )
     # The root directory of the reflex library.
-
     ROOT_DIR = Path(__file__).parents[2]
 
     RELEASES_URL = f"https://api.github.com/repos/reflex-dev/templates/releases"
@@ -101,27 +96,51 @@ class Templates(SimpleNamespace):
     DEFAULT = "blank"
 
     # The reflex.build frontend host
-    REFLEX_BUILD_FRONTEND = os.environ.get(
-        "REFLEX_BUILD_FRONTEND", "https://flexgen.reflex.run"
-    )
+    REFLEX_BUILD_FRONTEND = "https://flexgen.reflex.run"
 
     # The reflex.build backend host
-    REFLEX_BUILD_BACKEND = os.environ.get(
-        "REFLEX_BUILD_BACKEND", "https://flexgen-prod-flexgen.fly.dev"
-    )
+    REFLEX_BUILD_BACKEND = "https://flexgen-prod-flexgen.fly.dev"
 
-    # The URL to redirect to reflex.build
-    REFLEX_BUILD_URL = (
-        REFLEX_BUILD_FRONTEND + "/gen?reflex_init_token={reflex_init_token}"
-    )
+    @classproperty
+    @classmethod
+    def REFLEX_BUILD_URL(cls):
+        """The URL to redirect to reflex.build.
 
-    # The URL to poll waiting for the user to select a generation.
-    REFLEX_BUILD_POLL_URL = REFLEX_BUILD_BACKEND + "/api/init/{reflex_init_token}"
+        Returns:
+            The URL to redirect to reflex.build.
+        """
+        from reflex.config import environment
 
-    # The URL to fetch the generation's reflex code
-    REFLEX_BUILD_CODE_URL = (
-        REFLEX_BUILD_BACKEND + "/api/gen/{generation_hash}/refactored"
-    )
+        return (
+            environment.REFLEX_BUILD_FRONTEND
+            + "/gen?reflex_init_token={reflex_init_token}"
+        )
+
+    @classproperty
+    @classmethod
+    def REFLEX_BUILD_POLL_URL(cls):
+        """The URL to poll waiting for the user to select a generation.
+
+        Returns:
+            The URL to poll waiting for the user to select a generation.
+        """
+        from reflex.config import environment
+
+        return environment.REFLEX_BUILD_BACKEND + "/api/init/{reflex_init_token}"
+
+    @classproperty
+    @classmethod
+    def REFLEX_BUILD_CODE_URL(cls):
+        """The URL to fetch the generation's reflex code.
+
+        Returns:
+            The URL to fetch the generation's reflex code.
+        """
+        from reflex.config import environment
+
+        return (
+            environment.REFLEX_BUILD_BACKEND + "/api/gen/{generation_hash}/refactored"
+        )
 
     class Dirs(SimpleNamespace):
         """Folders used by the template system of Reflex."""

reflex/constants/config.py

@@ -1,6 +1,5 @@
 """Config constants."""
 
-import os
 from pathlib import Path
 from types import SimpleNamespace
 
@@ -9,7 +8,7 @@ from reflex.constants.base import Dirs, Reflex
 from .compiler import Ext
 
 # Alembic migrations
-ALEMBIC_CONFIG = os.environ.get("ALEMBIC_CONFIG", "alembic.ini")
+ALEMBIC_CONFIG = "alembic.ini"
 
 
 class Config(SimpleNamespace):

reflex/constants/installer.py

@@ -3,9 +3,11 @@
 from __future__ import annotations
 
 import platform
+from pathlib import Path
 from types import SimpleNamespace
 
-from .base import IS_WINDOWS, Reflex
+from .base import IS_WINDOWS
+from .utils import classproperty
 
 
 def get_fnm_name() -> str | None:
@@ -36,12 +38,9 @@ class Bun(SimpleNamespace):
 
     # The Bun version.
     VERSION = "1.1.29"
+
     # Min Bun Version
     MIN_VERSION = "0.7.0"
-    # The directory to store the bun.
-    ROOT_PATH = Reflex.DIR / "bun"
-    # Default bun path.
-    DEFAULT_PATH = ROOT_PATH / "bin" / ("bun" if not IS_WINDOWS else "bun.exe")
 
     # URL to bun install script.
     INSTALL_URL = "https://raw.githubusercontent.com/reflex-dev/reflex/main/scripts/bun_install.sh"
@@ -50,11 +49,31 @@ class Bun(SimpleNamespace):
     WINDOWS_INSTALL_URL = (
         "https://raw.githubusercontent.com/reflex-dev/reflex/main/scripts/install.ps1"
     )
+
     # Path of the bunfig file
     CONFIG_PATH = "bunfig.toml"
 
-    # The environment variable to use the system installed bun.
-    USE_SYSTEM_VAR = "REFLEX_USE_SYSTEM_BUN"
+    @classproperty
+    @classmethod
+    def ROOT_PATH(cls):
+        """The directory to store the bun.
+
+        Returns:
+            The directory to store the bun.
+        """
+        from reflex.config import environment
+
+        return environment.REFLEX_DIR / "bun"
+
+    @classproperty
+    @classmethod
+    def DEFAULT_PATH(cls):
+        """Default bun path.
+
+        Returns:
+            The default bun path.
+        """
+        return cls.ROOT_PATH / "bin" / ("bun" if not IS_WINDOWS else "bun.exe")
 
 
 # FNM config.
@@ -63,17 +82,36 @@ class Fnm(SimpleNamespace):
 
     # The FNM version.
     VERSION = "1.35.1"
-    # The directory to store fnm.
-    DIR = Reflex.DIR / "fnm"
+
     FILENAME = get_fnm_name()
-    # The fnm executable binary.
-    EXE = DIR / ("fnm.exe" if IS_WINDOWS else "fnm")
 
     # The URL to the fnm release binary
     INSTALL_URL = (
         f"https://github.com/Schniz/fnm/releases/download/v{VERSION}/{FILENAME}.zip"
     )
 
+    @classproperty
+    @classmethod
+    def DIR(cls) -> Path:
+        """The directory to store fnm.
+
+        Returns:
+            The directory to store fnm.
+        """
+        from reflex.config import environment
+
+        return environment.REFLEX_DIR / "fnm"
+
+    @classproperty
+    @classmethod
+    def EXE(cls):
+        """The fnm executable binary.
+
+        Returns:
+            The fnm executable binary.
+        """
+        return cls.DIR / ("fnm.exe" if IS_WINDOWS else "fnm")
+
 
 # Node / NPM config
 class Node(SimpleNamespace):
@@ -84,23 +122,41 @@ class Node(SimpleNamespace):
     # The minimum required node version.
     MIN_VERSION = "18.17.0"
 
-    # The node bin path.
-    BIN_PATH = (
-        Fnm.DIR
-        / "node-versions"
-        / f"v{VERSION}"
-        / "installation"
-        / ("bin" if not IS_WINDOWS else "")
-    )
+    @classproperty
+    @classmethod
+    def BIN_PATH(cls):
+        """The node bin path.
 
-    # The default path where node is installed.
-    PATH = BIN_PATH / ("node.exe" if IS_WINDOWS else "node")
+        Returns:
+            The node bin path.
+        """
+        return (
+            Fnm.DIR
+            / "node-versions"
+            / f"v{cls.VERSION}"
+            / "installation"
+            / ("bin" if not IS_WINDOWS else "")
+        )
 
-    # The default path where npm is installed.
-    NPM_PATH = BIN_PATH / "npm"
+    @classproperty
+    @classmethod
+    def PATH(cls):
+        """The default path where node is installed.
 
-    # The environment variable to use the system installed node.
-    USE_SYSTEM_VAR = "REFLEX_USE_SYSTEM_NODE"
+        Returns:
+            The default path where node is installed.
+        """
+        return cls.BIN_PATH / ("node.exe" if IS_WINDOWS else "node")
+
+    @classproperty
+    @classmethod
+    def NPM_PATH(cls):
+        """The default path where npm is installed.
+
+        Returns:
+            The default path where npm is installed.
+        """
+        return cls.BIN_PATH / "npm"
 
 
 class PackageJson(SimpleNamespace):

reflex/constants/utils.py

@@ -0,0 +1,32 @@
+"""Utility functions for constants."""
+
+from typing import Any, Callable, Generic, Type
+
+from typing_extensions import TypeVar
+
+T = TypeVar("T")
+V = TypeVar("V")
+
+
+class classproperty(Generic[T, V]):
+    """A class property decorator."""
+
+    def __init__(self, getter: Callable[[Type[T]], V]) -> None:
+        """Initialize the class property.
+
+        Args:
+            getter: The getter function.
+        """
+        self.getter = getattr(getter, "__func__", getter)
+
+    def __get__(self, instance: Any, owner: Type[T]) -> V:
+        """Get the value of the class property.
+
+        Args:
+            instance: The instance of the class.
+            owner: The class itself.
+
+        Returns:
+            The value of the class property.
+        """
+        return self.getter(owner)

reflex/custom_components/custom_components.py

@@ -17,7 +17,7 @@ import typer
 from tomlkit.exceptions import TOMLKitError
 
 from reflex import constants
-from reflex.config import get_config
+from reflex.config import environment, get_config
 from reflex.constants import CustomComponents
 from reflex.utils import console
 
@@ -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(
-        os.getenv("TWINE_USERNAME"),
+        environment.TWINE_USERNAME,
         "-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(
-        os.getenv("TWINE_PASSWORD"),
+        environment.TWINE_PASSWORD,
         "-p",
         "--password",
         show_default="TWINE_PASSWORD environment variable value if set",

reflex/model.py

@@ -2,9 +2,7 @@
 
 from __future__ import annotations
 
-import os
 from collections import defaultdict
-from pathlib import Path
 from typing import Any, ClassVar, Optional, Type, Union
 
 import alembic.autogenerate
@@ -18,9 +16,8 @@ import sqlalchemy
 import sqlalchemy.exc
 import sqlalchemy.orm
 
-from reflex import constants
 from reflex.base import Base
-from reflex.config import get_config
+from reflex.config import environment, get_config
 from reflex.utils import console
 from reflex.utils.compat import sqlmodel, sqlmodel_field_has_primary_key
 
@@ -41,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 Path(constants.ALEMBIC_CONFIG).exists():
+    if environment.ALEMBIC_CONFIG.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 = os.environ.get("SQLALCHEMY_ECHO") == "True"
+    echo_db_query = environment.SQLALCHEMY_ECHO
     # 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)
@@ -234,7 +231,7 @@ class Model(Base, sqlmodel.SQLModel):  # pyright: ignore [reportGeneralTypeIssue
         Returns:
             tuple of (config, script_directory)
         """
-        config = alembic.config.Config(constants.ALEMBIC_CONFIG)
+        config = alembic.config.Config(environment.ALEMBIC_CONFIG)
         return config, alembic.script.ScriptDirectory(
             config.get_main_option("script_location", default="version"),
         )
@@ -269,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(constants.ALEMBIC_CONFIG),
-            directory=str(Path(constants.ALEMBIC_CONFIG).parent / "alembic"),
+            config=alembic.config.Config(environment.ALEMBIC_CONFIG),
+            directory=str(environment.ALEMBIC_CONFIG.parent / "alembic"),
         )
 
     @classmethod
@@ -290,7 +287,7 @@ class Model(Base, sqlmodel.SQLModel):  # pyright: ignore [reportGeneralTypeIssue
         Returns:
             True when changes have been detected.
         """
-        if not Path(constants.ALEMBIC_CONFIG).exists():
+        if not environment.ALEMBIC_CONFIG.exists():
             return False
 
         config, script_directory = cls._alembic_config()
@@ -391,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 Path(constants.ALEMBIC_CONFIG).exists():
+        if not environment.ALEMBIC_CONFIG.exists():
             return
 
         with cls.get_db_engine().connect() as connection:

reflex/reflex.py

@@ -13,7 +13,7 @@ from reflex_cli.deployments import deployments_cli
 from reflex_cli.utils import dependency
 
 from reflex import constants
-from reflex.config import get_config
+from reflex.config import environment, get_config
 from reflex.custom_components.custom_components import custom_components_cli
 from reflex.state import reset_disk_state_manager
 from reflex.utils import console, redir, telemetry
@@ -420,7 +420,7 @@ def db_init():
         return
 
     # Check the alembic config.
-    if Path(constants.ALEMBIC_CONFIG).exists():
+    if environment.ALEMBIC_CONFIG.exists():
         console.error(
             "Database is already initialized. Use "
             "[bold]reflex db makemigrations[/bold] to create schema change "

reflex/state.py

@@ -8,7 +8,6 @@ import copy
 import dataclasses
 import functools
 import inspect
-import os
 import pickle
 import sys
 import uuid
@@ -64,6 +63,7 @@ from redis.exceptions import ResponseError
 import reflex.istate.dynamic
 from reflex import constants
 from reflex.base import Base
+from reflex.config import environment
 from reflex.event import (
     BACKGROUND_TASK_MARKER,
     Event,
@@ -3274,11 +3274,7 @@ class StateManagerRedis(StateManager):
             )
         except ResponseError:
             # Some redis servers only allow out-of-band configuration, so ignore errors here.
-            ignore_config_error = os.environ.get(
-                "REFLEX_IGNORE_REDIS_CONFIG_ERROR",
-                None,
-            )
-            if not ignore_config_error:
+            if not environment.REFLEX_IGNORE_REDIS_CONFIG_ERROR:
                 raise
         async with self.redis.pubsub() as pubsub:
             await pubsub.psubscribe(lock_key_channel)

reflex/utils/build.py

@@ -23,18 +23,6 @@ def set_env_json():
     )
 
 
-def set_os_env(**kwargs):
-    """Set os environment variables.
-
-    Args:
-        kwargs: env key word args.
-    """
-    for key, value in kwargs.items():
-        if not value:
-            continue
-        os.environ[key.upper()] = value
-
-
 def generate_sitemap_config(deploy_url: str, export=False):
     """Generate the sitemap config file.
 

reflex/utils/exceptions.py

@@ -135,3 +135,7 @@ class SetUndefinedStateVarError(ReflexError, AttributeError):
 
 class StateSchemaMismatchError(ReflexError, TypeError):
     """Raised when the serialized schema of a state class does not match the current schema."""
+
+
+class EnvironmentVarValueError(ReflexError, ValueError):
+    """Raised when an environment variable is set to an invalid value."""

reflex/utils/exec.py

@@ -15,7 +15,7 @@ from urllib.parse import urljoin
 import psutil
 
 from reflex import constants
-from reflex.config import get_config
+from reflex.config import environment, get_config
 from reflex.constants.base import LogLevel
 from reflex.utils import console, path_ops
 from reflex.utils.prerequisites import get_web_dir
@@ -184,7 +184,7 @@ def should_use_granian():
     Returns:
         True if Granian should be used.
     """
-    return os.getenv("REFLEX_USE_GRANIAN", "0") == "1"
+    return environment.REFLEX_USE_GRANIAN
 
 
 def get_app_module():

reflex/utils/net.py

@@ -1,9 +1,8 @@
 """Helpers for downloading files from the network."""
 
-import os
-
 import httpx
 
+from ..config import environment
 from . import console
 
 
@@ -13,8 +12,7 @@ def _httpx_verify_kwarg() -> bool:
     Returns:
         True if SSL verification is enabled, False otherwise
     """
-    ssl_no_verify = os.environ.get("SSL_NO_VERIFY", "").lower() in ["true", "1", "yes"]
-    return not ssl_no_verify
+    return not environment.SSL_NO_VERIFY
 
 
 def get(url: str, **kwargs) -> httpx.Response:

reflex/utils/path_ops.py

@@ -9,6 +9,7 @@ import shutil
 from pathlib import Path
 
 from reflex import constants
+from reflex.config import environment
 
 # Shorthand for join.
 join = os.linesep.join
@@ -129,30 +130,13 @@ def which(program: str | Path) -> str | Path | None:
     return shutil.which(str(program))
 
 
-def use_system_install(var_name: str) -> bool:
-    """Check if the system install should be used.
-
-    Args:
-        var_name: The name of the environment variable.
-
-    Raises:
-        ValueError: If the variable name is invalid.
-
-    Returns:
-        Whether the associated env var should use the system install.
-    """
-    if not var_name.startswith("REFLEX_USE_SYSTEM_"):
-        raise ValueError("Invalid system install variable name.")
-    return os.getenv(var_name, "").lower() in ["true", "1", "yes"]
-
-
 def use_system_node() -> bool:
     """Check if the system node should be used.
 
     Returns:
         Whether the system node should be used.
     """
-    return use_system_install(constants.Node.USE_SYSTEM_VAR)
+    return environment.REFLEX_USE_SYSTEM_NODE
 
 
 def use_system_bun() -> bool:
@@ -161,7 +145,7 @@ def use_system_bun() -> bool:
     Returns:
         Whether the system bun should be used.
     """
-    return use_system_install(constants.Bun.USE_SYSTEM_VAR)
+    return environment.REFLEX_USE_SYSTEM_BUN
 
 
 def get_node_bin_path() -> Path | None:

reflex/utils/prerequisites.py

@@ -33,7 +33,7 @@ from redis.asyncio import Redis
 
 from reflex import constants, model
 from reflex.compiler import templates
-from reflex.config import Config, get_config
+from reflex.config import Config, environment, get_config
 from reflex.utils import console, net, path_ops, processes
 from reflex.utils.exceptions import GeneratedCodeHasNoFunctionDefs
 from reflex.utils.format import format_library_name
@@ -69,8 +69,7 @@ def get_web_dir() -> Path:
     Returns:
         The working directory.
     """
-    workdir = Path(os.getenv("REFLEX_WEB_WORKDIR", constants.Dirs.WEB))
-    return workdir
+    return environment.REFLEX_WEB_WORKDIR
 
 
 def _python_version_check():
@@ -250,7 +249,7 @@ def windows_npm_escape_hatch() -> bool:
     Returns:
         If the user has set REFLEX_USE_NPM.
     """
-    return os.environ.get("REFLEX_USE_NPM", "").lower() in ["true", "1", "yes"]
+    return environment.REFLEX_USE_NPM
 
 
 def get_app(reload: bool = False) -> ModuleType:
@@ -992,7 +991,7 @@ def needs_reinit(frontend: bool = True) -> bool:
         return False
 
     # Make sure the .reflex directory exists.
-    if not constants.Reflex.DIR.exists():
+    if not environment.REFLEX_DIR.exists():
         return True
 
     # Make sure the .web directory exists in frontend mode.
@@ -1097,7 +1096,7 @@ def ensure_reflex_installation_id() -> Optional[int]:
     """
     try:
         initialize_reflex_user_directory()
-        installation_id_file = constants.Reflex.DIR / "installation_id"
+        installation_id_file = environment.REFLEX_DIR / "installation_id"
 
         installation_id = None
         if installation_id_file.exists():
@@ -1122,7 +1121,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(constants.Reflex.DIR)
+    path_ops.mkdir(environment.REFLEX_DIR)
 
 
 def initialize_frontend_dependencies():
@@ -1145,7 +1144,7 @@ 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 Path(constants.ALEMBIC_CONFIG).exists():
+    if get_config().db_url is not None and not environment.ALEMBIC_CONFIG.exists():
         console.error(
             "Database is not initialized. Run [bold]reflex db init[/bold] first."
         )
@@ -1155,7 +1154,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 Path(constants.ALEMBIC_CONFIG).exists():
+    if get_config().db_url is None or not environment.ALEMBIC_CONFIG.exists():
         return
     with model.Model.get_db_engine().connect() as connection:
         try:

reflex/utils/registry.py

@@ -1,9 +1,8 @@
 """Utilities for working with registries."""
 
-import os
-
 import httpx
 
+from reflex.config import environment
 from reflex.utils import console, net
 
 
@@ -56,7 +55,4 @@ def _get_npm_registry() -> str:
     Returns:
         str:
     """
-    if npm_registry := os.environ.get("NPM_CONFIG_REGISTRY", ""):
-        return npm_registry
-    else:
-        return get_best_registry()
+    return environment.NPM_CONFIG_REGISTRY or get_best_registry()

reflex/utils/types.py

@@ -274,6 +274,20 @@ def is_optional(cls: GenericType) -> bool:
     return is_union(cls) and type(None) in get_args(cls)
 
 
+def value_inside_optional(cls: GenericType) -> GenericType:
+    """Get the value inside an Optional type or the original type.
+
+    Args:
+        cls: The class to check.
+
+    Returns:
+        The value inside the Optional type or the original type.
+    """
+    if is_union(cls) and len(args := get_args(cls)) >= 2 and type(None) in args:
+        return unionize(*[arg for arg in args if arg is not type(None)])
+    return cls
+
+
 def get_property_hint(attr: Any | None) -> GenericType | None:
     """Check if an attribute is a property and return its type hint.
 

tests/units/test_config.py

@@ -5,6 +5,7 @@ import pytest
 
 import reflex as rx
 import reflex.config
+from reflex.config import environment
 from reflex.constants import Endpoint
 
 
@@ -178,7 +179,7 @@ def test_replace_defaults(
 
 
 def reflex_dir_constant():
-    return rx.constants.Reflex.DIR
+    return environment.REFLEX_DIR
 
 
 def test_reflex_dir_env_var(monkeypatch, tmp_path):