-
Notifications
You must be signed in to change notification settings - Fork 2.3k
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Add InvokeAIAppConfig schema migration system #6243
Changes from 14 commits
6ad1948
36495b7
b612c73
e39f035
aca9e44
6f128c8
5d411e4
d12fb7d
984dd93
ab9ebef
6eaed9a
048306b
ab086a7
8144a26
d248775
d852ca7
59deef9
36b1434
d5aee87
a48abfa
2bba7f3
2dd42d0
31f6302
fc23b16
6946a38
18b5aaf
4c081d5
d487102
7d8b011
964adb8
00ccd73
6e40142
8b76d11
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,89 @@ | ||
# Copyright 2024 Lincoln D. Stein and the InvokeAI Development Team | ||
|
||
""" | ||
Utility class for migrating among versions of the InvokeAI app config schema. | ||
""" | ||
|
||
from dataclasses import dataclass | ||
from typing import Any, Callable, List, TypeAlias | ||
|
||
from packaging.version import Version | ||
|
||
AppConfigDict: TypeAlias = dict[str, Any] | ||
MigrationFunction: TypeAlias = Callable[[AppConfigDict], AppConfigDict] | ||
|
||
|
||
@dataclass | ||
class MigrationEntry: | ||
"""Defines an individual migration.""" | ||
|
||
from_version: Version | ||
to_version: Version | ||
function: MigrationFunction | ||
|
||
|
||
class ConfigMigrator: | ||
"""This class allows migrators to register their input and output versions.""" | ||
|
||
_migrations: List[MigrationEntry] = [] | ||
|
||
@classmethod | ||
def register( | ||
cls, | ||
from_version: str, | ||
to_version: str, | ||
) -> Callable[[MigrationFunction], MigrationFunction]: | ||
"""Define a decorator which registers the migration between two versions.""" | ||
|
||
def decorator(function: MigrationFunction) -> MigrationFunction: | ||
if any(from_version == m.from_version for m in cls._migrations): | ||
raise ValueError( | ||
f"function {function.__name__} is trying to register a migration for version {str(from_version)}, but this migration has already been registered." | ||
) | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. With the added strictness for contiguous migrations, this should probably raise when either There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Fixed. |
||
cls._migrations.append( | ||
MigrationEntry(from_version=Version(from_version), to_version=Version(to_version), function=function) | ||
) | ||
return function | ||
|
||
return decorator | ||
|
||
@staticmethod | ||
def _check_for_discontinuities(migrations: List[MigrationEntry]) -> None: | ||
current_version = Version("3.0.0") | ||
for m in migrations: | ||
if current_version != m.from_version: | ||
raise ValueError( | ||
f"Migration functions are not continuous. Expected from_version={current_version} but got from_version={m.from_version}, for migration function {m.function.__name__}" | ||
) | ||
current_version = m.to_version | ||
|
||
@classmethod | ||
def migrate(cls, config_dict: AppConfigDict) -> AppConfigDict: | ||
""" | ||
Use the registered migration steps to bring config up to latest version. | ||
|
||
:param config: The original configuration. | ||
:return: The new configuration, lifted up to the latest version. | ||
|
||
As a side effect, the new configuration will be written to disk. | ||
If an inconsistency in the registered migration steps' `from_version` | ||
and `to_version` parameters are identified, this will raise a | ||
ValueError exception. | ||
""" | ||
# Sort migrations by version number and raise a ValueError if | ||
# any version range overlaps are detected. | ||
sorted_migrations = sorted(cls._migrations, key=lambda x: x.from_version) | ||
cls._check_for_discontinuities(sorted_migrations) | ||
|
||
if "InvokeAI" in config_dict: | ||
version = Version("3.0.0") | ||
else: | ||
version = Version(config_dict["schema_version"]) | ||
|
||
for migration in sorted_migrations: | ||
if version == migration.from_version and version < migration.to_version: | ||
config_dict = migration.function(config_dict) | ||
version = migration.to_version | ||
|
||
config_dict["schema_version"] = str(version) | ||
return config_dict |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Because this class is a stateful singleton, testing is difficult. Right now, during tests, this class has all migrations registered in the "live" app. Test failures could possibly pop up over time as more migrations are registered.
Making this an instanced class would mitigate those issues.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Fixed.
ConfigMigrator
is now an instanced class, andtest_config.py
creates new instances of the migrator for testing decorator functionality, and continuity checking.The migration steps themselves have been moved into their own module now:
invokeai.app.services.config.migrations
.I also relocated
get_config()
intoinvokeai.app.services.config.config_migrate
, and changed all places that are importingconfig_default
to import frominvokeai.app.services.config
instead (get_config()
is reexported from__init__.py
) . This avoided circular import problems and makesconfig_default.py
cleaner looking overall.