prefect.settings

Prefect settings are defined using BaseSettings from pydantic_settings. BaseSettings can load setting values from system environment variables and each additionally specified env_file.

The recommended user-facing way to access Prefect settings at this time is to import specific setting objects directly, like from prefect.settings import PREFECT_API_URL; print(PREFECT_API_URL.value()).

Importantly, we replace the callback mechanism for updating settings with an "after" model_validator that updates dependent settings. After https://github.com/pydantic/pydantic/issues/9789 is resolved, we will be able to define context-aware defaults for settings, at which point we will not need to use the "after" model_validator.

Profile

Bases: BaseModel

A user profile containing settings.

Source code in src/prefect/settings/profiles.py

class Profile(BaseModel):
    """A user profile containing settings."""

    model_config = ConfigDict(extra="ignore", arbitrary_types_allowed=True)

    name: str
    settings: Annotated[Dict[Setting, Any], BeforeValidator(_cast_settings)] = Field(
        default_factory=dict
    )
    source: Optional[Path] = None

    def to_environment_variables(self) -> Dict[str, str]:
        """Convert the profile settings to a dictionary of environment variables."""
        return {
            setting.name: str(value)
            for setting, value in self.settings.items()
            if value is not None
        }

    def validate_settings(self):
        errors: List[Tuple[Setting, ValidationError]] = []
        for setting, value in self.settings.items():
            try:
                model_fields = Settings.model_fields
                annotation = None
                for section in setting.accessor.split("."):
                    annotation = model_fields[section].annotation
                    if inspect.isclass(annotation) and issubclass(
                        annotation, BaseSettings
                    ):
                        model_fields = annotation.model_fields

                TypeAdapter(annotation).validate_python(value)
            except ValidationError as e:
                errors.append((setting, e))
        if errors:
            raise ProfileSettingsValidationError(errors)

to_environment_variables()

Convert the profile settings to a dictionary of environment variables.

Source code in src/prefect/settings/profiles.py

def to_environment_variables(self) -> Dict[str, str]:
    """Convert the profile settings to a dictionary of environment variables."""
    return {
        setting.name: str(value)
        for setting, value in self.settings.items()
        if value is not None
    }

ProfilesCollection

" A utility class for working with a collection of profiles.

Profiles in the collection must have unique names.

The collection may store the name of the active profile.

Source code in src/prefect/settings/profiles.py

class ProfilesCollection:
    """ "
    A utility class for working with a collection of profiles.

    Profiles in the collection must have unique names.

    The collection may store the name of the active profile.
    """

    def __init__(
        self, profiles: Iterable[Profile], active: Optional[str] = None
    ) -> None:
        self.profiles_by_name = {profile.name: profile for profile in profiles}
        self.active_name = active

    @property
    def names(self) -> Set[str]:
        """
        Return a set of profile names in this collection.
        """
        return set(self.profiles_by_name.keys())

    @property
    def active_profile(self) -> Optional[Profile]:
        """
        Retrieve the active profile in this collection.
        """
        if self.active_name is None:
            return None
        return self[self.active_name]

    def set_active(self, name: Optional[str], check: bool = True):
        """
        Set the active profile name in the collection.

        A null value may be passed to indicate that this collection does not determine
        the active profile.
        """
        if check and name is not None and name not in self.names:
            raise ValueError(f"Unknown profile name {name!r}.")
        self.active_name = name

    def update_profile(
        self,
        name: str,
        settings: Dict[Setting, Any],
        source: Optional[Path] = None,
    ) -> Profile:
        """
        Add a profile to the collection or update the existing on if the name is already
        present in this collection.

        If updating an existing profile, the settings will be merged. Settings can
        be dropped from the existing profile by setting them to `None` in the new
        profile.

        Returns the new profile object.
        """
        existing = self.profiles_by_name.get(name)

        # Convert the input to a `Profile` to cast settings to the correct type
        profile = Profile(name=name, settings=settings, source=source)

        if existing:
            new_settings = {**existing.settings, **profile.settings}

            # Drop null keys to restore to default
            for key, value in tuple(new_settings.items()):
                if value is None:
                    new_settings.pop(key)

            new_profile = Profile(
                name=profile.name,
                settings=new_settings,
                source=source or profile.source,
            )
        else:
            new_profile = profile

        self.profiles_by_name[new_profile.name] = new_profile

        return new_profile

    def add_profile(self, profile: Profile) -> None:
        """
        Add a profile to the collection.

        If the profile name already exists, an exception will be raised.
        """
        if profile.name in self.profiles_by_name:
            raise ValueError(
                f"Profile name {profile.name!r} already exists in collection."
            )

        self.profiles_by_name[profile.name] = profile

    def remove_profile(self, name: str) -> None:
        """
        Remove a profile from the collection.
        """
        self.profiles_by_name.pop(name)

    def without_profile_source(self, path: Optional[Path]) -> "ProfilesCollection":
        """
        Remove profiles that were loaded from a given path.

        Returns a new collection.
        """
        return ProfilesCollection(
            [
                profile
                for profile in self.profiles_by_name.values()
                if profile.source != path
            ],
            active=self.active_name,
        )

    def to_dict(self):
        """
        Convert to a dictionary suitable for writing to disk.
        """
        return {
            "active": self.active_name,
            "profiles": {
                profile.name: profile.to_environment_variables()
                for profile in self.profiles_by_name.values()
            },
        }

    def __getitem__(self, name: str) -> Profile:
        return self.profiles_by_name[name]

    def __iter__(self):
        return self.profiles_by_name.__iter__()

    def items(self):
        return self.profiles_by_name.items()

    def __eq__(self, __o: object) -> bool:
        if not isinstance(__o, ProfilesCollection):
            return False

        return (
            self.profiles_by_name == __o.profiles_by_name
            and self.active_name == __o.active_name
        )

    def __repr__(self) -> str:
        return (
            f"ProfilesCollection(profiles={list(self.profiles_by_name.values())!r},"
            f" active={self.active_name!r})>"
        )

