65 lines
3.0 KiB
Python
65 lines
3.0 KiB
Python
from django.db import migrations, models
|
||
from django.db.models import Count, F
|
||
|
||
|
||
def reconcile_schema(apps, schema_editor):
|
||
"""Réconcilie la base réelle après les changements d'état seuls de 0004."""
|
||
entry = apps.get_model('time_tracking', 'TimeEntry')
|
||
rows = entry.objects.using(schema_editor.connection.alias)
|
||
# Ne jamais inventer une heure ni supprimer un doublon historique.
|
||
if rows.filter(start_time__isnull=True).exists():
|
||
raise RuntimeError('Corrigez les pointages sans heure d’entrée avant la migration 0005.')
|
||
if rows.values('user_id', 'start_date', 'start_time').annotate(n=Count('pk')).filter(n__gt=1).exists():
|
||
raise RuntimeError('Corrigez les doublons de pointages avant la migration 0005.')
|
||
rows.filter(end_date__isnull=True).update(end_date=F('start_date'))
|
||
|
||
connection = schema_editor.connection
|
||
with connection.cursor() as cursor:
|
||
columns = {
|
||
column.name: column
|
||
for column in connection.introspection.get_table_description(cursor, entry._meta.db_table)
|
||
}
|
||
constraints = connection.introspection.get_constraints(cursor, entry._meta.db_table)
|
||
|
||
for field_name in ('start_time', 'end_date'):
|
||
if columns[field_name].null_ok:
|
||
old_field = entry._meta.get_field(field_name).clone()
|
||
old_field.set_attributes_from_name(field_name)
|
||
old_field.model = entry
|
||
old_field.null = True
|
||
new_field = old_field.clone()
|
||
new_field.set_attributes_from_name(field_name)
|
||
new_field.model = entry
|
||
new_field.null = False
|
||
schema_editor.alter_field(entry, old_field, new_field, strict=True)
|
||
|
||
fields = ['user', 'start_date', 'start_time']
|
||
target = 'unique_time_entry_start'
|
||
if target not in constraints:
|
||
schema_editor.add_constraint(entry, models.UniqueConstraint(fields=fields, name=target))
|
||
# L'ancien unique_together a été retiré de l'état, mais peut encore exister en SQL.
|
||
for name, constraint in constraints.items():
|
||
if (name != target and constraint['unique']
|
||
and constraint['columns'] == ['user_id', 'start_date', 'start_time']):
|
||
schema_editor.remove_constraint(entry, models.UniqueConstraint(fields=fields, name=name))
|
||
|
||
|
||
class Migration(migrations.Migration):
|
||
atomic = False
|
||
dependencies = [('time_tracking', '0004_alter_timeentry_options_and_more')]
|
||
|
||
operations = [
|
||
migrations.SeparateDatabaseAndState(
|
||
database_operations=[migrations.RunPython(reconcile_schema, atomic=False)],
|
||
state_operations=[
|
||
migrations.AlterField(model_name='timeentry', name='end_date', field=models.DateField()),
|
||
migrations.AddConstraint(
|
||
model_name='timeentry',
|
||
constraint=models.UniqueConstraint(
|
||
fields=('user', 'start_date', 'start_time'), name='unique_time_entry_start',
|
||
),
|
||
),
|
||
],
|
||
),
|
||
]
|