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        )