1from __future__ import annotations
2
3from typing import TYPE_CHECKING, Any
4
5if TYPE_CHECKING:
6 from plain.models.backends.base.base import BaseDatabaseWrapper
7 from plain.models.backends.base.schema import BaseDatabaseSchemaEditor
8 from plain.models.migrations.state import ProjectState
9
10
11class Operation:
12 """
13 Base class for migration operations.
14
15 It's responsible for both mutating the in-memory model state
16 (see db/migrations/state.py) to represent what it performs, as well
17 as actually performing it against a live database.
18
19 Note that some operations won't modify memory state at all (e.g. data
20 copying operations), and some will need their modifications to be
21 optionally specified by the user (e.g. custom Python code snippets)
22
23 Due to the way this class deals with deconstruction, it should be
24 considered immutable.
25 """
26
27 # Can this migration be represented as SQL? (things like RunPython cannot)
28 reduces_to_sql = True
29
30 # Should this operation be forced as atomic even on backends with no
31 # DDL transaction support (i.e., does it have no DDL, like RunPython)
32 atomic = False
33
34 # Should this operation be considered safe to elide and optimize across?
35 elidable = False
36
37 serialization_expand_args: list[str] = []
38
39 def __new__(cls, *args: Any, **kwargs: Any) -> Operation:
40 # We capture the arguments to make returning them trivial
41 self = object.__new__(cls)
42 self._constructor_args = (args, kwargs) # type: ignore[attr-defined]
43 return self # type: ignore[return-value]
44
45 def deconstruct(self) -> tuple[str, tuple[Any, ...], dict[str, Any]]:
46 """
47 Return a 3-tuple of class import path (or just name if it lives
48 under plain.models.migrations), positional arguments, and keyword
49 arguments.
50 """
51 return (
52 self.__class__.__name__,
53 self._constructor_args[0], # type: ignore[attr-defined]
54 self._constructor_args[1], # type: ignore[attr-defined]
55 )
56
57 def state_forwards(self, package_label: str, state: ProjectState) -> None:
58 """
59 Take the state from the previous migration, and mutate it
60 so that it matches what this migration would perform.
61 """
62 raise NotImplementedError(
63 "subclasses of Operation must provide a state_forwards() method"
64 )
65
66 def database_forwards(
67 self,
68 package_label: str,
69 schema_editor: BaseDatabaseSchemaEditor,
70 from_state: ProjectState,
71 to_state: ProjectState,
72 ) -> None:
73 """
74 Perform the mutation on the database schema in the normal
75 (forwards) direction.
76 """
77 raise NotImplementedError(
78 "subclasses of Operation must provide a database_forwards() method"
79 )
80
81 def describe(self) -> str:
82 """
83 Output a brief summary of what the action does.
84 """
85 return f"{self.__class__.__name__}: {self._constructor_args}" # type: ignore[attr-defined]
86
87 @property
88 def migration_name_fragment(self) -> str | None:
89 """
90 A filename part suitable for automatically naming a migration
91 containing this operation, or None if not applicable.
92 """
93 return None
94
95 def references_model(self, name: str, package_label: str) -> bool:
96 """
97 Return True if there is a chance this operation references the given
98 model name (as a string), with an app label for accuracy.
99
100 Used for optimization. If in doubt, return True;
101 returning a false positive will merely make the optimizer a little
102 less efficient, while returning a false negative may result in an
103 unusable optimized migration.
104 """
105 return True
106
107 def references_field(self, model_name: str, name: str, package_label: str) -> bool:
108 """
109 Return True if there is a chance this operation references the given
110 field name, with an app label for accuracy.
111
112 Used for optimization. If in doubt, return True.
113 """
114 return self.references_model(model_name, package_label)
115
116 def allow_migrate_model(self, connection: BaseDatabaseWrapper, model: Any) -> bool:
117 """Return whether or not a model may be migrated."""
118 if not model.model_options.can_migrate(connection):
119 return False
120
121 return True
122
123 def reduce(
124 self, operation: Operation, package_label: str
125 ) -> list[Operation] | bool:
126 """
127 Return either a list of operations the actual operation should be
128 replaced with or a boolean that indicates whether or not the specified
129 operation can be optimized across.
130 """
131 if self.elidable:
132 return [operation]
133 elif operation.elidable:
134 return [self]
135 return False
136
137 def __repr__(self) -> str:
138 return "<{} {}{}>".format(
139 self.__class__.__name__,
140 ", ".join(map(repr, self._constructor_args[0])), # type: ignore[attr-defined]
141 ",".join(" {}={!r}".format(*x) for x in self._constructor_args[1].items()), # type: ignore[attr-defined]
142 )