3983484430

Build Type

push

github

Committed by Simon Charette

Commit Message

Upgraded and tested latest supported deps.

Run Details

169 of 191 branches covered (88.48%)

Branch coverage included in aggregate %.

3 of 3 new or added lines in 1 file covered. (100.0%)

2 existing lines in 1 file now uncovered.

537 of 558 relevant lines covered (96.24%)

4.8 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

91.67

/syzygy/operations.py

from contextlib import contextmanager

from django.db.migrations import operations
from django.db.models.fields import NOT_PROVIDED

from .constants import Stage


def _alter_field_db_default_sql_params(schema_editor, model, name, drop=False):
    field = model._meta.get_field(name)
    changes_sql, params = schema_editor._alter_column_default_sql(
        model, None, field, drop=drop
    )
    sql = schema_editor.sql_alter_column % {
        "table": schema_editor.quote_name(model._meta.db_table),
        "changes": changes_sql,
    }
    return sql, params


def _alter_field_db_default(schema_editor, model, name, drop=False):
    sql, params = _alter_field_db_default_sql_params(
        schema_editor, model, name, drop=drop
    )
    schema_editor.execute(sql, params)


@contextmanager
def _force_field_alteration(schema_editor):
    # Django 3.2 implements an optimization to prevent SQLite table rebuilds
    # when unnecessary. Until proper db_default alteration support lands this
    # optimization has to be disabled under some circumstances.
    _field_should_be_altered = getattr(schema_editor, "_field_should_be_altered", None)
    if _field_should_be_altered is None:
        yield
        return
    schema_editor._field_should_be_altered = lambda old_field, new_field: True
    try:
        yield
    finally:
        schema_editor._field_should_be_altered = _field_should_be_altered


@contextmanager
def _include_column_default(schema_editor, field_name):
    column_sql_ = schema_editor.column_sql

    def column_sql(model, field, include_default=False):
        include_default |= field.name == field_name
        # XXX: SQLite doesn't support parameterized DDL but this isn't an
        # issue upstream since this method is never called with
        # `include_default=True` due to table rebuild.
        sql, params = column_sql_(model, field, include_default)
        return sql % tuple(params), ()

    schema_editor.column_sql = column_sql
    try:
        with _force_field_alteration(schema_editor):
            yield
    finally:
        schema_editor.column_sql = column_sql_


class PreRemoveField(operations.AlterField):
    """
    Perform database operations required to make sure an application with a
    rolling deployment won't crash prior to a field removal.

    If the field has a `default` value defined its corresponding column is
    altered to use it until the field is removed otherwise the field is made
    NULL'able if it's not already.
    """

    def state_forwards(self, app_label, state):
        pass

    def database_forwards(self, app_label, schema_editor, from_state, to_state):
        to_state = to_state.clone()
        super().state_forwards(app_label, to_state)
        model = to_state.apps.get_model(app_label, self.model_name)
        if not self.allow_migrate_model(schema_editor.connection.alias, model):
            return
        # Ensure column meant to be removed have database level default defined
        # or is made NULL'able prior to removal to allow INSERT during the
        # deployment stage.
        field = model._meta.get_field(self.name)
        if field.default is not NOT_PROVIDED:
            if schema_editor.connection.vendor == "sqlite":
                with _include_column_default(schema_editor, self.name):
                    super().database_forwards(
                        app_label, schema_editor, from_state, to_state
                    )
            else:
                _alter_field_db_default(schema_editor, model, self.name)
        else:
            nullable_field = field.clone()
            nullable_field.null = True
            operation = operations.AlterField(
                self.model_name, self.name, nullable_field
            )
            operation.state_forwards(app_label, to_state)
            operation.database_forwards(app_label, schema_editor, from_state, to_state)

    @property
    def migration_name_fragment(self):
        if self.field.default is not NOT_PROVIDED:
            return "set_db_default_%s_%s" % (
                self.model_name_lower,
                self.name,
            )
        return "set_nullable_%s_%s" % (
            self.model_name_lower,
            self.name,
        )

    def describe(self):
        if self.field.default is not NOT_PROVIDED:
            return "Set database DEFAULT of field %s on %s" % (
                self.name,
                self.model_name,
            )
        return "Set field %s of %s NULLable" % (self.name, self.model_name)


