Skip to content

Instantly share code, notes, and snippets.

@danizen
Created December 18, 2024 18:26
Show Gist options
  • Save danizen/61de10c4c601fa36c97a65546b785248 to your computer and use it in GitHub Desktop.
Save danizen/61de10c4c601fa36c97a65546b785248 to your computer and use it in GitHub Desktop.
from __future__ import annotations
from collections.abc import Collection
from pathlib import Path
from typing import Optional, Type, Any, Literal, Union, Annotated
from typing_extensions import Self
from pydantic import ValidationError
from pydantic.fields import AliasChoices, Field
from pydantic_settings import (
BaseSettings,
PydanticBaseSettingsSource,
CliSettingsSource,
)
class MyConfigSettings(BaseSettings):
job_name: str = Field(alias=AliasChoices('JOB_NAME', 'job_name'))
job_run_id: str = Field(alias=AliasChoices('JOB_RUN_ID', 'job_run_id'))
workflow_name: Optional[str] = Field(default=None, alias=AliasChoices('WORKFLOW_NAME', 'workflow_name'))
workflow_run_id: Optional[str] = Field(default=None, alias=AliasChoices('WORKFLOW_RUN_ID', 'workflow_run_id'))
@classmethod
def settings_customise_sources(
cls,
settings_cls: Type[BaseSettings],
init_settings: PydanticBaseSettingsSource,
env_settings: PydanticBaseSettingsSource,
dotenv_settings: PydanticBaseSettingsSource,
file_secret_settings: PydanticBaseSettingsSource,
) -> tuple[PydanticBaseSettingsSource, ...]:
"""
This method overrides the default settings sources so that initially, only
init_settings are considered.
"""
return (init_settings,)
@classmethod
def load(cls, args: Optional[Collection[str]] = None, **kwargs: Any) -> Self:
class WrapperConfig(cls):
@classmethod
def settings_customise_sources(
cls,
settings_cls: Type[BaseSettings],
init_settings: PydanticBaseSettingsSource,
env_settings: PydanticBaseSettingsSource,
dotenv_settings: PydanticBaseSettingsSource,
file_secret_settings: PydanticBaseSettingsSource,
) -> tuple[PydanticBaseSettingsSource, ...]:
"""
Restore the sources you want
"""
return (
CliSettingsSource(cli_parse_args=True, cli_ignore_unknown_args=True),
init_settings,
env_settings,
dotenv_settings,
file_secret_settings,
)
return WrapperConfig(**kwargs, _cli_parse_args=args)
# Following test code is now OK
def test_something_depending_on_schema():
my_config = MyConfigSettings(
job_name='test_job',
job_run_id='test_job_inst_1'
)
something_that_needs_config(my_config)
from __future__ import annotations
from typing import Optional
from pydantic.fields import AliasChoices, Field
from pydantic_settings import BaseSettings
class MyConfigSettings(BaseSettings, cli_parse_args=True):
job_name: str = Field(alias=AliasChoices('JOB_NAME', 'job_name'))
job_run_id: str = Field(alias=AliasChoices('JOB_RUN_ID', 'job_run_id'))
workflow_name: Optional[str] = Field(default=None, alias=AliasChoices('WORKFLOW_NAME', 'workflow_name'))
workflow_run_id: Optional[str] = Field(default=None, alias=AliasChoices('WORKFLOW_RUN_ID', 'workflow_run_id'))
# Following test code violates principle of least surprise, need to do DI that only Python can support
def test_something_depending_on_schema():
my_config = MyConfigSettings(
job_name='test_job',
job_run_id='test_job_inst_1'
)
something_that_needs_config(my_config)
# If someone runs pytest -k depending_on_schema, there is an error because the "-k" is not supported.
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment