Skip to content

prefect.settings.base

PrefectBaseSettings

Bases: BaseSettings

Source code in src/prefect/settings/base.py 
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
class PrefectBaseSettings(BaseSettings):
    @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, ...]:
        """
        Define an order for Prefect settings sources.

        The order of the returned callables decides the priority of inputs; first item is the highest priority.

        See https://docs.pydantic.dev/latest/concepts/pydantic_settings/#customise-settings-sources
        """
        env_filter = set()
        for field in settings_cls.model_fields.values():
            if field.validation_alias is not None and isinstance(
                field.validation_alias, AliasChoices
            ):
                for alias in field.validation_alias.choices:
                    if isinstance(alias, AliasPath) and len(alias.path) > 0:
                        env_filter.add(alias.path[0])
        return (
            init_settings,
            EnvFilterSettingsSource(
                settings_cls,
                case_sensitive=cls.model_config.get("case_sensitive"),
                env_prefix=cls.model_config.get("env_prefix"),
                env_nested_delimiter=cls.model_config.get("env_nested_delimiter"),
                env_ignore_empty=cls.model_config.get("env_ignore_empty"),
                env_parse_none_str=cls.model_config.get("env_parse_none_str"),
                env_parse_enums=cls.model_config.get("env_parse_enums"),
                env_filter=list(env_filter),
            ),
            dotenv_settings,
            file_secret_settings,
            PrefectTomlConfigSettingsSource(settings_cls),
            PyprojectTomlConfigSettingsSource(settings_cls),
            ProfileSettingsTomlLoader(settings_cls),
        )

    def to_environment_variables(
        self,
        exclude_unset: bool = False,
        include_secrets: bool = True,
    ) -> Dict[str, str]:
        """Convert the settings object to a dictionary of environment variables."""

        env: Dict[str, Any] = self.model_dump(
            exclude_unset=exclude_unset,
            mode="json",
            context={"include_secrets": include_secrets},
        )
        env_variables = {}
        for key in self.model_fields.keys():
            if isinstance(child_settings := getattr(self, key), PrefectBaseSettings):
                child_env = child_settings.to_environment_variables(
                    exclude_unset=exclude_unset,
                    include_secrets=include_secrets,
                )
                env_variables.update(child_env)
            elif (value := env.get(key)) is not None:
                env_variables[
                    f"{self.model_config.get('env_prefix')}{key.upper()}"
                ] = str(value)
        return env_variables

    @model_serializer(
        mode="wrap", when_used="always"
    )  # TODO: reconsider `when_used` default for more control
    def ser_model(
        self, handler: SerializerFunctionWrapHandler, info: SerializationInfo
    ) -> Any:
        jsonable_self = handler(self)
        # iterate over fields to ensure child models that have been updated are also included
        for key in self.model_fields.keys():
            if info.exclude and key in info.exclude:
                continue
            if info.include and key not in info.include:
                continue

            child_include = None
            child_exclude = None
            if info.include and key in info.include and isinstance(info.include, dict):
                child_include = info.include[key]

            if isinstance(child_settings := getattr(self, key), PrefectBaseSettings):
                child_jsonable = child_settings.model_dump(
                    mode=info.mode,
                    include=child_include,  # type: ignore
                    exclude=child_exclude,
                    exclude_unset=info.exclude_unset,
                    context=info.context,
                    by_alias=info.by_alias,
                    exclude_none=info.exclude_none,
                    exclude_defaults=info.exclude_defaults,
                    serialize_as_any=info.serialize_as_any,
                )
                if child_jsonable:
                    jsonable_self[key] = child_jsonable
        if info.context and info.context.get("include_secrets") is True:
            jsonable_self.update(
                {
                    field_name: visit_collection(
                        expr=getattr(self, field_name),
                        visit_fn=partial(handle_secret_render, context=info.context),
                        return_data=True,
                    )
                    for field_name in set(jsonable_self.keys())  # type: ignore
                }
            )

        return jsonable_self

settings_customise_sources(settings_cls, init_settings, env_settings, dotenv_settings, file_secret_settings) classmethod

Define an order for Prefect settings sources.

The order of the returned callables decides the priority of inputs; first item is the highest priority.

See https://docs.pydantic.dev/latest/concepts/pydantic_settings/#customise-settings-sources

Source code in src/prefect/settings/base.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
@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, ...]:
    """
    Define an order for Prefect settings sources.

    The order of the returned callables decides the priority of inputs; first item is the highest priority.

    See https://docs.pydantic.dev/latest/concepts/pydantic_settings/#customise-settings-sources
    """
    env_filter = set()
    for field in settings_cls.model_fields.values():
        if field.validation_alias is not None and isinstance(
            field.validation_alias, AliasChoices
        ):
            for alias in field.validation_alias.choices:
                if isinstance(alias, AliasPath) and len(alias.path) > 0:
                    env_filter.add(alias.path[0])
    return (
        init_settings,
        EnvFilterSettingsSource(
            settings_cls,
            case_sensitive=cls.model_config.get("case_sensitive"),
            env_prefix=cls.model_config.get("env_prefix"),
            env_nested_delimiter=cls.model_config.get("env_nested_delimiter"),
            env_ignore_empty=cls.model_config.get("env_ignore_empty"),
            env_parse_none_str=cls.model_config.get("env_parse_none_str"),
            env_parse_enums=cls.model_config.get("env_parse_enums"),
            env_filter=list(env_filter),
        ),
        dotenv_settings,
        file_secret_settings,
        PrefectTomlConfigSettingsSource(settings_cls),
        PyprojectTomlConfigSettingsSource(settings_cls),
        ProfileSettingsTomlLoader(settings_cls),
    )

to_environment_variables(exclude_unset=False, include_secrets=True)

Convert the settings object to a dictionary of environment variables.

Source code in src/prefect/settings/base.py 
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
def to_environment_variables(
    self,
    exclude_unset: bool = False,
    include_secrets: bool = True,
) -> Dict[str, str]:
    """Convert the settings object to a dictionary of environment variables."""

    env: Dict[str, Any] = self.model_dump(
        exclude_unset=exclude_unset,
        mode="json",
        context={"include_secrets": include_secrets},
    )
    env_variables = {}
    for key in self.model_fields.keys():
        if isinstance(child_settings := getattr(self, key), PrefectBaseSettings):
            child_env = child_settings.to_environment_variables(
                exclude_unset=exclude_unset,
                include_secrets=include_secrets,
            )
            env_variables.update(child_env)
        elif (value := env.get(key)) is not None:
            env_variables[
                f"{self.model_config.get('env_prefix')}{key.upper()}"
            ] = str(value)
    return env_variables

PrefectSettingsConfigDict

Bases: SettingsConfigDict

Configuration for the behavior of Prefect settings models.

Source code in src/prefect/settings/base.py 
146
147
148
149
150
151
152
153
154
155
156
157
158
class PrefectSettingsConfigDict(SettingsConfigDict, total=False):
    """
    Configuration for the behavior of Prefect settings models.
    """

    prefect_toml_table_header: tuple[str, ...]
    """
    Header of the TOML table within a prefect.toml file to use when filling variables.
    This is supplied as a `tuple[str, ...]` instead of a `str` to accommodate for headers
    containing a `.`.

    To use the root table, exclude this config setting or provide an empty tuple.
    """

prefect_toml_table_header: tuple[str, ...] instance-attribute

Header of the TOML table within a prefect.toml file to use when filling variables. This is supplied as a tuple[str, ...] instead of a str to accommodate for headers containing a ..

To use the root table, exclude this config setting or provide an empty tuple.