[V5] openbb-core: Add Backwards-Compatible Layered Config System (openbb.toml) (#7492)

* add backwards-compatible layered config system

* fix path reference in windows test

* need python version-aware handling in docstring generator
This commit is contained in:
Danglewood
2026-05-07 12:37:05 -07:00
committed by GitHub
parent f18493c293
commit ac39b23c9b
9 changed files with 1574 additions and 3 deletions

View File

@@ -0,0 +1,27 @@
"""Layered configuration discovery for openbb-core.
Mirrors the resolution order ``openbb-cli`` ships with — pyproject →
user-global → project → explicit → .env → real-shell env vars — and
maps the merged result onto the existing ``SystemService`` and
``UserService`` singletons. New code should import from this
subpackage; the legacy ``user_settings.json`` / ``system_settings.json``
disk files keep working unchanged as one layer in the cascade.
"""
from openbb_core.app.config.loader import (
apply_config_to_services,
apply_settings_to_env,
load_config,
load_env_files,
load_layered_config,
render_config_template,
)
__all__ = [
"apply_config_to_services",
"apply_settings_to_env",
"load_config",
"load_env_files",
"load_layered_config",
"render_config_template",
]

View File

@@ -0,0 +1,545 @@
"""Layered configuration loader for openbb-core.
Resolution order (highest priority last; later layers overwrite earlier ones):
1. Built-in defaults (model field defaults on ``SystemSettings`` /
``UserSettings``).
2. The on-disk JSONs the existing services have always loaded —
``~/.openbb_platform/system_settings.json`` and
``user_settings.json``. Treated as defaults so a user who has only
ever touched the JSONs sees zero behavior change.
3. ``[tool.openbb]`` in the nearest ancestor ``pyproject.toml``,
walking up from CWD. Lets a library/service ship a known-good
baseline (default credentials shape, preferred output format) with
its distribution.
4. ``~/.openbb_platform/openbb.toml`` (or ``.openbb.toml``) — the
user-global config in the same directory the JSON files live in,
intended as the friendlier sibling of those raw JSON dumps.
5. ``openbb.toml`` (or ``.openbb.toml``) in the nearest ancestor
directory of CWD — project-local override that doesn't pollute
``pyproject.toml``.
6. ``$OPENBB_CONFIG`` env var (or an explicit ``path=`` to
``load_config``) — swappable, scoped configurations
(``staging.toml``, ``prod.toml``, ...).
7. ``.env`` files (``~/.openbb_platform/.env`` plus
``$OPENBB_ENV_FILE``) — applied to ``os.environ`` so the env-var
layer below picks them up without coupling.
8. ``OPENBB_*`` real-shell env vars — already honored by the
``Env`` singleton and various model fields. Real shell exports
always beat dotfiles + TOML.
This module covers (3)-(7); the existing services + ``Env`` cover (1),
(2), and (8). The merged result of (3)-(6) is the "layered config" —
``apply_config_to_services`` pushes it onto the singleton services so
downstream code reads the layered values via the existing
``SystemService().system_settings`` / ``UserService().default_user_settings``
APIs.
Schema (every key + section optional)::
# Top-level promotions — convenience shortcuts for the most-tweaked
# System fields. Setting these here is identical to the matching
# entry under ``[system]``; both forms are accepted, the top-level
# form wins on conflicts.
debug-mode = true
test-mode = false
headless = false
[system]
debug_mode = true
headless = false
[system.api_settings]
prefix = "/api/v1"
[system.api_settings.cors]
allow_origins = ["*"]
[user]
[user.credentials]
fmp_api_key = "..."
polygon_api_key = "..."
[user.preferences]
data_directory = "/data/openbb"
output_type = "dataframe"
[user.defaults]
[user.defaults.commands."/equity/price/historical"]
provider = "fmp"
The ``[system]``, ``[user]``, and their nested tables are deep-merged
across layers. Top-level promotions cascade onto the corresponding
``[system]`` keys (top-level wins). See the ``render_config_template``
output for a fully documented version.
"""
from __future__ import annotations
import os
import sys
from pathlib import Path
from typing import Any
if sys.version_info >= (3, 11): # pragma: no cover — version-conditional import
import tomllib
else: # pragma: no cover — exercised only on the 3.10 backport path
import tomli as tomllib # ty: ignore[unresolved-import]
from openbb_core.app.constants import OPENBB_DIRECTORY
DEFAULT_CONFIG_NAMES: tuple[str, ...] = ("openbb.toml", ".openbb.toml")
PYPROJECT_NAME = "pyproject.toml"
PYPROJECT_TABLE: tuple[str, ...] = ("tool", "openbb")
EXPLICIT_CONFIG_ENV = "OPENBB_CONFIG"
EXPLICIT_ENV_FILE_ENV = "OPENBB_ENV_FILE"
USER_OPENBB_DIR = OPENBB_DIRECTORY
USER_OPENBB_TOML_NAMES: tuple[str, ...] = ("openbb.toml", ".openbb.toml")
USER_OPENBB_ENV_NAME = ".env"
# Top-level convenience keys that promote onto the matching ``[system]``
# field. Same idea as the cli's ``_TOP_LEVEL_SETTINGS_PROMOTIONS`` —
# the most-tweaked subset of the System surface gets first-class
# top-level placement so users don't have to dig into ``[system]``.
_TOP_LEVEL_SYSTEM_PROMOTIONS: tuple[str, ...] = (
"debug_mode",
"test_mode",
"headless",
"logging_suppress",
"allow_mutable_extensions",
"allow_on_command_output",
)
def _walk_up(start: Path | None = None):
"""Yield ``start`` and every parent up to the filesystem root."""
cur = (start or Path.cwd()).resolve()
yield cur
yield from cur.parents
def _read_toml(path: Path) -> dict[str, Any]:
"""Safe-load a TOML file. Missing / unreadable files yield ``{}``."""
try:
with path.open("rb") as fh:
return tomllib.load(fh)
except (OSError, tomllib.TOMLDecodeError):
return {}
def _find_first(start: Path | None, names: tuple[str, ...]) -> Path | None:
"""Walk up from ``start`` looking for any filename in ``names``."""
for parent in _walk_up(start):
for name in names:
candidate = parent / name
if candidate.is_file():
return candidate
return None
def _find_pyproject_section(start: Path | None = None) -> dict[str, Any]:
"""Find the nearest ancestor ``pyproject.toml`` with ``[tool.openbb]``.
Walks up from ``start`` (or CWD) and returns the section dict, or
``{}`` when no matching pyproject is found.
"""
py = _find_first(start, (PYPROJECT_NAME,))
if py is None:
return {}
data = _read_toml(py)
node: Any = data
for key in PYPROJECT_TABLE:
if not isinstance(node, dict) or key not in node:
return {}
node = node[key]
return node if isinstance(node, dict) else {}
def _deep_merge(base: dict[str, Any], override: dict[str, Any]) -> None:
"""In-place deep merge: nested dicts merge, scalars / lists overwrite."""
for k, v in override.items():
if k in base and isinstance(base[k], dict) and isinstance(v, dict):
_deep_merge(base[k], v)
else:
base[k] = v
def _normalize_keys(d: dict[str, Any]) -> dict[str, Any]:
"""Normalize top-level kebab-case keys to snake_case.
The ``[system]`` / ``[user]`` nested tables keep their original
keys — those names go straight into Settings model field lookups
so they must round-trip exactly. Only the top-level convenience
keys (``debug-mode``, ``test-mode``, ...) get the snake-case
treatment so users can write either form.
"""
return {k.replace("-", "_"): v for k, v in d.items()}
def _user_global_toml() -> Path | None:
"""Locate ``~/.openbb_platform/openbb.toml`` (or the dotted variant)."""
if not USER_OPENBB_DIR.is_dir():
return None
for name in USER_OPENBB_TOML_NAMES:
candidate = USER_OPENBB_DIR / name
if candidate.is_file():
return candidate
return None
def load_config(
explicit_path: str | os.PathLike[str] | None = None,
*,
start: Path | None = None,
) -> dict[str, Any]:
"""Resolve the layered openbb-core configuration.
Layers, lowest to highest priority:
* ``[tool.openbb]`` from the nearest ancestor ``pyproject.toml``
* ``~/.openbb_platform/openbb.toml`` (or ``.openbb.toml``)
* ``openbb.toml`` (or ``.openbb.toml``) walking up from ``start``
(defaults to CWD)
* ``explicit_path`` if provided, otherwise ``$OPENBB_CONFIG``
Returns a single merged dict with normalized top-level keys
(kebab-case → snake_case). Nested ``[system]`` / ``[user]`` tables
keep their original casing so they map directly to model fields.
Returns ``{}`` when no layer is found — callers should fall back
to disk JSONs / model defaults.
"""
layers: list[dict[str, Any]] = [_find_pyproject_section(start)]
user_toml = _user_global_toml()
if user_toml is not None:
layers.append(_read_toml(user_toml))
project_toml = _find_first(start, DEFAULT_CONFIG_NAMES)
if project_toml is not None:
layers.append(_read_toml(project_toml))
explicit = explicit_path or os.environ.get(EXPLICIT_CONFIG_ENV)
if explicit:
path = Path(explicit).expanduser()
if path.is_file():
layers.append(_read_toml(path))
merged: dict[str, Any] = {}
for layer in layers:
_deep_merge(merged, _normalize_keys(layer))
return merged
def _promote_top_level_keys(config: dict[str, Any]) -> dict[str, Any]:
"""Fold top-level convenience keys into ``[system]``.
Returns a new dict — caller's input untouched. Top-level wins on
conflict because users who set both clearly meant the explicit
short form.
"""
if not config:
return {}
out = {k: v for k, v in config.items() if k != "system" or isinstance(v, dict)}
system_section: dict[str, Any] = dict(out.get("system") or {})
for key in _TOP_LEVEL_SYSTEM_PROMOTIONS:
if key in out:
system_section[key] = out.pop(key)
if system_section:
out["system"] = system_section
return out
def apply_config_to_services(config: dict[str, Any] | None) -> dict[str, list[str]]:
"""Push the layered config onto ``SystemService`` and ``UserService``.
Both services are singletons that already loaded their on-disk JSON
(the canonical "user has manually edited this" surface). This
function takes the layered TOML config — typically the output of
``load_config()`` — and merges its ``[system]`` / ``[user]``
sections into the live service settings, then re-validates so any
nested-model coercion (CORS lists, Defaults sub-tables, etc.)
runs against the merged dict rather than the raw TOML.
Real shell exports through the ``Env`` singleton stay
authoritative — TOML provides defaults, env vars override them via
the model-field validator paths.
Returns ``{"system": [field1, ...], "user": [field1, ...]}``: the
fields each service updated, for logging / debugging.
"""
from pydantic import BaseModel
from openbb_core.app.model.credentials import Credentials
from openbb_core.app.model.defaults import Defaults
from openbb_core.app.model.preferences import Preferences
from openbb_core.app.model.system_settings import SystemSettings
from openbb_core.app.service.system_service import SystemService
from openbb_core.app.service.user_service import UserService
if not config:
return {"system": [], "user": []}
promoted = _promote_top_level_keys(config)
applied: dict[str, list[str]] = {"system": [], "user": []}
system_overrides = promoted.get("system") or {}
if isinstance(system_overrides, dict) and system_overrides:
service = SystemService()
current = service.system_settings.model_dump()
merged = _deep_merge_copy(current, system_overrides)
# ``SystemSettings`` is frozen — replace, don't mutate.
service.system_settings = SystemSettings.model_validate(merged)
applied["system"] = sorted(system_overrides.keys())
user_overrides = promoted.get("user") or {}
if isinstance(user_overrides, dict) and user_overrides:
# ``UserSettings.__init__`` re-loads from disk on construction —
# going through ``model_validate`` would clobber the merged
# values with whatever's already on disk. Patch the live
# singleton's sub-models instead so the layered config wins
# without the __init__ override fighting us.
u_service = UserService()
live = u_service.default_user_settings
sub_model_map: dict[str, type[BaseModel]] = {
"credentials": Credentials,
"preferences": Preferences,
"defaults": Defaults,
}
for section, model_cls in sub_model_map.items():
section_overrides = user_overrides.get(section)
if not isinstance(section_overrides, dict):
continue
current_section = getattr(live, section).model_dump()
merged_section = _deep_merge_copy(current_section, section_overrides)
setattr(live, section, model_cls.model_validate(merged_section))
applied["user"] = sorted(user_overrides.keys())
return applied
def _deep_merge_copy(base: dict[str, Any], override: dict[str, Any]) -> dict[str, Any]:
"""Non-mutating ``_deep_merge`` — returns a fresh dict."""
out: dict[str, Any] = {}
_deep_merge(out, base)
_deep_merge(out, override)
return out
def load_env_files(
explicit_path: str | os.PathLike[str] | None = None,
) -> list[Path]:
"""Apply ``.env`` files into ``os.environ`` for subsequent env lookups.
Layers, applied in order so later files override earlier ones, and
pre-existing ``os.environ`` values still win (real shell exports
always beat dotfiles):
* ``~/.openbb_platform/.env`` — the canonical openbb-platform
dotenv that ``Env`` already loads at import time. Re-loading
here is a no-op when ``Env`` already ran but covers the case
where this loader runs first.
* ``explicit_path`` if provided, otherwise
``$OPENBB_ENV_FILE``.
Returns the list of files actually applied, for logging.
"""
# pylint: disable=import-outside-toplevel
from dotenv import dotenv_values
candidates: list[Path] = []
user_env = USER_OPENBB_DIR / USER_OPENBB_ENV_NAME
if user_env.is_file():
candidates.append(user_env)
explicit = explicit_path or os.environ.get(EXPLICIT_ENV_FILE_ENV)
if explicit:
path = Path(explicit).expanduser()
if path.is_file():
candidates.append(path)
loaded: list[Path] = []
for path in candidates:
for k, v in dotenv_values(path).items():
if v is None:
continue
os.environ.setdefault(k, v)
loaded.append(path)
return loaded
def apply_settings_to_env(config: dict[str, Any] | None) -> list[str]:
"""Inject promoted top-level system flags into ``os.environ``.
The few system-level flags ``Env`` reads (``OPENBB_DEBUG_MODE``,
``OPENBB_ALLOW_MUTABLE_EXTENSIONS``, etc.) get seeded from the
layered config so code that consults ``Env`` rather than
``SystemService`` (some pre-existing call sites) sees the same
values as ``SystemSettings``. Uses ``setdefault`` so real shell
exports stay authoritative.
Returns the list of env var names set, for logging.
"""
if not config:
return []
promoted = _promote_top_level_keys(config)
applied: list[str] = []
system_section = promoted.get("system") or {}
if not isinstance(system_section, dict): # pragma: no cover — defensive
# ``_promote_top_level_keys`` already filters out non-dict
# ``system`` values, so this guard is unreachable via the
# normal load path. Kept in case a future refactor opens the
# door to a non-dict landing here.
return []
for key, value in system_section.items():
if key not in _TOP_LEVEL_SYSTEM_PROMOTIONS:
continue
env_key = f"OPENBB_{key.upper()}"
if env_key in os.environ:
continue
if isinstance(value, bool):
os.environ[env_key] = "True" if value else "False"
else:
os.environ[env_key] = str(value)
applied.append(env_key)
return applied
def load_layered_config(
explicit_path: str | os.PathLike[str] | None = None,
*,
explicit_env_file: str | os.PathLike[str] | None = None,
apply_to_services: bool = True,
apply_to_env: bool = True,
) -> dict[str, Any]:
"""End-to-end discovery: TOML cascade + ``.env`` + optional service push.
Convenience wrapper that bundles the steps a typical bootstrap
sequence runs:
1. ``load_env_files`` — drop dotfile values into ``os.environ``
(real exports still win via ``setdefault``).
2. ``load_config`` — walk the TOML cascade and produce the merged
dict.
3. ``apply_settings_to_env`` (when ``apply_to_env=True``) — seed
``OPENBB_*`` env vars for the promoted top-level flags so any
code path reading via ``Env`` sees the same values.
4. ``apply_config_to_services`` (when ``apply_to_services=True``)
— push ``[system]`` / ``[user]`` sections onto the singleton
services so existing call sites surface the layered values.
Returns the merged config dict so callers can inspect / re-apply
if needed.
"""
load_env_files(explicit_env_file)
config = load_config(explicit_path)
if apply_to_env:
apply_settings_to_env(config)
if apply_to_services:
apply_config_to_services(config)
return config
def _toml_quote(value: Any) -> str:
"""Render a Python value as a TOML literal (string / bool / number)."""
if isinstance(value, str):
return '"' + value.replace("\\", "\\\\").replace('"', '\\"') + '"'
if isinstance(value, bool):
return "true" if value else "false"
if isinstance(value, (int, float)):
return str(value)
if isinstance(value, list):
return "[" + ", ".join(_toml_quote(v) for v in value) + "]"
return '"' + str(value) + '"'
def render_config_template(active: dict[str, Any] | None = None) -> str:
"""Render a documented TOML template covering every supported setting.
Each section walks the live ``SystemSettings`` / ``UserSettings``
model fields and emits a comment block + commented-out line per
field. Fields with values resolved from any layer surface
uncommented so users can tweak in place. The top-level promotions
block is rendered first as the "most likely tweak" surface.
"""
# pylint: disable=import-outside-toplevel
from openbb_core.app.model.system_settings import SystemSettings
from openbb_core.app.model.user_settings import UserSettings
a = active or {}
parts: list[str] = [
"# openbb-core configuration template.\n"
"#\n"
"# Drop this file at any of the following discovery locations:\n"
"# * ~/.openbb_platform/openbb.toml (user-global)\n"
"# * ./openbb.toml or ./.openbb.toml (project-local; walks up from CWD)\n"
"# * [tool.openbb] in pyproject.toml (ships with a dev project)\n"
"# * --config PATH or $OPENBB_CONFIG (explicit, swappable)\n"
"#\n"
"# Resolution order (lowest → highest priority):\n"
"# defaults -> system_settings.json / user_settings.json -> pyproject\n"
"# -> ~/.openbb_platform/openbb.toml -> ./openbb.toml -> $OPENBB_CONFIG\n"
"# -> .env files -> OPENBB_* env vars\n"
"#\n"
"# Lines starting with # are commented out. Either uncomment to set a value\n"
"# (and the matching env var override still wins), or leave commented to\n"
"# fall through to the next layer.\n\n",
"# ── Top-level promotions ────────────────────────────────────────────────\n"
"# Convenience shortcuts for the most-tweaked System fields. Setting these\n"
"# at the top level overrides the matching key under [system].\n\n",
]
for promoted in _TOP_LEVEL_SYSTEM_PROMOTIONS:
kebab = promoted.replace("_", "-")
active_value = a.get(promoted) or (a.get("system") or {}).get(promoted)
commented = "# " if active_value is None else ""
rendered = _toml_quote(active_value) if active_value is not None else "false"
parts.append(
f"# OPENBB_{promoted.upper()} (env equivalent)\n"
f"{commented}{kebab} = {rendered}\n\n"
)
parts.append(
"# ── [system] ────────────────────────────────────────────────────────────\n"
"# Maps directly onto the SystemSettings model. Every field surfaces here;\n"
"# nested tables (api_settings, python_settings) get their own sub-sections.\n"
"[system]\n\n"
)
parts.append(_render_settings_section("system", SystemSettings, a.get("system")))
parts.append(
"# ── [user] ──────────────────────────────────────────────────────────────\n"
"# Maps onto the UserSettings model — credentials, preferences, defaults.\n"
"[user]\n\n"
)
parts.append(_render_settings_section("user", UserSettings, a.get("user")))
return "".join(parts)
def _render_settings_section(
name: str,
model: Any,
active: dict[str, Any] | None,
) -> str:
"""Render a section's commented-out fields from the model definition."""
# pylint: disable=import-outside-toplevel
out: list[str] = []
active_dict = active if isinstance(active, dict) else {}
for field_name, field in sorted(model.model_fields.items()):
if field_name.startswith("_"): # pragma: no cover — defensive
# Pydantic doesn't expose ``_``-prefixed names in
# ``model_fields``; the guard exists in case a future
# subclass reaches this with private attrs surfaced.
continue
active_value = active_dict.get(field_name)
if isinstance(active_value, dict):
# Nested table — emit as ``[name.field_name]``. The actual
# nested fields render as a sub-call.
out.append(f"[{name}.{field_name}]\n\n")
continue
if isinstance(active_value, list):
commented = ""
rendered = _toml_quote(active_value)
elif active_value is None:
commented = "# "
default = field.default if field.default is not None else ""
rendered = _toml_quote(default)
else:
commented = ""
rendered = _toml_quote(active_value)
description = (field.description or field_name).split("\n")[0]
out.append(f"# {description}\n{commented}{field_name} = {rendered}\n\n")
return "".join(out)

View File

@@ -0,0 +1,104 @@
"""AgGrid Server-Side Row Model (SSRM) request/response shapes.
Routes whose POST body is (or subclasses) ``AgGridRowsRequest`` and/or
whose 200 response is ``AgGridRowsResponse`` get auto-promoted to widget
``type: "ssrm_table"`` by the platform_api widgets.json generator. The
type signature is the opt-in — authors don't need to set
``widget_config={"type": "ssrm_table"}`` on the route.
The field names mirror what AgGrid's server-side datasource sends and
expects, so they are camelCase by design and must not be renamed.
"""
from __future__ import annotations
from typing import Any, Generic, Literal, TypeVar
from pydantic import BaseModel, ConfigDict, Field
Row = TypeVar("Row")
SortDirection = Literal["asc", "desc"]
FilterType = Literal["text", "number", "date", "set", "boolean"]
class AgSortColumn(BaseModel):
"""One entry in AgGrid's ``sortModel``.
AgGrid emits ``[{"colId": "<field>", "sort": "asc"|"desc"}]`` for
every active sort. Order matters — the first entry is the primary
sort key.
"""
model_config = ConfigDict(extra="forbid")
colId: str
sort: SortDirection
class AgFilterModel(BaseModel):
"""One entry in AgGrid's ``filterModel`` keyed by colId.
AgGrid's filter payload varies by column type (text/number/date/set
each have their own operator vocabulary), so ``extra="allow"`` keeps
the door open for the fields ag-grid adds for advanced operators
(``filterTo``, ``dateFrom``, ``dateTo``, ``conditions``, ...).
"""
model_config = ConfigDict(extra="allow")
filterType: FilterType | None = None
type: str | None = None
filter: Any = None
filterTo: Any = None
values: list[Any] | None = None
class AgGroupingModel(BaseModel):
"""Row-grouping descriptor for SSRM.
Sent when the user drags a column into the row-group panel. The
backend only sees this when SSRM grouping is enabled on the widget.
"""
model_config = ConfigDict(extra="forbid")
rowGroupCols: list[dict[str, Any]] = Field(default_factory=list)
valueCols: list[dict[str, Any]] = Field(default_factory=list)
pivotCols: list[dict[str, Any]] = Field(default_factory=list)
pivotMode: bool = False
groupKeys: list[str] = Field(default_factory=list)
class AgGridRowsRequest(BaseModel):
"""Request body sent by an AgGrid SSRM datasource.
Subclass this on a route to add provider-specific filters/params;
the widget generator detects the SSRM contract by walking up the
schema's ``allOf`` chain, so subclasses are recognized just like
direct uses.
"""
model_config = ConfigDict(extra="allow")
startRow: int = Field(ge=0)
endRow: int = Field(ge=0)
sortModel: list[AgSortColumn] = Field(default_factory=list)
filterModel: dict[str, AgFilterModel] = Field(default_factory=dict)
grouping: AgGroupingModel | None = None
class AgGridRowsResponse(BaseModel, Generic[Row]):
"""Response body the SSRM datasource expects from the backend.
``rowData`` carries the page; ``rowCount`` is the *total* number of
rows the datasource should report to AgGrid. ``rowCount`` lets the
grid render the scroll thumb correctly even before all pages have
loaded — set it to ``len(rowData)`` only when this is the last
page.
"""
model_config = ConfigDict(extra="forbid")
rowData: list[Row]
rowCount: int = Field(ge=0)

View File

@@ -96,7 +96,13 @@ class DocstringGenerator:
_type = "Optional[int]" if is_optional else "int"
origin = get_origin(_type)
if origin is Union:
# On Python 3.10-3.13, ``X | Y`` produces ``types.UnionType``
# whose ``get_origin()`` is ``types.UnionType`` — distinct from
# ``typing.Union``. 3.14 unified them, but until 3.13 is gone
# we have to accept both so the container-aware ``openbb_*``
# strip below covers ``list[X] | None`` as well as
# ``Union[list[X], None]``.
if origin is Union or origin is UnionType:
args = get_args(_type)
type_names = []
has_none = False

View File

@@ -19,6 +19,7 @@ dependencies = [
"aiohttp>=3.13.3",
"ruff>=0.15,<0.16",
"pyjwt>=2.12.0,<3.0.0",
"tomli>=2.0.0; python_version<'3.11'",
]
[project.optional-dependencies]

View File

@@ -0,0 +1,682 @@
"""Tests for ``openbb_core.app.config.loader``.
Covers TOML cascade discovery, deep-merge behavior, top-level
promotion, ``.env`` loading, and the ``apply_config_to_services``
push onto the singleton services.
"""
import os
from pathlib import PurePosixPath
import pytest
@pytest.fixture
def reset_singletons():
"""Snapshot + restore SystemService / UserService state.
Both services are singletons keyed by their class — we mutate the
settings via ``apply_config_to_services`` then need to roll back so
other tests aren't affected.
"""
from openbb_core.app.service.system_service import SystemService
from openbb_core.app.service.user_service import UserService
sys_orig = SystemService().system_settings
user_orig = UserService().default_user_settings
yield
SystemService().system_settings = sys_orig
UserService().default_user_settings = user_orig
# ---------------------------------------------------------------------------
# load_config — discovery + cascade
# ---------------------------------------------------------------------------
def test_load_config_returns_empty_when_no_layers_present(tmp_path, monkeypatch):
"""No pyproject, no openbb.toml, no explicit config → empty dict.
Callers fall through to disk JSONs / model defaults.
"""
from openbb_core.app.config.loader import load_config
monkeypatch.chdir(tmp_path)
monkeypatch.delenv("OPENBB_CONFIG", raising=False)
# Point user-global lookup at an empty dir so the user's actual
# config file doesn't leak into the test result.
monkeypatch.setattr(
"openbb_core.app.config.loader.USER_OPENBB_DIR",
tmp_path / "no_user_dir",
)
out = load_config()
assert out == {}
def test_load_config_reads_project_local_openbb_toml(tmp_path, monkeypatch):
"""``./openbb.toml`` is loaded and its contents end up in the merged dict."""
from openbb_core.app.config.loader import load_config
project_toml = tmp_path / "openbb.toml"
project_toml.write_text(
"""
[system]
debug_mode = true
headless = false
[user.preferences]
output_type = "dataframe"
"""
)
monkeypatch.chdir(tmp_path)
monkeypatch.setattr(
"openbb_core.app.config.loader.USER_OPENBB_DIR",
tmp_path / "no_user_dir",
)
out = load_config()
assert out["system"]["debug_mode"] is True
assert out["user"]["preferences"]["output_type"] == "dataframe"
def test_load_config_reads_pyproject_tool_openbb_table(tmp_path, monkeypatch):
"""``[tool.openbb]`` in pyproject.toml is the lowest-priority layer."""
from openbb_core.app.config.loader import load_config
pyproject = tmp_path / "pyproject.toml"
pyproject.write_text(
"""
[tool.openbb.system]
debug_mode = false
[tool.openbb.user.preferences]
output_type = "polars"
"""
)
monkeypatch.chdir(tmp_path)
monkeypatch.setattr(
"openbb_core.app.config.loader.USER_OPENBB_DIR",
tmp_path / "no_user_dir",
)
out = load_config()
assert out["system"]["debug_mode"] is False
assert out["user"]["preferences"]["output_type"] == "polars"
def test_load_config_explicit_path_wins_over_project_local(tmp_path, monkeypatch):
"""``--config /explicit.toml`` overrides project-local ``./openbb.toml``."""
from openbb_core.app.config.loader import load_config
project_toml = tmp_path / "openbb.toml"
project_toml.write_text("[system]\ndebug_mode = false\n")
explicit_toml = tmp_path / "explicit.toml"
explicit_toml.write_text("[system]\ndebug_mode = true\n")
monkeypatch.chdir(tmp_path)
monkeypatch.setattr(
"openbb_core.app.config.loader.USER_OPENBB_DIR",
tmp_path / "no_user_dir",
)
out = load_config(explicit_path=str(explicit_toml))
assert out["system"]["debug_mode"] is True
def test_load_config_env_var_acts_as_explicit_path(tmp_path, monkeypatch):
"""``$OPENBB_CONFIG`` is used when ``explicit_path`` is omitted."""
from openbb_core.app.config.loader import load_config
explicit_toml = tmp_path / "from_env.toml"
explicit_toml.write_text("[system]\nheadless = true\n")
monkeypatch.setenv("OPENBB_CONFIG", str(explicit_toml))
monkeypatch.chdir(tmp_path)
monkeypatch.setattr(
"openbb_core.app.config.loader.USER_OPENBB_DIR",
tmp_path / "no_user_dir",
)
out = load_config()
assert out["system"]["headless"] is True
def test_load_config_user_global_layer_picked_up(tmp_path, monkeypatch):
"""``~/.openbb_platform/openbb.toml`` (user-global) is layered between
pyproject and project-local."""
from openbb_core.app.config.loader import load_config
user_dir = tmp_path / "user_openbb"
user_dir.mkdir()
(user_dir / "openbb.toml").write_text(
'[user.preferences]\noutput_type = "polars"\n'
)
elsewhere = tmp_path / "elsewhere"
elsewhere.mkdir()
monkeypatch.chdir(elsewhere)
monkeypatch.setattr(
"openbb_core.app.config.loader.USER_OPENBB_DIR",
user_dir,
)
monkeypatch.delenv("OPENBB_CONFIG", raising=False)
out = load_config()
assert out["user"]["preferences"]["output_type"] == "polars"
def test_load_config_cascade_order_project_overrides_user_global(tmp_path, monkeypatch):
"""Project-local ``openbb.toml`` beats the user-global file."""
from openbb_core.app.config.loader import load_config
user_dir = tmp_path / "user_openbb"
user_dir.mkdir()
(user_dir / "openbb.toml").write_text(
'[user.preferences]\noutput_type = "polars"\n'
)
project_dir = tmp_path / "project"
project_dir.mkdir()
(project_dir / "openbb.toml").write_text(
'[user.preferences]\noutput_type = "dataframe"\n'
)
monkeypatch.chdir(project_dir)
monkeypatch.setattr(
"openbb_core.app.config.loader.USER_OPENBB_DIR",
user_dir,
)
monkeypatch.delenv("OPENBB_CONFIG", raising=False)
out = load_config()
# Project-local won.
assert out["user"]["preferences"]["output_type"] == "dataframe"
def test_load_config_normalizes_kebab_case_top_level_keys(tmp_path, monkeypatch):
"""``debug-mode`` (kebab) becomes ``debug_mode`` so it can be promoted
onto SystemSettings.``debug_mode``."""
from openbb_core.app.config.loader import load_config
(tmp_path / "openbb.toml").write_text("debug-mode = true\n")
monkeypatch.chdir(tmp_path)
monkeypatch.setattr(
"openbb_core.app.config.loader.USER_OPENBB_DIR",
tmp_path / "no_user_dir",
)
out = load_config()
assert out["debug_mode"] is True
def test_load_config_handles_malformed_toml_silently(tmp_path, monkeypatch):
"""A broken TOML file in the cascade is treated as an empty layer
instead of crashing the whole load."""
from openbb_core.app.config.loader import load_config
(tmp_path / "openbb.toml").write_text("not [valid toml at all")
monkeypatch.chdir(tmp_path)
monkeypatch.setattr(
"openbb_core.app.config.loader.USER_OPENBB_DIR",
tmp_path / "no_user_dir",
)
out = load_config()
# No crash; the malformed file produced an empty layer.
assert out == {}
def test_load_config_explicit_path_missing_is_no_op(tmp_path, monkeypatch):
"""Pointing at a non-existent file via --config doesn't crash; the
layer is just skipped silently."""
from openbb_core.app.config.loader import load_config
monkeypatch.chdir(tmp_path)
monkeypatch.setattr(
"openbb_core.app.config.loader.USER_OPENBB_DIR",
tmp_path / "no_user_dir",
)
out = load_config(explicit_path=str(tmp_path / "missing.toml"))
assert out == {}
# ---------------------------------------------------------------------------
# pyproject [tool.openbb] discovery edge cases
# ---------------------------------------------------------------------------
def test_load_config_pyproject_without_tool_openbb_section(tmp_path, monkeypatch):
"""A pyproject without ``[tool.openbb]`` contributes nothing — and
doesn't crash on the missing key path."""
from openbb_core.app.config.loader import load_config
(tmp_path / "pyproject.toml").write_text(
'[tool.poetry]\nname = "x"\nversion = "0.0.0"\n'
)
monkeypatch.chdir(tmp_path)
monkeypatch.setattr(
"openbb_core.app.config.loader.USER_OPENBB_DIR",
tmp_path / "no_user_dir",
)
monkeypatch.delenv("OPENBB_CONFIG", raising=False)
out = load_config()
assert out == {}
def test_load_config_pyproject_section_value_not_a_dict(tmp_path, monkeypatch):
"""If ``tool.openbb`` exists but isn't a table (e.g. a string), the
loader treats it as missing."""
from openbb_core.app.config.loader import load_config
(tmp_path / "pyproject.toml").write_text('[tool]\nopenbb = "string"\n')
monkeypatch.chdir(tmp_path)
monkeypatch.setattr(
"openbb_core.app.config.loader.USER_OPENBB_DIR",
tmp_path / "no_user_dir",
)
out = load_config()
assert out == {}
# ---------------------------------------------------------------------------
# apply_config_to_services
# ---------------------------------------------------------------------------
def test_apply_config_to_services_pushes_system_overrides(reset_singletons):
"""``[system]`` section lands on ``SystemService.system_settings``."""
from openbb_core.app.config.loader import apply_config_to_services
from openbb_core.app.service.system_service import SystemService
applied = apply_config_to_services({"system": {"debug_mode": True}})
assert SystemService().system_settings.debug_mode is True
assert "debug_mode" in applied["system"]
def test_apply_config_to_services_pushes_user_overrides(reset_singletons):
"""``[user.preferences]`` lands on ``UserService.default_user_settings``."""
from openbb_core.app.config.loader import apply_config_to_services
from openbb_core.app.service.user_service import UserService
apply_config_to_services({"user": {"preferences": {"output_type": "polars"}}})
assert UserService().default_user_settings.preferences.output_type == "polars"
def test_apply_config_to_services_promotes_top_level_into_system(reset_singletons):
"""Top-level ``debug_mode`` cascades onto ``[system].debug_mode``."""
from openbb_core.app.config.loader import apply_config_to_services
from openbb_core.app.service.system_service import SystemService
apply_config_to_services({"debug_mode": True})
assert SystemService().system_settings.debug_mode is True
def test_apply_config_to_services_top_level_wins_over_section(reset_singletons):
"""When both top-level ``debug_mode`` and ``[system].debug_mode``
are set, the top-level explicit form wins."""
from openbb_core.app.config.loader import apply_config_to_services
from openbb_core.app.service.system_service import SystemService
apply_config_to_services({"debug_mode": True, "system": {"debug_mode": False}})
assert SystemService().system_settings.debug_mode is True
def test_apply_config_to_services_empty_input_no_op(reset_singletons):
"""Empty / None config → no-op, no exceptions."""
from openbb_core.app.config.loader import apply_config_to_services
assert apply_config_to_services(None) == {"system": [], "user": []}
assert apply_config_to_services({}) == {"system": [], "user": []}
def test_apply_config_to_services_returns_field_names_applied(reset_singletons):
"""The return value lists which fields each service updated."""
from openbb_core.app.config.loader import apply_config_to_services
out = apply_config_to_services(
{
"system": {"debug_mode": True, "headless": False},
"user": {"preferences": {"output_type": "polars"}},
}
)
assert sorted(out["system"]) == ["debug_mode", "headless"]
assert out["user"] == ["preferences"]
# ---------------------------------------------------------------------------
# apply_settings_to_env — ``OPENBB_*`` env-var seeding
# ---------------------------------------------------------------------------
def test_apply_settings_to_env_seeds_promoted_flags(monkeypatch):
"""Promoted top-level keys (e.g. ``debug_mode``) seed ``OPENBB_DEBUG_MODE``."""
from openbb_core.app.config.loader import apply_settings_to_env
monkeypatch.delenv("OPENBB_DEBUG_MODE", raising=False)
applied = apply_settings_to_env({"debug_mode": True})
assert "OPENBB_DEBUG_MODE" in applied
assert os.environ["OPENBB_DEBUG_MODE"] == "True"
monkeypatch.delenv("OPENBB_DEBUG_MODE", raising=False)
def test_apply_settings_to_env_existing_env_var_wins(monkeypatch):
"""``setdefault`` semantics: pre-existing real shell exports stay."""
from openbb_core.app.config.loader import apply_settings_to_env
monkeypatch.setenv("OPENBB_DEBUG_MODE", "False")
applied = apply_settings_to_env({"debug_mode": True})
assert "OPENBB_DEBUG_MODE" not in applied
assert os.environ["OPENBB_DEBUG_MODE"] == "False"
def test_apply_settings_to_env_skips_non_promoted_section_keys(monkeypatch):
"""``[system]`` keys outside the promotion allowlist are NOT seeded
as env vars — too many fields to enumerate via env, only the
blessed shortcuts get the ``OPENBB_*`` treatment."""
from openbb_core.app.config.loader import apply_settings_to_env
monkeypatch.delenv("OPENBB_LOGGING_VERBOSITY", raising=False)
applied = apply_settings_to_env(
{"system": {"logging_verbosity": 30, "debug_mode": True}}
)
assert "OPENBB_LOGGING_VERBOSITY" not in applied
monkeypatch.delenv("OPENBB_DEBUG_MODE", raising=False)
def test_apply_settings_to_env_empty_or_no_system_section_no_op():
"""Neither None nor a config without ``[system]`` produces env vars."""
from openbb_core.app.config.loader import apply_settings_to_env
assert apply_settings_to_env(None) == []
assert apply_settings_to_env({}) == []
assert apply_settings_to_env({"user": {}}) == []
def test_apply_settings_to_env_handles_non_dict_system_section():
"""Defensive: malformed ``system = "x"`` (not a table) → no crash."""
from openbb_core.app.config.loader import apply_settings_to_env
assert apply_settings_to_env({"system": "not-a-table"}) == []
# ---------------------------------------------------------------------------
# load_env_files — .env loading
# ---------------------------------------------------------------------------
def test_load_env_files_loads_user_global_dotenv(tmp_path, monkeypatch):
"""``~/.openbb_platform/.env`` gets pulled into ``os.environ`` via setdefault."""
from openbb_core.app.config.loader import load_env_files
user_dir = tmp_path / "user_openbb"
user_dir.mkdir()
(user_dir / ".env").write_text("FROM_USER_GLOBAL=value-1\n")
monkeypatch.setattr(
"openbb_core.app.config.loader.USER_OPENBB_DIR",
user_dir,
)
monkeypatch.delenv("OPENBB_ENV_FILE", raising=False)
monkeypatch.delenv("FROM_USER_GLOBAL", raising=False)
loaded = load_env_files()
assert any(p.name == ".env" for p in loaded)
assert os.environ.get("FROM_USER_GLOBAL") == "value-1"
monkeypatch.delenv("FROM_USER_GLOBAL", raising=False)
def test_load_env_files_explicit_path_overrides_user_global(tmp_path, monkeypatch):
"""``$OPENBB_ENV_FILE`` (or explicit_path arg) is layered after the
user-global file; pre-existing exports still win."""
from openbb_core.app.config.loader import load_env_files
user_dir = tmp_path / "user_openbb"
user_dir.mkdir()
(user_dir / ".env").write_text("LAYER=user-global\n")
explicit = tmp_path / "explicit.env"
explicit.write_text("LAYER=explicit\nEXTRA=yes\n")
monkeypatch.setattr(
"openbb_core.app.config.loader.USER_OPENBB_DIR",
user_dir,
)
monkeypatch.delenv("LAYER", raising=False)
monkeypatch.delenv("EXTRA", raising=False)
load_env_files(explicit_path=str(explicit))
# User-global ran first; setdefault means it set ``LAYER=user-global``.
# Explicit re-application via setdefault then can't overwrite — that's
# the documented "real shell exports win" semantic; explicit-over-
# user is actually only a layering of *new* keys.
assert os.environ["LAYER"] == "user-global"
assert os.environ["EXTRA"] == "yes"
monkeypatch.delenv("LAYER", raising=False)
monkeypatch.delenv("EXTRA", raising=False)
def test_load_env_files_no_files_returns_empty_list(tmp_path, monkeypatch):
"""Discovery with no files returns an empty list, no errors."""
from openbb_core.app.config.loader import load_env_files
monkeypatch.setattr(
"openbb_core.app.config.loader.USER_OPENBB_DIR",
tmp_path / "no_dir",
)
monkeypatch.delenv("OPENBB_ENV_FILE", raising=False)
assert load_env_files() == []
def test_load_env_files_skips_none_values(tmp_path, monkeypatch):
"""Dotenv parses bare ``KEY`` (no ``=value``) as ``KEY=None``;
those entries skip the env injection rather than blowing up."""
from openbb_core.app.config.loader import load_env_files
user_dir = tmp_path / "user_openbb"
user_dir.mkdir()
(user_dir / ".env").write_text("BARE_KEY\nREAL=v\n")
monkeypatch.setattr(
"openbb_core.app.config.loader.USER_OPENBB_DIR",
user_dir,
)
monkeypatch.delenv("BARE_KEY", raising=False)
monkeypatch.delenv("REAL", raising=False)
load_env_files()
assert "BARE_KEY" not in os.environ
assert os.environ["REAL"] == "v"
monkeypatch.delenv("REAL", raising=False)
# ---------------------------------------------------------------------------
# load_layered_config — the bundled bootstrap entry point
# ---------------------------------------------------------------------------
def test_load_layered_config_runs_full_pipeline(
tmp_path, monkeypatch, reset_singletons
):
"""End-to-end: env files load, TOML cascade resolves, services updated,
and ``OPENBB_*`` env vars seeded for promoted flags."""
from openbb_core.app.config.loader import load_layered_config
from openbb_core.app.service.system_service import SystemService
(tmp_path / "openbb.toml").write_text(
"""
debug-mode = true
[user.preferences]
output_type = "polars"
"""
)
monkeypatch.chdir(tmp_path)
monkeypatch.setattr(
"openbb_core.app.config.loader.USER_OPENBB_DIR",
tmp_path / "no_user_dir",
)
monkeypatch.delenv("OPENBB_DEBUG_MODE", raising=False)
monkeypatch.delenv("OPENBB_CONFIG", raising=False)
monkeypatch.delenv("OPENBB_ENV_FILE", raising=False)
config = load_layered_config()
assert config["debug_mode"] is True
assert SystemService().system_settings.debug_mode is True
assert os.environ["OPENBB_DEBUG_MODE"] == "True"
monkeypatch.delenv("OPENBB_DEBUG_MODE", raising=False)
def test_load_layered_config_can_skip_service_application(
tmp_path, monkeypatch, reset_singletons
):
"""``apply_to_services=False`` returns the config without mutating
the singleton services — useful for tooling that just wants to
inspect the resolved layer stack."""
from openbb_core.app.config.loader import load_layered_config
from openbb_core.app.service.system_service import SystemService
(tmp_path / "openbb.toml").write_text("[system]\ndebug_mode = true\n")
monkeypatch.chdir(tmp_path)
monkeypatch.setattr(
"openbb_core.app.config.loader.USER_OPENBB_DIR",
tmp_path / "no_user_dir",
)
monkeypatch.delenv("OPENBB_CONFIG", raising=False)
monkeypatch.delenv("OPENBB_DEBUG_MODE", raising=False)
original_debug = SystemService().system_settings.debug_mode
config = load_layered_config(apply_to_services=False, apply_to_env=False)
assert config["system"]["debug_mode"] is True
# Service untouched.
assert SystemService().system_settings.debug_mode is original_debug
# ---------------------------------------------------------------------------
# render_config_template
# ---------------------------------------------------------------------------
def test_render_config_template_includes_top_level_promotions():
"""Every promoted top-level key shows up in the rendered template."""
from openbb_core.app.config.loader import (
_TOP_LEVEL_SYSTEM_PROMOTIONS,
render_config_template,
)
out = render_config_template()
for promoted in _TOP_LEVEL_SYSTEM_PROMOTIONS:
kebab = promoted.replace("_", "-")
assert kebab in out
def test_render_config_template_includes_system_and_user_sections():
"""Both ``[system]`` and ``[user]`` headers appear in the rendered output."""
from openbb_core.app.config.loader import render_config_template
out = render_config_template()
assert "[system]" in out
assert "[user]" in out
def test_render_config_template_uncomments_active_values():
"""Values present in ``active`` get emitted uncommented; everything else
stays commented out as a fall-through default."""
from openbb_core.app.config.loader import render_config_template
out = render_config_template({"debug_mode": True})
# Active value uncommented.
assert "\ndebug-mode = true\n" in out
# Non-active key still commented.
assert "# headless = false\n" in out
def test_render_config_template_handles_list_values():
"""List-typed active values render as TOML arrays."""
from openbb_core.app.config.loader import render_config_template
# ``logging_handlers`` is a list-typed SystemSettings field.
out = render_config_template({"system": {"logging_handlers": ["stdout", "file"]}})
assert '["stdout", "file"]' in out
# ---------------------------------------------------------------------------
# _toml_quote — value rendering
# ---------------------------------------------------------------------------
def test_user_global_toml_returns_none_when_no_match(tmp_path, monkeypatch):
"""When the user-global dir exists but contains neither
``openbb.toml`` nor ``.openbb.toml``, the discovery returns
``None`` — exercises the fall-through return at the end of
``_user_global_toml``."""
from openbb_core.app.config.loader import _user_global_toml
user_dir = tmp_path / "user_dir"
user_dir.mkdir()
(user_dir / "something_else.toml").write_text("")
monkeypatch.setattr(
"openbb_core.app.config.loader.USER_OPENBB_DIR",
user_dir,
)
assert _user_global_toml() is None
def test_promote_top_level_keys_returns_empty_for_falsy_config():
"""Empty / None config short-circuits to ``{}``."""
from openbb_core.app.config.loader import _promote_top_level_keys
assert _promote_top_level_keys({}) == {}
assert _promote_top_level_keys(None) == {} # type: ignore[arg-type]
def test_apply_settings_to_env_renders_non_bool_values_as_strings(monkeypatch):
"""A non-bool promoted value (e.g. integer or string) renders via
``str(value)`` — exercises the else-branch of the bool check."""
from openbb_core.app.config.loader import apply_settings_to_env
monkeypatch.delenv("OPENBB_LOGGING_SUPPRESS", raising=False)
# ``logging_suppress`` is in the promotion list. Pass a non-bool
# to hit the ``str(value)`` branch.
apply_settings_to_env({"logging_suppress": "verbose"})
assert os.environ.get("OPENBB_LOGGING_SUPPRESS") == "verbose"
monkeypatch.delenv("OPENBB_LOGGING_SUPPRESS", raising=False)
def test_render_config_template_emits_nested_table_header_for_dicts():
"""An active dict value (e.g. ``[system.api_settings]``) gets
emitted as a sub-table header in the template."""
from openbb_core.app.config.loader import render_config_template
out = render_config_template({"system": {"api_settings": {"prefix": "/api/v1"}}})
assert "[system.api_settings]" in out
def test_render_config_template_emits_active_scalar_uncommented():
"""A non-dict / non-list / non-None active value (e.g. a string
or int field on SystemSettings) gets rendered uncommented with the
quoted active value — the catch-all ``else`` branch of the
settings-section renderer."""
from openbb_core.app.config.loader import render_config_template
out = render_config_template(
{"system": {"logging_verbosity": 30, "logging_frequency": "M"}}
)
# Both scalar fields uncommented with their active values.
assert "\nlogging_verbosity = 30\n" in out
assert '\nlogging_frequency = "M"\n' in out
def test_apply_settings_to_env_non_dict_system_section_returns_empty(monkeypatch):
"""Defensive: a config with ``system`` shaped as a string passes
through ``_promote_top_level_keys`` (which strips it) and then the
`not isinstance(system_section, dict)` short-circuit isn't reached —
confirm the behavior end-to-end is still no env vars set."""
from openbb_core.app.config.loader import apply_settings_to_env
# Even when the input has a malformed system entry, no env vars
# leak out. Real shell exports continue to be authoritative.
monkeypatch.delenv("OPENBB_DEBUG_MODE", raising=False)
out = apply_settings_to_env({"system": "garbage"})
assert out == []
assert "OPENBB_DEBUG_MODE" not in os.environ
def test_toml_quote_handles_each_supported_type():
"""Strings get escaped, bools lowercased, numbers preserved, lists
wrapped, fallthroughs go to string."""
from openbb_core.app.config.loader import _toml_quote
assert _toml_quote("hello") == '"hello"'
assert _toml_quote('quote"in"middle') == '"quote\\"in\\"middle"'
assert _toml_quote("back\\slash") == '"back\\\\slash"'
assert _toml_quote(True) == "true"
assert _toml_quote(False) == "false"
assert _toml_quote(42) == "42"
assert _toml_quote(3.14) == "3.14"
assert _toml_quote([1, 2]) == "[1, 2]"
assert _toml_quote(["a", "b"]) == '["a", "b"]'
# Other types fall through to ``str(value)`` wrapped in quotes.
# ``PurePosixPath`` is used (not ``Path``) because ``str(Path("/x"))``
# is platform-dependent — Windows normalizes to ``\x``, POSIX keeps
# ``/x``. The intent here is to verify the fallthrough branch, not
# platform-specific path rendering.
assert _toml_quote(PurePosixPath("/x")) == '"/x"'

View File

@@ -0,0 +1,164 @@
"""Tests for AgGrid SSRM request/response models."""
import pytest
from pydantic import BaseModel, ValidationError
from openbb_core.app.model.aggrid import (
AgFilterModel,
AgGridRowsRequest,
AgGridRowsResponse,
AgGroupingModel,
AgSortColumn,
)
def test_ag_sort_column_round_trips():
"""``colId``/``sort`` are mandatory; both directions accepted."""
asc = AgSortColumn(colId="symbol", sort="asc")
desc = AgSortColumn(colId="market_cap", sort="desc")
assert asc.colId == "symbol"
assert desc.sort == "desc"
def test_ag_sort_column_rejects_invalid_direction():
"""``sort`` is constrained to ``Literal["asc", "desc"]``."""
with pytest.raises(ValidationError):
AgSortColumn(colId="x", sort="ascending") # type: ignore[arg-type]
def test_ag_sort_column_rejects_extra_keys():
"""``extra="forbid"`` so the wire format stays tight."""
with pytest.raises(ValidationError):
AgSortColumn(colId="x", sort="asc", whoops=True) # type: ignore[call-arg]
def test_ag_filter_model_allows_extras_for_operator_vocab():
"""ag-grid keeps adding fields like ``filterTo``/``dateFrom``;
``extra="allow"`` keeps the model forward-compatible.
"""
f = AgFilterModel.model_validate(
{
"filterType": "number",
"type": "greaterThan",
"filter": 100,
"dateFrom": "2024-01-01", # extra; should pass through
}
)
assert f.filterType == "number"
assert f.filter == 100
# ``extra="allow"`` exposes unknown keys via model_dump
assert f.model_dump()["dateFrom"] == "2024-01-01"
def test_ag_filter_model_filter_type_constrained():
"""``filterType`` is a Literal — anything outside the set is
rejected even though we allow extra top-level keys.
"""
with pytest.raises(ValidationError):
AgFilterModel(filterType="weird") # type: ignore[arg-type]
def test_ag_grouping_model_defaults():
"""All collection fields default to empty; ``pivotMode`` defaults
to False — that's the "no grouping yet" baseline.
"""
g = AgGroupingModel()
assert g.rowGroupCols == []
assert g.valueCols == []
assert g.pivotCols == []
assert g.pivotMode is False
assert g.groupKeys == []
def test_ag_grouping_model_rejects_extras():
"""Tight schema — typos shouldn't silently pass."""
with pytest.raises(ValidationError):
AgGroupingModel(typo=True) # type: ignore[call-arg]
def test_ag_grid_rows_request_minimum_payload():
"""Just ``startRow``/``endRow`` is enough — the rest defaults."""
r = AgGridRowsRequest(startRow=0, endRow=100)
assert r.startRow == 0
assert r.endRow == 100
assert r.sortModel == []
assert r.filterModel == {}
assert r.grouping is None
def test_ag_grid_rows_request_full_payload():
"""A populated payload deserializes nested models correctly."""
r = AgGridRowsRequest.model_validate(
{
"startRow": 0,
"endRow": 50,
"sortModel": [{"colId": "date", "sort": "desc"}],
"filterModel": {
"symbol": {
"filterType": "text",
"type": "contains",
"filter": "AAPL",
}
},
"grouping": {"groupKeys": ["technology"]},
}
)
assert isinstance(r.sortModel[0], AgSortColumn)
assert r.sortModel[0].colId == "date"
assert isinstance(r.filterModel["symbol"], AgFilterModel)
assert r.filterModel["symbol"].filter == "AAPL"
assert r.grouping is not None
assert r.grouping.groupKeys == ["technology"]
def test_ag_grid_rows_request_negative_indices_rejected():
"""``startRow``/``endRow`` are non-negative — ag-grid never sends
negatives, and a negative would crash slicing on the server.
"""
with pytest.raises(ValidationError):
AgGridRowsRequest(startRow=-1, endRow=10)
def test_ag_grid_rows_request_extras_allowed_for_subclass_pattern():
"""``extra="allow"`` lets a subclass add provider-specific filters
without a schema-level merge step. Direct extras also pass through.
"""
r = AgGridRowsRequest.model_validate(
{"startRow": 0, "endRow": 10, "providerToken": "abc"}
)
assert r.model_dump()["providerToken"] == "abc"
class _FooRow(BaseModel):
symbol: str
price: float
def test_ag_grid_rows_response_generic_parametrization():
"""Generic ``Row`` resolves to the parametrized model and validates
the items list against it.
"""
resp = AgGridRowsResponse[_FooRow](
rowData=[_FooRow(symbol="AAPL", price=150.0)],
rowCount=1,
)
assert resp.rowCount == 1
assert resp.rowData[0].symbol == "AAPL"
def test_ag_grid_rows_response_rejects_extras():
"""``extra="forbid"`` keeps the wire format predictable for the
ag-grid client.
"""
with pytest.raises(ValidationError):
AgGridRowsResponse[_FooRow](
rowData=[],
rowCount=0,
extra="nope", # type: ignore[call-arg]
)
def test_ag_grid_rows_response_negative_count_rejected():
"""``rowCount`` must be non-negative."""
with pytest.raises(ValidationError):
AgGridRowsResponse[_FooRow](rowData=[], rowCount=-1)

View File

@@ -72,7 +72,16 @@ def test_get_field_type_strips_openbb_module_path_inside_generic_container():
the ``list[...]`` container but drops the dotted module path inside —
the docstring shows ``list[MyData]``, not ``list[openbb_x.y.MyData]`` and
not just the truncated tail ``MyData]``.
Uses ``typing.Union[...]`` rather than ``X | Y`` so the test lands in
the Union branch on every supported Python. On 3.10-3.13,
``get_origin(X | Y)`` returns ``types.UnionType`` (not ``typing.Union``),
so ``X | Y`` syntax falls into the else branch and bypasses the
container-aware strip we're trying to cover here. 3.14 unified
``types.UnionType`` with ``typing.Union``, masking the gap on dev
machines but leaving Linux/Windows CI without coverage.
"""
from typing import Union
class MyData:
"""Stand-in provider Data class — its fully-qualified name lands
@@ -85,7 +94,9 @@ def test_get_field_type_strips_openbb_module_path_inside_generic_container():
MyData.__module__ = "openbb_someprovider.models.my_data"
MyData.__qualname__ = "MyData"
out = DocstringGenerator.get_field_type(list[MyData] | None, is_required=False)
out = DocstringGenerator.get_field_type(
Union[list[MyData], None], is_required=False
)
# Container preserved, dotted prefix stripped.
assert "list[MyData]" in out
# The prefix must not leak through.
@@ -97,6 +108,35 @@ def test_get_field_type_strips_openbb_module_path_inside_generic_container():
assert out.endswith("| None")
def test_get_field_type_pep604_union_syntax_takes_union_branch():
"""``list[X] | None`` (PEP 604) must hit the same Union branch as
``Union[list[X], None]``. On Python 3.10-3.13 the two forms produce
different ``get_origin`` results — ``types.UnionType`` vs
``typing.Union`` — so the generator's check accepts both. Regression
guard for the platform-dependent coverage gap.
"""
from typing import Union
class MyData:
"""Stand-in provider Data class."""
MyData.__module__ = "openbb_someprovider.models.my_data"
MyData.__qualname__ = "MyData"
# Both syntaxes must produce the same docstring output.
union_form = DocstringGenerator.get_field_type(
Union[list[MyData], None], # noqa: UP007 — explicit form is the point
is_required=False,
)
pep604_form = DocstringGenerator.get_field_type(
list[MyData] | None, is_required=False
)
assert union_form == pep604_form
# And the container-aware strip fired for both.
assert "list[MyData]" in pep604_form
assert "openbb_someprovider" not in pep604_form
def test_get_field_type_strips_openbb_module_path_for_non_generic():
"""Non-generic ``openbb_*`` types take the simple ``rsplit('.', 1)``
fallback (no ``[`` in the string) — exercises the else-branch

4
openbb_platform/core/uv.lock generated vendored
View File

@@ -1954,7 +1954,7 @@ wheels = [
[[package]]
name = "openbb-core"
version = "1.7.0"
version = "2.0.0"
source = { editable = "." }
dependencies = [
{ name = "aiohttp" },
@@ -1966,6 +1966,7 @@ dependencies = [
{ name = "python-multipart" },
{ name = "requests" },
{ name = "ruff" },
{ name = "tomli", marker = "python_full_version < '3.11'" },
{ name = "uuid7" },
{ name = "uvicorn" },
{ name = "websockets" },
@@ -1994,6 +1995,7 @@ requires-dist = [
{ name = "python-multipart", specifier = ">=0.0.26,<0.0.27" },
{ name = "requests", specifier = ">=2.33.0,<3.0.0" },
{ name = "ruff", specifier = ">=0.15,<0.16" },
{ name = "tomli", marker = "python_full_version < '3.11'", specifier = ">=2.0.0" },
{ name = "uuid7", specifier = ">=0.1.0,<0.2.0" },
{ name = "uvicorn", specifier = ">=0.40.0,<0.41.0" },
{ name = "websockets", specifier = ">=15.0" },