Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@ dev = [

[project]
name = "src-py-lib"
version = "0.1.6"
version = "0.1.7"
description = "Reusable libraries for Sourcegraph projects"
readme = "README.md"
requires-python = ">=3.11"
Expand Down
4 changes: 4 additions & 0 deletions src/src_py_lib/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -52,6 +52,8 @@
Config,
ConfigError,
config_field,
config_field_names,
config_help_formatter,
config_snapshot,
)
from src_py_lib.utils.config import (
Expand Down Expand Up @@ -150,6 +152,8 @@ def _script_name() -> str:
"TraceContext",
"aliased_batched_query",
"config_field",
"config_field_names",
"config_help_formatter",
"config_snapshot",
"configure_logging",
"critical",
Expand Down
8 changes: 5 additions & 3 deletions src/src_py_lib/clients/sourcegraph.py
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,6 @@
traceparent_header,
)

DEFAULT_SOURCEGRAPH_ENDPOINT = "https://sourcegraph.com"
SOURCEGRAPH_EXTERNAL_SERVICE_NODE_TYPE: Final[str] = "ExternalService"
SOURCEGRAPH_REPOSITORY_NODE_TYPE: Final[str] = "Repository"
REQUEST_TRACE_HEADER: Final[str] = "X-Sourcegraph-Request-Trace"
Expand Down Expand Up @@ -158,18 +157,21 @@ class SourcegraphClientConfig(Config):
"""Config fields needed to build a Sourcegraph API client."""

src_endpoint: str = config_field(
default=DEFAULT_SOURCEGRAPH_ENDPOINT,
default="",
env_var="SRC_ENDPOINT",
cli_flag="--src-endpoint",
metavar="URL",
help=f"Sourcegraph instance URL (default: {DEFAULT_SOURCEGRAPH_ENDPOINT})",
help="Sourcegraph instance URL",
help_group="Sourcegraph",
required=True,
)
src_access_token: str = config_field(
default="",
env_var="SRC_ACCESS_TOKEN",
cli_flag="--src-access-token",
metavar="TOKEN",
help="Sourcegraph access token, or op:// secret reference",
help_group="Sourcegraph",
secret=True,
required=True,
)
Expand Down
139 changes: 121 additions & 18 deletions src/src_py_lib/utils/config.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@
from dataclasses import dataclass, replace
from pathlib import Path
from types import UnionType
from typing import Any, Final, Literal, TypeVar, Union, cast, get_args, get_origin
from typing import Any, Final, Literal, TypeAlias, TypeVar, Union, cast, get_args, get_origin

from dotenv import dotenv_values
from pydantic import BaseModel, ConfigDict, Field, ValidationError
Expand All @@ -33,6 +33,7 @@
CONFIG_HELP_MIN_POSITION: Final[int] = 24
CONFIG_HELP_MAX_POSITION_LIMIT: Final[int] = 48
CONFIG_HELP_PADDING: Final[int] = 4
DEFAULT_CONFIG_HELP_GROUP: Final[str] = "Config"
_CONFIG_OPTION_KEY: Final[str] = "src_py_lib_config_option"
_MISSING: Final[object] = object()

Expand Down Expand Up @@ -67,6 +68,7 @@ class ConfigOption:
cli_const: object | None = None
metavar: str | None = None
help: str = ""
help_group: str = DEFAULT_CONFIG_HELP_GROUP
secret: bool = False
required: bool = False

Expand All @@ -78,6 +80,7 @@ class Config(BaseModel):


ConfigType = TypeVar("ConfigType", bound=Config)
ConfigFieldSource: TypeAlias = str | type[Config]


