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