active_profile: Optional[Profile] property

Retrieve the active profile in this collection.

names: Set[str] property

Return a set of profile names in this collection.

add_profile(profile)

Add a profile to the collection.

If the profile name already exists, an exception will be raised.

Source code in src/prefect/settings/profiles.py

def add_profile(self, profile: Profile) -> None:
    """
    Add a profile to the collection.

    If the profile name already exists, an exception will be raised.
    """
    if profile.name in self.profiles_by_name:
        raise ValueError(
            f"Profile name {profile.name!r} already exists in collection."
        )

    self.profiles_by_name[profile.name] = profile

remove_profile(name)

Remove a profile from the collection.

Source code in src/prefect/settings/profiles.py

def remove_profile(self, name: str) -> None:
    """
    Remove a profile from the collection.
    """
    self.profiles_by_name.pop(name)

set_active(name, check=True)

Set the active profile name in the collection.

A null value may be passed to indicate that this collection does not determine the active profile.

Source code in src/prefect/settings/profiles.py

def set_active(self, name: Optional[str], check: bool = True):
    """
    Set the active profile name in the collection.

    A null value may be passed to indicate that this collection does not determine
    the active profile.
    """
    if check and name is not None and name not in self.names:
        raise ValueError(f"Unknown profile name {name!r}.")
    self.active_name = name

to_dict()

Convert to a dictionary suitable for writing to disk.

Source code in src/prefect/settings/profiles.py

def to_dict(self):
    """
    Convert to a dictionary suitable for writing to disk.
    """
    return {
        "active": self.active_name,
        "profiles": {
            profile.name: profile.to_environment_variables()
            for profile in self.profiles_by_name.values()
        },
    }

update_profile(name, settings, source=None)

Add a profile to the collection or update the existing on if the name is already present in this collection.

If updating an existing profile, the settings will be merged. Settings can be dropped from the existing profile by setting them to None in the new profile.

Returns the new profile object.

Source code in src/prefect/settings/profiles.py

def update_profile(
    self,
    name: str,
    settings: Dict[Setting, Any],
    source: Optional[Path] = None,
) -> Profile:
    """
    Add a profile to the collection or update the existing on if the name is already
    present in this collection.

    If updating an existing profile, the settings will be merged. Settings can
    be dropped from the existing profile by setting them to `None` in the new
    profile.

    Returns the new profile object.
    """
    existing = self.profiles_by_name.get(name)

    # Convert the input to a `Profile` to cast settings to the correct type
    profile = Profile(name=name, settings=settings, source=source)

    if existing:
        new_settings = {**existing.settings, **profile.settings}

        # Drop null keys to restore to default
        for key, value in tuple(new_settings.items()):
            if value is None:
                new_settings.pop(key)

        new_profile = Profile(
            name=profile.name,
            settings=new_settings,
            source=source or profile.source,
        )
    else:
        new_profile = profile

    self.profiles_by_name[new_profile.name] = new_profile

    return new_profile

without_profile_source(path)

Remove profiles that were loaded from a given path.

Returns a new collection.

Source code in src/prefect/settings/profiles.py

def without_profile_source(self, path: Optional[Path]) -> "ProfilesCollection":
    """
    Remove profiles that were loaded from a given path.

    Returns a new collection.
    """
    return ProfilesCollection(
        [
            profile
            for profile in self.profiles_by_name.values()
            if profile.source != path
        ],
        active=self.active_name,
    )

Setting

Mimics the old Setting object for compatibility with existing code.

Source code in src/prefect/settings/legacy.py

class Setting:
    """Mimics the old Setting object for compatibility with existing code."""

    def __init__(
        self, name: str, default: Any, type_: Any, accessor: Optional[str] = None
    ):
        self._name = name
        self._default = default
        self._type = type_
        if accessor is None:
            self.accessor = _env_var_to_accessor(name)
        else:
            self.accessor = accessor

    @property
    def name(self):
        return self._name

    @property
    def is_secret(self):
        if self._type in _SECRET_TYPES:
            return True
        for secret_type in _SECRET_TYPES:
            if secret_type in get_args(self._type):
                return True
        return False

    def default(self):
        return self._default

    def value(self: Self) -> Any:
        if (
            self.name == "PREFECT_TEST_SETTING"
            or self.name == "PREFECT_TESTING_TEST_SETTING"
        ):
            if (
                "PREFECT_TEST_MODE" in os.environ
                or "PREFECT_TESTING_TEST_MODE" in os.environ
            ):
                return get_current_settings().testing.test_setting
            else:
                return None

        return self.value_from(get_current_settings())

    def value_from(self: Self, settings: "Settings") -> Any:
        path = self.accessor.split(".")
        current_value = settings
        for key in path:
            current_value = getattr(current_value, key, None)
        if isinstance(current_value, _SECRET_TYPES):
            return current_value.get_secret_value()  # type: ignore
        return current_value

    def __bool__(self) -> bool:
        return bool(self.value())

    def __str__(self) -> str:
        return str(self.value())

    def __repr__(self) -> str:
        return f"<{self.name}: {self._type!r}>"

    def __eq__(self, __o: object) -> bool:
        return __o.__eq__(self.value())

    def __hash__(self) -> int:
        return hash((type(self), self.name))

Settings

Bases: PrefectBaseSettings

Settings for Prefect using Pydantic settings.

See https://docs.pydantic.dev/latest/concepts/pydantic_settings

Source code in src/prefect/settings/models/root.py

class Settings(PrefectBaseSettings):
    """
    Settings for Prefect using Pydantic settings.

    See https://docs.pydantic.dev/latest/concepts/pydantic_settings
    """

    model_config = _build_settings_config()

    home: Annotated[Path, BeforeValidator(lambda x: Path(x).expanduser())] = Field(
        default=Path("~") / ".prefect",
        description="The path to the Prefect home directory. Defaults to ~/.prefect",
    )

    profiles_path: Optional[Path] = Field(
        default=None,
        description="The path to a profiles configuration file.",
    )

    debug_mode: bool = Field(
        default=False,
        description="If True, enables debug mode which may provide additional logging and debugging features.",
    )

    api: APISettings = Field(
        default_factory=APISettings,
        description="Settings for interacting with the Prefect API",
    )

    cli: CLISettings = Field(
        default_factory=CLISettings,
        description="Settings for controlling CLI behavior",
    )

    client: ClientSettings = Field(
        default_factory=ClientSettings,
        description="Settings for for controlling API client behavior",
    )

    cloud: CloudSettings = Field(
        default_factory=CloudSettings,
        description="Settings for interacting with Prefect Cloud",
    )

    deployments: DeploymentsSettings = Field(
        default_factory=DeploymentsSettings,
        description="Settings for configuring deployments defaults",
    )

    experiments: ExperimentsSettings = Field(
        default_factory=ExperimentsSettings,
        description="Settings for controlling experimental features",
    )

    flows: FlowsSettings = Field(
        default_factory=FlowsSettings,
        description="Settings for controlling flow behavior",
    )

    internal: InternalSettings = Field(
        default_factory=InternalSettings,
        description="Settings for internal Prefect machinery",
    )

    logging: LoggingSettings = Field(
        default_factory=LoggingSettings,
        description="Settings for controlling logging behavior",
    )

    results: ResultsSettings = Field(
        default_factory=ResultsSettings,
        description="Settings for controlling result storage behavior",
    )

    runner: RunnerSettings = Field(
        default_factory=RunnerSettings,
        description="Settings for controlling runner behavior",
    )

    server: ServerSettings = Field(
        default_factory=ServerSettings,
        description="Settings for controlling server behavior",
    )

    tasks: TasksSettings = Field(
        default_factory=TasksSettings,
        description="Settings for controlling task behavior",
    )

    testing: TestingSettings = Field(
        default_factory=TestingSettings,
        description="Settings used during testing",
    )

    worker: WorkerSettings = Field(
        default_factory=WorkerSettings,
        description="Settings for controlling worker behavior",
    )

    ui_url: Optional[str] = Field(
        default=None,
        description="The URL of the Prefect UI. If not set, the client will attempt to infer it.",
    )

    silence_api_url_misconfiguration: bool = Field(
        default=False,
        description="""
        If `True`, disable the warning when a user accidentally misconfigure its `PREFECT_API_URL`
        Sometimes when a user manually set `PREFECT_API_URL` to a custom url,reverse-proxy for example,
        we would like to silence this warning so we will set it to `FALSE`.
        """,
    )

    ###########################################################################
    # allow deprecated access to PREFECT_SOME_SETTING_NAME

    def __getattribute__(self, name: str) -> Any:
        from prefect.settings.legacy import _env_var_to_accessor

        if name.startswith("PREFECT_"):
            accessor = _env_var_to_accessor(name)
            warnings.warn(
                f"Accessing `Settings().{name}` is deprecated. Use `Settings().{accessor}` instead.",
                DeprecationWarning,
                stacklevel=2,
            )
            path = accessor.split(".")
            value = super().__getattribute__(path[0])
            for key in path[1:]:
                value = getattr(value, key)
            return value
        return super().__getattribute__(name)

    ###########################################################################

    @model_validator(mode="after")
    def post_hoc_settings(self) -> Self:
        """refactor on resolution of https://github.com/pydantic/pydantic/issues/9789

        we should not be modifying __pydantic_fields_set__ directly, but until we can
        define dependencies between defaults in a first-class way, we need clean up
        post-hoc default assignments to keep set/unset fields correct after instantiation.
        """
        if self.ui_url is None:
            self.ui_url = _default_ui_url(self)
            self.__pydantic_fields_set__.remove("ui_url")
        if self.server.ui.api_url is None:
            if self.api.url:
                self.server.ui.api_url = self.api.url
                self.server.ui.__pydantic_fields_set__.remove("api_url")
            else:
                self.server.ui.api_url = (
                    f"http://{self.server.api.host}:{self.server.api.port}/api"
                )
                self.server.ui.__pydantic_fields_set__.remove("api_url")
        if self.profiles_path is None or "PREFECT_HOME" in str(self.profiles_path):
            self.profiles_path = Path(f"{self.home}/profiles.toml")
            self.__pydantic_fields_set__.remove("profiles_path")
        if self.results.local_storage_path is None:
            self.results.local_storage_path = Path(f"{self.home}/storage")
            self.results.__pydantic_fields_set__.remove("local_storage_path")
        if self.server.memo_store_path is None:
            self.server.memo_store_path = Path(f"{self.home}/memo_store.toml")
            self.server.__pydantic_fields_set__.remove("memo_store_path")
        if self.debug_mode or self.testing.test_mode:
            self.logging.level = "DEBUG"
            self.internal.logging_level = "DEBUG"
            self.logging.__pydantic_fields_set__.remove("level")
            self.internal.__pydantic_fields_set__.remove("logging_level")

        if self.logging.config_path is None:
            self.logging.config_path = Path(f"{self.home}/logging.yml")
            self.logging.__pydantic_fields_set__.remove("config_path")
        # Set default database connection URL if not provided
        if self.server.database.connection_url is None:
            self.server.database.connection_url = _default_database_connection_url(self)
            self.server.database.__pydantic_fields_set__.remove("connection_url")
        db_url = (
            self.server.database.connection_url.get_secret_value()
            if isinstance(self.server.database.connection_url, SecretStr)
            else self.server.database.connection_url
        )
        if (
            "PREFECT_API_DATABASE_PASSWORD" in db_url
            or "PREFECT_SERVER_DATABASE_PASSWORD" in db_url
        ):
            if self.server.database.password is None:
                raise ValueError(
                    "database password is None - please set PREFECT_SERVER_DATABASE_PASSWORD"
                )
            db_url = db_url.replace(
                "${PREFECT_API_DATABASE_PASSWORD}",
                self.server.database.password.get_secret_value()
                if self.server.database.password
                else "",
            )
            db_url = db_url.replace(
                "${PREFECT_SERVER_DATABASE_PASSWORD}",
                self.server.database.password.get_secret_value()
                if self.server.database.password
                else "",
            )
            self.server.database.connection_url = SecretStr(db_url)
            self.server.database.__pydantic_fields_set__.remove("connection_url")
        return self

    @model_validator(mode="after")
    def emit_warnings(self) -> Self:
        """More post-hoc validation of settings, including warnings for misconfigurations."""
        if not self.silence_api_url_misconfiguration:
            _warn_on_misconfigured_api_url(self)
        return self

    ##########################################################################
    # Settings methods

    def copy_with_update(
        self: Self,
        updates: Optional[Mapping["Setting", Any]] = None,
        set_defaults: Optional[Mapping["Setting", Any]] = None,
        restore_defaults: Optional[Iterable["Setting"]] = None,
    ) -> Self:
        """
        Create a new Settings object with validation.

        Arguments:
            updates: A mapping of settings to new values. Existing values for the
                given settings will be overridden.
            set_defaults: A mapping of settings to new default values. Existing values for
                the given settings will only be overridden if they were not set.
            restore_defaults: An iterable of settings to restore to their default values.

        Returns:
            A new Settings object.
        """
        # To restore defaults, we need to resolve the setting path and then
        # set the default value on the new settings object. When restoring
        # defaults, all settings sources will be ignored.
        restore_defaults_obj = {}
        for r in restore_defaults or []:
            path = r.accessor.split(".")
            model = self
            for key in path[:-1]:
                model = model.model_fields[key].annotation
                assert model is not None, f"Invalid setting path: {r.accessor}"

            model_field = model.model_fields[path[-1]]
            assert model is not None, f"Invalid setting path: {r.accessor}"
            if hasattr(model_field, "default"):
                default = model_field.default
            elif (
                hasattr(model_field, "default_factory") and model_field.default_factory
            ):
                default = model_field.default_factory()
            else:
                raise ValueError(f"No default value for setting: {r.accessor}")
            set_in_dict(
                restore_defaults_obj,
                r.accessor,
                default,
            )
        updates = updates or {}
        set_defaults = set_defaults or {}

        set_defaults_obj = {}
        for setting, value in set_defaults.items():
            set_in_dict(set_defaults_obj, setting.accessor, value)

        updates_obj = {}
        for setting, value in updates.items():
            set_in_dict(updates_obj, setting.accessor, value)

        new_settings = self.__class__.model_validate(
            deep_merge_dicts(
                set_defaults_obj,
                self.model_dump(exclude_unset=True),
                restore_defaults_obj,
                updates_obj,
            )
        )
        return new_settings

    def hash_key(self) -> str:
        """
        Return a hash key for the settings object.  This is needed since some
        settings may be unhashable, like lists.
        """
        env_variables = self.to_environment_variables()
        return str(hash(tuple((key, value) for key, value in env_variables.items())))

copy_with_update(updates=None, set_defaults=None, restore_defaults=None)

Create a new Settings object with validation.

Parameters:

Name	Type	Description	Default
updates	Optional[Mapping[Setting, Any]]	A mapping of settings to new values. Existing values for the given settings will be overridden.	None
set_defaults	Optional[Mapping[Setting, Any]]	A mapping of settings to new default values. Existing values for the given settings will only be overridden if they were not set.	None
restore_defaults	Optional[Iterable[Setting]]	An iterable of settings to restore to their default values.	None

Returns:

Type	Description
Self	A new Settings object.

Source code in src/prefect/settings/models/root.py

def copy_with_update(
    self: Self,
    updates: Optional[Mapping["Setting", Any]] = None,
    set_defaults: Optional[Mapping["Setting", Any]] = None,
    restore_defaults: Optional[Iterable["Setting"]] = None,
) -> Self:
    """
    Create a new Settings object with validation.

    Arguments:
        updates: A mapping of settings to new values. Existing values for the
            given settings will be overridden.
        set_defaults: A mapping of settings to new default values. Existing values for
            the given settings will only be overridden if they were not set.
        restore_defaults: An iterable of settings to restore to their default values.

    Returns:
        A new Settings object.
    """
    # To restore defaults, we need to resolve the setting path and then
    # set the default value on the new settings object. When restoring
    # defaults, all settings sources will be ignored.
    restore_defaults_obj = {}
    for r in restore_defaults or []:
        path = r.accessor.split(".")
        model = self
        for key in path[:-1]:
            model = model.model_fields[key].annotation
            assert model is not None, f"Invalid setting path: {r.accessor}"

        model_field = model.model_fields[path[-1]]
        assert model is not None, f"Invalid setting path: {r.accessor}"
        if hasattr(model_field, "default"):
            default = model_field.default
        elif (
            hasattr(model_field, "default_factory") and model_field.default_factory
        ):
            default = model_field.default_factory()
        else:
            raise ValueError(f"No default value for setting: {r.accessor}")
        set_in_dict(
            restore_defaults_obj,
            r.accessor,
            default,
        )
    updates = updates or {}
    set_defaults = set_defaults or {}

    set_defaults_obj = {}
    for setting, value in set_defaults.items():
        set_in_dict(set_defaults_obj, setting.accessor, value)

    updates_obj = {}
    for setting, value in updates.items():
        set_in_dict(updates_obj, setting.accessor, value)

    new_settings = self.__class__.model_validate(
        deep_merge_dicts(
            set_defaults_obj,
            self.model_dump(exclude_unset=True),
            restore_defaults_obj,
            updates_obj,
        )
    )
    return new_settings

emit_warnings()

More post-hoc validation of settings, including warnings for misconfigurations.

Source code in src/prefect/settings/models/root.py

@model_validator(mode="after")
def emit_warnings(self) -> Self:
    """More post-hoc validation of settings, including warnings for misconfigurations."""
    if not self.silence_api_url_misconfiguration:
        _warn_on_misconfigured_api_url(self)
    return self

hash_key()

Return a hash key for the settings object. This is needed since some settings may be unhashable, like lists.

Source code in src/prefect/settings/models/root.py

def hash_key(self) -> str:
    """
    Return a hash key for the settings object.  This is needed since some
    settings may be unhashable, like lists.
    """
    env_variables = self.to_environment_variables()
    return str(hash(tuple((key, value) for key, value in env_variables.items())))

post_hoc_settings()

refactor on resolution of https://github.com/pydantic/pydantic/issues/9789

we should not be modifying pydantic_fields_set directly, but until we can define dependencies between defaults in a first-class way, we need clean up post-hoc default assignments to keep set/unset fields correct after instantiation.

Source code in src/prefect/settings/models/root.py

@model_validator(mode="after")
def post_hoc_settings(self) -> Self:
    """refactor on resolution of https://github.com/pydantic/pydantic/issues/9789

    we should not be modifying __pydantic_fields_set__ directly, but until we can
    define dependencies between defaults in a first-class way, we need clean up
    post-hoc default assignments to keep set/unset fields correct after instantiation.
    """
    if self.ui_url is None:
        self.ui_url = _default_ui_url(self)
        self.__pydantic_fields_set__.remove("ui_url")
    if self.server.ui.api_url is None:
        if self.api.url:
            self.server.ui.api_url = self.api.url
            self.server.ui.__pydantic_fields_set__.remove("api_url")
        else:
            self.server.ui.api_url = (
                f"http://{self.server.api.host}:{self.server.api.port}/api"
            )
            self.server.ui.__pydantic_fields_set__.remove("api_url")
    if self.profiles_path is None or "PREFECT_HOME" in str(self.profiles_path):
        self.profiles_path = Path(f"{self.home}/profiles.toml")
        self.__pydantic_fields_set__.remove("profiles_path")
    if self.results.local_storage_path is None:
        self.results.local_storage_path = Path(f"{self.home}/storage")
        self.results.__pydantic_fields_set__.remove("local_storage_path")
    if self.server.memo_store_path is None:
        self.server.memo_store_path = Path(f"{self.home}/memo_store.toml")
        self.server.__pydantic_fields_set__.remove("memo_store_path")
    if self.debug_mode or self.testing.test_mode:
        self.logging.level = "DEBUG"
        self.internal.logging_level = "DEBUG"
        self.logging.__pydantic_fields_set__.remove("level")
        self.internal.__pydantic_fields_set__.remove("logging_level")

    if self.logging.config_path is None:
        self.logging.config_path = Path(f"{self.home}/logging.yml")
        self.logging.__pydantic_fields_set__.remove("config_path")
    # Set default database connection URL if not provided
    if self.server.database.connection_url is None:
        self.server.database.connection_url = _default_database_connection_url(self)
        self.server.database.__pydantic_fields_set__.remove("connection_url")
    db_url = (
        self.server.database.connection_url.get_secret_value()
        if isinstance(self.server.database.connection_url, SecretStr)
        else self.server.database.connection_url
    )
    if (
        "PREFECT_API_DATABASE_PASSWORD" in db_url
        or "PREFECT_SERVER_DATABASE_PASSWORD" in db_url
    ):
        if self.server.database.password is None:
            raise ValueError(
                "database password is None - please set PREFECT_SERVER_DATABASE_PASSWORD"
            )
        db_url = db_url.replace(
            "${PREFECT_API_DATABASE_PASSWORD}",
            self.server.database.password.get_secret_value()
            if self.server.database.password
            else "",
        )
        db_url = db_url.replace(
            "${PREFECT_SERVER_DATABASE_PASSWORD}",
            self.server.database.password.get_secret_value()
            if self.server.database.password
            else "",
        )
        self.server.database.connection_url = SecretStr(db_url)
        self.server.database.__pydantic_fields_set__.remove("connection_url")
    return self

get_current_settings()

Returns a settings object populated with values from the current settings context or, if no settings context is active, the environment.

Source code in src/prefect/settings/context.py

def get_current_settings() -> Settings:
    """
    Returns a settings object populated with values from the current settings context
    or, if no settings context is active, the environment.
    """
    from prefect.context import SettingsContext

    settings_context = SettingsContext.get()
    if settings_context is not None:
        return settings_context.settings

    return Settings()

load_current_profile()

Load the current profile from the default and current profile paths.

This will not include settings from the current settings context. Only settings that have been persisted to the profiles file will be saved.

Source code in src/prefect/settings/profiles.py

def load_current_profile():
    """
    Load the current profile from the default and current profile paths.

    This will _not_ include settings from the current settings context. Only settings
    that have been persisted to the profiles file will be saved.
    """
    import prefect.context

    profiles = load_profiles()
    context = prefect.context.get_settings_context()

    if context:
        profiles.set_active(context.profile.name)

    return profiles.active_profile

load_profile(name)

Load a single profile by name.

Source code in src/prefect/settings/profiles.py

def load_profile(name: str) -> Profile:
    """
    Load a single profile by name.
    """
    profiles = load_profiles()
    try:
        return profiles[name]
    except KeyError:
        raise ValueError(f"Profile {name!r} not found.")

load_profiles(include_defaults=True)

Load profiles from the current profile path. Optionally include profiles from the default profile path.

Source code in src/prefect/settings/profiles.py

def load_profiles(include_defaults: bool = True) -> ProfilesCollection:
    """
    Load profiles from the current profile path. Optionally include profiles from the
    default profile path.
    """
    current_settings = get_current_settings()
    default_profiles = _read_profiles_from(DEFAULT_PROFILES_PATH)

    if current_settings.profiles_path is None:
        raise RuntimeError(
            "No profiles path set; please ensure `PREFECT_PROFILES_PATH` is set."
        )

    if not include_defaults:
        if not current_settings.profiles_path.exists():
            return ProfilesCollection([])
        return _read_profiles_from(current_settings.profiles_path)

    user_profiles_path = current_settings.profiles_path
    profiles = default_profiles
    if user_profiles_path.exists():
        user_profiles = _read_profiles_from(user_profiles_path)

        # Merge all of the user profiles with the defaults
        for name in user_profiles:
            if not (source := user_profiles[name].source):
                raise ValueError(f"Profile {name!r} has no source.")
            profiles.update_profile(
                name,
                settings=user_profiles[name].settings,
                source=source,
            )

        if user_profiles.active_name:
            profiles.set_active(user_profiles.active_name, check=False)

    return profiles

save_profiles(profiles)

Writes all non-default profiles to the current profiles path.

Source code in src/prefect/settings/profiles.py

def save_profiles(profiles: ProfilesCollection) -> None:
    """
    Writes all non-default profiles to the current profiles path.
    """
    profiles_path = get_current_settings().profiles_path
    assert profiles_path is not None, "Profiles path is not set."
    profiles = profiles.without_profile_source(DEFAULT_PROFILES_PATH)
    return _write_profiles_to(profiles_path, profiles)

temporary_settings(updates=None, set_defaults=None, restore_defaults=None)

Temporarily override the current settings by entering a new profile.

See Settings.copy_with_update for details on different argument behavior.

Examples:

>>> from prefect.settings import PREFECT_API_URL
>>>
>>> with temporary_settings(updates={PREFECT_API_URL: "foo"}):
>>>    assert PREFECT_API_URL.value() == "foo"
>>>
>>>    with temporary_settings(set_defaults={PREFECT_API_URL: "bar"}):
>>>         assert PREFECT_API_URL.value() == "foo"
>>>
>>>    with temporary_settings(restore_defaults={PREFECT_API_URL}):
>>>         assert PREFECT_API_URL.value() is None
>>>
>>>         with temporary_settings(set_defaults={PREFECT_API_URL: "bar"})
>>>             assert PREFECT_API_URL.value() == "bar"
>>> assert PREFECT_API_URL.value() is None

Source code in src/prefect/settings/context.py

@contextmanager
def temporary_settings(
    updates: Optional[Mapping["Setting", Any]] = None,
    set_defaults: Optional[Mapping["Setting", Any]] = None,
    restore_defaults: Optional[Iterable["Setting"]] = None,
) -> Generator[Settings, None, None]:
    """
    Temporarily override the current settings by entering a new profile.

    See `Settings.copy_with_update` for details on different argument behavior.

    Examples:
        >>> from prefect.settings import PREFECT_API_URL
        >>>
        >>> with temporary_settings(updates={PREFECT_API_URL: "foo"}):
        >>>    assert PREFECT_API_URL.value() == "foo"
        >>>
        >>>    with temporary_settings(set_defaults={PREFECT_API_URL: "bar"}):
        >>>         assert PREFECT_API_URL.value() == "foo"
        >>>
        >>>    with temporary_settings(restore_defaults={PREFECT_API_URL}):
        >>>         assert PREFECT_API_URL.value() is None
        >>>
        >>>         with temporary_settings(set_defaults={PREFECT_API_URL: "bar"})
        >>>             assert PREFECT_API_URL.value() == "bar"
        >>> assert PREFECT_API_URL.value() is None
    """
    import prefect.context

    context = prefect.context.get_settings_context()

    if not restore_defaults:
        restore_defaults = []

    new_settings = context.settings.copy_with_update(
        updates=updates, set_defaults=set_defaults, restore_defaults=restore_defaults
    )

    with prefect.context.SettingsContext(
        profile=context.profile, settings=new_settings
    ):
        yield new_settings

update_current_profile(settings)

Update the persisted data for the profile currently in-use.

If the profile does not exist in the profiles file, it will be created.

Given settings will be merged with the existing settings as described in ProfilesCollection.update_profile.

Returns:

Type	Description
Profile	The new profile.

Source code in src/prefect/settings/profiles.py

def update_current_profile(
    settings: Dict[Union[str, Setting], Any],
) -> Profile:
    """
    Update the persisted data for the profile currently in-use.

    If the profile does not exist in the profiles file, it will be created.

    Given settings will be merged with the existing settings as described in
    `ProfilesCollection.update_profile`.

    Returns:
        The new profile.
    """
    import prefect.context

    current_profile = prefect.context.get_settings_context().profile

    if not current_profile:
        from prefect.exceptions import MissingProfileError

        raise MissingProfileError("No profile is currently in use.")

    profiles = load_profiles()

    # Ensure the current profile's settings are present
    profiles.update_profile(current_profile.name, current_profile.settings)
    # Then merge the new settings in
    new_profile = profiles.update_profile(
        current_profile.name, _cast_settings(settings)
    )

    new_profile.validate_settings()

    save_profiles(profiles)

    return profiles[current_profile.name]