def config_field(
Expand All @@ -91,6 +94,7 @@ def config_field(
cli_const: object | None = None,
metavar: str | None = None,
help: str = "",
help_group: str = DEFAULT_CONFIG_HELP_GROUP,
secret: bool = False,
required: bool = False,
gt: int | float | None = None,
Expand All @@ -110,6 +114,7 @@ def config_field(
cli_const=cli_const,
metavar=metavar,
help=help,
help_group=help_group,
secret=secret,
required=required,
)
Expand Down Expand Up @@ -140,6 +145,38 @@ def config_options(config_cls: type[Config]) -> tuple[ConfigOption, ...]:
return tuple(options)


def config_field_names(*sources: ConfigFieldSource) -> tuple[str, ...]:
"""Return Config field names from Config classes and explicit field names.

Use this to define reusable CLI argument sets from Config mixins while
keeping the field metadata defined once on the Config classes.
"""
names: list[str] = []
for source in sources:
if isinstance(source, str):
names.append(source)
continue
names.extend(option.field_name for option in config_options(source))
return tuple(dict.fromkeys(names))


def config_help_formatter(
config_cls: type[Config],
*,
include_env_file: bool = True,
include_fields: Iterable[str] | None = None,
exclude_fields: Iterable[str] = (),
) -> type[argparse.HelpFormatter]:
"""Return a help formatter aligned for the selected Config fields."""
max_help_position = _config_help_max_position(
config_cls,
include_env_file=include_env_file,
include_fields=include_fields,
exclude_fields=exclude_fields,
)
return _config_help_formatter(max_help_position)


def load_config_env_file(path: Path | None) -> dict[str, str]:
"""Load key/value pairs from a `.env` file.

Expand Down Expand Up @@ -192,22 +229,19 @@ def add_config_arguments(
config_cls: type[Config],
*,
include_env_file: bool = True,
include_fields: Iterable[str] | None = None,
exclude_fields: Iterable[str] = (),
) -> None:
"""Add Config CLI flags to an argparse parser."""
group = parser.add_argument_group(
"Config",
"These options override matching environment variables and .env values",
)
if include_env_file:
group.add_argument(
"--env-file",
dest="env_file",
default=None,
metavar="PATH",
help="Read Config .env values from PATH (default: .env)",
)
groups: dict[str, Any] = {}

for option in config_options(config_cls):
def argument_group(title: str) -> Any:
if title not in groups:
groups[title] = parser.add_argument_group(title)
return groups[title]

for option in _selected_config_options(config_cls, include_fields, exclude_fields):
group = argument_group(option.help_group or DEFAULT_CONFIG_HELP_GROUP)
field_info = config_cls.model_fields[option.field_name]
argument_kwargs: dict[str, Any] = {
"dest": option.field_name,
Expand All @@ -227,6 +261,15 @@ def add_config_arguments(
argument_kwargs["action"] = option.cli_action
group.add_argument(option.cli_flag, *option.cli_aliases, **argument_kwargs)

if include_env_file:
argument_group(DEFAULT_CONFIG_HELP_GROUP).add_argument(
"--env-file",
dest="env_file",
default=None,
metavar="PATH",
help="Read Config .env values from PATH (default: .env)",
)


def config_parse_args(
config_cls: type[ConfigType],
Expand All @@ -235,19 +278,32 @@ def config_parse_args(
argv: Sequence[str] | None = None,
description: str | None = None,
include_env_file: bool = True,
include_fields: Iterable[str] | None = None,
exclude_fields: Iterable[str] = (),
env: Mapping[str, str] | None = None,
base_dir: Path | None = None,
resolve_op_refs: bool = True,
op_client: OnePasswordClient | None = None,
require: Iterable[str] = (),
) -> ConfigType:
"""Parse Config CLI flags and return a validated Config model."""
max_help_position = _config_help_max_position(config_cls, include_env_file=include_env_file)
formatter_class = config_help_formatter(
config_cls,
include_env_file=include_env_file,
include_fields=include_fields,
exclude_fields=exclude_fields,
)
argument_parser = parser or argparse.ArgumentParser(
description=description,
formatter_class=_config_help_formatter(max_help_position),
formatter_class=formatter_class,
)
add_config_arguments(
argument_parser,
config_cls,
include_env_file=include_env_file,
include_fields=include_fields,
exclude_fields=exclude_fields,
)
add_config_arguments(argument_parser, config_cls, include_env_file=include_env_file)
args = argument_parser.parse_args(argv)
try:
return load_config_from_args(
Expand Down Expand Up @@ -277,12 +333,14 @@ def _config_help_max_position(
config_cls: type[Config],
*,
include_env_file: bool,
include_fields: Iterable[str] | None = None,
exclude_fields: Iterable[str] = (),
) -> int:
"""Return help-column width based on this Config's CLI arguments."""
invocation_lengths = [len("--env-file PATH")] if include_env_file else []
invocation_lengths.extend(
_config_option_invocation_length(config_cls, option)
for option in config_options(config_cls)
for option in _selected_config_options(config_cls, include_fields, exclude_fields)
)
longest_invocation = max(invocation_lengths, default=0)
return min(
Expand All @@ -291,6 +349,48 @@ def _config_help_max_position(
)


def _selected_config_options(
config_cls: type[Config],
include_fields: Iterable[str] | None,
exclude_fields: Iterable[str],
) -> tuple[ConfigOption, ...]:
"""Return options selected by field or env-var names.

When include_fields is set, its order controls the returned option order.
Without include_fields, Config model field order is preserved.
"""
options = config_options(config_cls)
excluded = _selected_config_field_names(options, exclude_fields) or set()
if include_fields is None:
return tuple(option for option in options if not _option_is_selected(option, excluded))

options_by_field_name = {option.field_name: option for option in options}
return tuple(
options_by_field_name[field_name]
for field_name in _selected_config_field_names_in_order(options, include_fields)
if field_name not in excluded
)


def _selected_config_field_names_in_order(
options: tuple[ConfigOption, ...],
selected: Iterable[str],
) -> tuple[str, ...]:
"""Return selected field names in caller order after validating names."""
names = (_option_by_name(options, name).field_name for name in selected)
return tuple(dict.fromkeys(names))


def _selected_config_field_names(
options: tuple[ConfigOption, ...],
selected: Iterable[str] | None,
) -> set[str] | None:
"""Return selected field names after validating field or env-var names."""
if selected is None:
return None
return {_option_by_name(options, name).field_name for name in selected}


def _config_option_invocation_length(config_cls: type[Config], option: ConfigOption) -> int:
"""Return argparse-style option invocation length for help alignment."""
field_info = config_cls.model_fields[option.field_name]
Expand Down Expand Up @@ -452,6 +552,7 @@ def _config_option_payload(option: ConfigOption) -> dict[str, object]:
"cli_const": option.cli_const,
"metavar": option.metavar,
"help": option.help,
"help_group": option.help_group,
"secret": option.secret,
"required": option.required,
}
Expand All @@ -467,6 +568,7 @@ def _config_option_from_payload(payload: Mapping[str, object]) -> ConfigOption |
cli_nargs = payload.get("cli_nargs")
metavar = payload.get("metavar")
help_text = payload.get("help")
help_group = payload.get("help_group")
return ConfigOption(
field_name="",
env_var=env_var,
Expand All @@ -477,6 +579,7 @@ def _config_option_from_payload(payload: Mapping[str, object]) -> ConfigOption |
cli_const=payload.get("cli_const"),
metavar=metavar if isinstance(metavar, str) else None,
help=help_text if isinstance(help_text, str) else "",
help_group=help_group if isinstance(help_group, str) else DEFAULT_CONFIG_HELP_GROUP,
secret=payload.get("secret") is True,
required=payload.get("required") is True,
)
Expand Down
4 changes: 4 additions & 0 deletions src/src_py_lib/utils/logging.py
Original file line number Diff line number Diff line change
Expand Up @@ -100,6 +100,7 @@ class LoggingConfig(Config):
cli_flag="--src-log-level",
metavar="LEVEL",
help="Log level (default: INFO)",
help_group="Logging",
)
verbose: bool = config_field(
default=False,
Expand All @@ -108,6 +109,7 @@ class LoggingConfig(Config):
cli_aliases=("-v",),
cli_action="store_true",
help="Alias for --src-log-level DEBUG",
help_group="Logging",
)
quiet: bool = config_field(
default=False,
Expand All @@ -116,6 +118,7 @@ class LoggingConfig(Config):
cli_aliases=("-q",),
cli_action="store_true",
help="Alias for --src-log-level WARNING",
help_group="Logging",
)
silent: bool = config_field(
default=False,
Expand All @@ -124,6 +127,7 @@ class LoggingConfig(Config):
cli_aliases=("-s",),
cli_action="store_true",
help="Alias for --src-log-level ERROR",
help_group="Logging",
)

@model_validator(mode="after")
Expand Down
2 changes: 2 additions & 0 deletions tests/test_import.py
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,8 @@ def test_root_public_api_exports_common_entrypoints(self) -> None:
self.assertIsNotNone(src_py_lib.SourcegraphClient)
self.assertIsNotNone(src_py_lib.SourcegraphClientConfig)
self.assertIsNotNone(src_py_lib.config_field)
self.assertIsNotNone(src_py_lib.config_field_names)
self.assertIsNotNone(src_py_lib.config_help_formatter)
self.assertIsNotNone(src_py_lib.gh_cli_token)
self.assertIsNotNone(src_py_lib.gcloud_adc_access_token)
self.assertIsNotNone(src_py_lib.info)
Expand Down
Loading