class AddField(operations.AddField):
    """
    Subclass of `AddField` that preserves the database default on database
    application.
    """

    @contextmanager
    def _prevent_drop_default(self, schema_editor, model):
        # On other backends the most straightforward way
        drop_default_sql_params = _alter_field_db_default_sql_params(
            schema_editor, model, self.name, drop=True
        )
        execute_ = schema_editor.execute

        def execute(sql, params=()):
            if (sql, params) == drop_default_sql_params:
                return
            return execute_(sql, params)

        schema_editor.execute = execute
        try:
            yield
        finally:
            schema_editor.execute = execute_

    def _preserve_column_default(self, schema_editor, model):
        # XXX: Hopefully future support for `Field.db_default` will add better
        # injection points to `BaseSchemaEditor.add_field`.
        if schema_editor.connection.vendor == "sqlite":
            # On the SQLite backend the strategy is different since it emulates
            # ALTER support by rebuilding tables.
            return _include_column_default(schema_editor, self.name)
        return self._prevent_drop_default(schema_editor, model)

    def database_forwards(self, app_label, schema_editor, from_state, to_state):
        model = to_state.apps.get_model(app_label, self.model_name)
        if not self.allow_migrate_model(schema_editor.connection.alias, model):
            return
        # Defer the removal of DEFAUT to `PostAddField`
        with self._preserve_column_default(schema_editor, model):
            return super().database_forwards(
                app_label, schema_editor, from_state, to_state
            )


class PostAddField(operations.AlterField):
    """
    Elidable operation that drops a previously preserved database default.
    """

    stage = Stage.POST_DEPLOY

    def state_forwards(self, app_label, state):
        pass

    def database_forwards(self, app_label, schema_editor, from_state, to_state):
        model = to_state.apps.get_model(app_label, self.model_name)
        if not self.allow_migrate_model(schema_editor.connection.alias, model):
            return
        if schema_editor.connection.vendor == "sqlite":
            # Trigger a table rebuild to DROP the database level DEFAULT
            with _force_field_alteration(schema_editor):
                super().database_forwards(
                    app_label, schema_editor, from_state, to_state
                )
        else:
            _alter_field_db_default(schema_editor, model, self.name, drop=True)

    def database_backwards(self, app_label, schema_editor, from_state, to_state):
        to_state = to_state.clone()
        super().state_forwards(app_label, to_state)
        model = to_state.apps.get_model(app_label, self.model_name)
        if not self.allow_migrate_model(schema_editor.connection.alias, model):
            return
        if schema_editor.connection.vendor == "sqlite":
            with _include_column_default(schema_editor, self.name):
                super().database_forwards(
                    app_label, schema_editor, from_state, to_state
                )
        else:
            _alter_field_db_default(schema_editor, model, self.name)

    @property
    def migration_name_fragment(self):
        return "drop_db_default_%s_%s" % (
            self.model_name_lower,
            self.name,
        )

    def describe(self):
        return "Drop database DEFAULT of field %s on %s" % (
            self.name,
            self.model_name,
        )


class StagedOperation(operations.base.Operation):
    stage: Stage

    def __init__(self, *args, **kwargs):
        self.stage = kwargs.pop("stage")
        super().__init__(*args, **kwargs)

    @classmethod
    def for_stage(cls, operation: operations.base.Operation, stage: Stage):
        _, args, kwargs = operation.deconstruct()
        kwargs["stage"] = stage
        return cls(*args, **kwargs)

    def deconstruct(self):
        name, args, kwargs = super().deconstruct()
        kwargs["stage"] = self.stage
        return name, args, kwargs


class RenameField(StagedOperation, operations.RenameField):
    """
    Subclass of ``RenameField`` that explicitly defines a stage for the rare
    instances where a rename operation is safe to perform.
    """


class RenameModel(StagedOperation, operations.RenameModel):
    """
    Subclass of ``RenameModel`` that explicitly defines a stage for the rare
    instances where a rename operation is safe to perform.
    """


class AlterField(StagedOperation, operations.AlterField):
    """
    Subclass of ``AlterField`` that allows explicitly defining a stage.
    """

1	from contextlib import contextmanager	5✔
2
3	from django.db.migrations import operations	5✔
4	from django.db.models.fields import NOT_PROVIDED	5✔
5
6	from .constants import Stage	5✔
7
8
9	def _alter_field_db_default_sql_params(schema_editor, model, name, drop=False):	5✔
10	field = model._meta.get_field(name)	5✔
11	changes_sql, params = schema_editor._alter_column_default_sql(	5✔
12	model, None, field, drop=drop
13	)
14	sql = schema_editor.sql_alter_column % {	5✔
15	"table": schema_editor.quote_name(model._meta.db_table),
16	"changes": changes_sql,
17	}
18	return sql, params	5✔
19
20
21	def _alter_field_db_default(schema_editor, model, name, drop=False):	5✔
22	sql, params = _alter_field_db_default_sql_params(	5✔
23	schema_editor, model, name, drop=drop
24	)
25	schema_editor.execute(sql, params)	5✔
26
27
28	@contextmanager	5✔
29	def _force_field_alteration(schema_editor):	4✔
30	# Django 3.2 implements an optimization to prevent SQLite table rebuilds
31	# when unnecessary. Until proper db_default alteration support lands this
32	# optimization has to be disabled under some circumstances.
33	_field_should_be_altered = getattr(schema_editor, "_field_should_be_altered", None)	5✔
34	if _field_should_be_altered is None:	5!
UNCOV 35	yield	×
UNCOV 36	return	×
37	schema_editor._field_should_be_altered = lambda old_field, new_field: True	5✔
38	try:	5✔
39	yield	5✔
40	finally:
41	schema_editor._field_should_be_altered = _field_should_be_altered	5✔
42
43
44	@contextmanager	5✔
45	def _include_column_default(schema_editor, field_name):	4✔
46	column_sql_ = schema_editor.column_sql	5✔
47
48	def column_sql(model, field, include_default=False):	5✔
49	include_default \|= field.name == field_name	5✔
50	# XXX: SQLite doesn't support parameterized DDL but this isn't an
51	# issue upstream since this method is never called with
52	# `include_default=True` due to table rebuild.
53	sql, params = column_sql_(model, field, include_default)	5✔
54	return sql % tuple(params), ()	5✔
55
56	schema_editor.column_sql = column_sql	5✔
57	try:	5✔
58	with _force_field_alteration(schema_editor):	5✔
59	yield	5✔
60	finally:
61	schema_editor.column_sql = column_sql_	5✔
62
63
64	class PreRemoveField(operations.AlterField):	5✔
65	"""
66	Perform database operations required to make sure an application with a
67	rolling deployment won't crash prior to a field removal.
68
69	If the field has a `default` value defined its corresponding column is
70	altered to use it until the field is removed otherwise the field is made
71	NULL'able if it's not already.
72	"""
73
74	def state_forwards(self, app_label, state):	5✔
75	pass	5✔
76
77	def database_forwards(self, app_label, schema_editor, from_state, to_state):	5✔
78	to_state = to_state.clone()	5✔
79	super().state_forwards(app_label, to_state)	5✔
80	model = to_state.apps.get_model(app_label, self.model_name)	5✔
81	if not self.allow_migrate_model(schema_editor.connection.alias, model):	5!
82	return	×
83	# Ensure column meant to be removed have database level default defined
84	# or is made NULL'able prior to removal to allow INSERT during the
85	# deployment stage.
86	field = model._meta.get_field(self.name)	5✔
87	if field.default is not NOT_PROVIDED:	5✔
88	if schema_editor.connection.vendor == "sqlite":	5✔
89	with _include_column_default(schema_editor, self.name):	5✔
90	super().database_forwards(	5✔
91	app_label, schema_editor, from_state, to_state
92	)
93	else:
94	_alter_field_db_default(schema_editor, model, self.name)	5✔
95	else:
96	nullable_field = field.clone()	5✔
97	nullable_field.null = True	5✔
98	operation = operations.AlterField(	5✔
99	self.model_name, self.name, nullable_field
100	)
101	operation.state_forwards(app_label, to_state)	5✔
102	operation.database_forwards(app_label, schema_editor, from_state, to_state)	5✔
103
104	@property	5✔
105	def migration_name_fragment(self):	4✔
106	if self.field.default is not NOT_PROVIDED:	5✔
107	return "set_db_default_%s_%s" % (	5✔
108	self.model_name_lower,
109	self.name,
110	)
111	return "set_nullable_%s_%s" % (	5✔
112	self.model_name_lower,
113	self.name,
114	)
115
116	def describe(self):	5✔
117	if self.field.default is not NOT_PROVIDED:	5✔
118	return "Set database DEFAULT of field %s on %s" % (	5✔
119	self.name,
120	self.model_name,
121	)
122	return "Set field %s of %s NULLable" % (self.name, self.model_name)	5✔
123
124
125	class AddField(operations.AddField):	5✔
126	"""
127	Subclass of `AddField` that preserves the database default on database
128	application.
129	"""
130
131	@contextmanager	5✔
132	def _prevent_drop_default(self, schema_editor, model):	4✔
133	# On other backends the most straightforward way
134	drop_default_sql_params = _alter_field_db_default_sql_params(	5✔
135	schema_editor, model, self.name, drop=True
136	)
137	execute_ = schema_editor.execute	5✔
138
139	def execute(sql, params=()):	5✔
140	if (sql, params) == drop_default_sql_params:	5✔
141	return	5✔
142	return execute_(sql, params)	5✔
143
144	schema_editor.execute = execute	5✔
145	try:	5✔
146	yield	5✔
147	finally:
148	schema_editor.execute = execute_	5✔
149
150	def _preserve_column_default(self, schema_editor, model):	5✔
151	# XXX: Hopefully future support for `Field.db_default` will add better
152	# injection points to `BaseSchemaEditor.add_field`.
153	if schema_editor.connection.vendor == "sqlite":	5✔
154	# On the SQLite backend the strategy is different since it emulates
155	# ALTER support by rebuilding tables.
156	return _include_column_default(schema_editor, self.name)	5✔
157	return self._prevent_drop_default(schema_editor, model)	5✔
158
159	def database_forwards(self, app_label, schema_editor, from_state, to_state):	5✔
160	model = to_state.apps.get_model(app_label, self.model_name)	5✔
161	if not self.allow_migrate_model(schema_editor.connection.alias, model):	5!
162	return	×
163	# Defer the removal of DEFAUT to `PostAddField`
164	with self._preserve_column_default(schema_editor, model):	5✔
165	return super().database_forwards(	5✔
166	app_label, schema_editor, from_state, to_state
167	)
168
169
170	class PostAddField(operations.AlterField):	5✔
171	"""
172	Elidable operation that drops a previously preserved database default.
173	"""
174
175	stage = Stage.POST_DEPLOY	5✔
176
177	def state_forwards(self, app_label, state):	5✔
178	pass	5✔
179
180	def database_forwards(self, app_label, schema_editor, from_state, to_state):	5✔
181	model = to_state.apps.get_model(app_label, self.model_name)	5✔
182	if not self.allow_migrate_model(schema_editor.connection.alias, model):	5!
183	return	×
184	if schema_editor.connection.vendor == "sqlite":	5✔
185	# Trigger a table rebuild to DROP the database level DEFAULT
186	with _force_field_alteration(schema_editor):	5✔
187	super().database_forwards(	5✔
188	app_label, schema_editor, from_state, to_state
189	)
190	else:
191	_alter_field_db_default(schema_editor, model, self.name, drop=True)	5✔
192
193	def database_backwards(self, app_label, schema_editor, from_state, to_state):	5✔
194	to_state = to_state.clone()	5✔
195	super().state_forwards(app_label, to_state)	5✔
196	model = to_state.apps.get_model(app_label, self.model_name)	5✔
197	if not self.allow_migrate_model(schema_editor.connection.alias, model):	5!
198	return	×
199	if schema_editor.connection.vendor == "sqlite":	5✔
200	with _include_column_default(schema_editor, self.name):	5✔
201	super().database_forwards(	5✔
202	app_label, schema_editor, from_state, to_state
203	)
204	else:
205	_alter_field_db_default(schema_editor, model, self.name)	5✔
206
207	@property	5✔
208	def migration_name_fragment(self):	4✔
209	return "drop_db_default_%s_%s" % (	5✔
210	self.model_name_lower,
211	self.name,
212	)
213
214	def describe(self):	5✔
215	return "Drop database DEFAULT of field %s on %s" % (	5✔
216	self.name,
217	self.model_name,
218	)
219
220
221	class StagedOperation(operations.base.Operation):	5✔
222	stage: Stage	5✔
223
224	def __init__(self, args, *kwargs):	5✔
225	self.stage = kwargs.pop("stage")	5✔
226	super().__init__(args, *kwargs)	5✔
227
228	@classmethod	5✔
229	def for_stage(cls, operation: operations.base.Operation, stage: Stage):	5✔
230	_, args, kwargs = operation.deconstruct()	5✔
231	kwargs["stage"] = stage	5✔
232	return cls(args, *kwargs)	5✔
233
234	def deconstruct(self):	5✔
235	name, args, kwargs = super().deconstruct()	×
236	kwargs["stage"] = self.stage	×
237	return name, args, kwargs	×
238
239
240	class RenameField(StagedOperation, operations.RenameField):	5✔
241	"""
242	Subclass of ``RenameField`` that explicitly defines a stage for the rare
243	instances where a rename operation is safe to perform.
244	"""
245
246
247	class RenameModel(StagedOperation, operations.RenameModel):	5✔
248	"""
249	Subclass of ``RenameModel`` that explicitly defines a stage for the rare
250	instances where a rename operation is safe to perform.
251	"""
252
253
254	class AlterField(StagedOperation, operations.AlterField):	5✔
255	"""
256	Subclass of ``AlterField`` that allows explicitly defining a stage.
257	"""

charettes / django-syzygy / 3983484430

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous