Produits Entr'ouvert » Chrono

# chrono - agendas system

# Copyright (C) 2021  Entr'ouvert

#

# This program is free software: you can redistribute it and/or modify it

# under the terms of the GNU Affero General Public License as published

# by the Free Software Foundation, either version 3 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

# GNU Affero General Public License for more details.

#

# You should have received a copy of the GNU Affero General Public License

# along with this program.  If not, see <http://www.gnu.org/licenses/>.

from django.apps import apps

from django.contrib.postgres.fields import JSONField

from django.core.management.base import BaseCommand, CommandError

from django.db import connection

class Command(BaseCommand):

    help = 'Ensure all JSON fields are of type jsonb'

    def handle(self, **options):

        for app in apps.get_models():

            for field in app._meta.get_fields():

                if isinstance(field, JSONField):

                    table_name = app._meta.db_table

                    column_name = app._meta.get_field(field.name).column

                    with connection.cursor() as cursor:

                        query = '''SELECT table_schema

                                     FROM information_schema.columns

                                    WHERE table_name = %s AND column_name = %s AND data_type != %s'''

                        cursor.execute(query, [table_name, column_name, 'jsonb'])

                        for line in cursor.fetchall():

                            alter_query = '''ALTER TABLE "%(schema_name)s"."%(table_name)s"

                                            ALTER COLUMN "%(column_name)s"

                                                    TYPE jsonb USING "%(column_name)s"::jsonb'''

                            params = {

                                'schema_name': line[0],

                                'table_name': table_name,

                                'column_name': column_name,

                            }

                            try:

                                cursor.execute(alter_query % params)

                            except Exception as e:

                                raise CommandError(e)

from django.contrib.postgres.fields import JSONField

                ('extra_data', JSONField(null=True)),

from django.contrib.postgres.fields import JSONField

                ('booking_errors', JSONField(default=dict)),

from django.contrib.postgres.fields import JSONField

            field=JSONField(null=True, verbose_name='Recurrence rule', blank=True),

from django.db import migrations

from chrono.utils.db import EnsureJsonbType

class Migration(migrations.Migration):

    dependencies = [

        ('agendas', '0078_absence_reasons'),

    ]

    operations = [

        EnsureJsonbType(model_name='booking', field_name='extra_data'),

        EnsureJsonbType(model_name='event', field_name='recurrence_rule'),

        EnsureJsonbType(model_name='eventcancellationreport', field_name='booking_errors'),

    ]

from django.contrib.postgres.fields import ArrayField, JSONField

    recurrence_rule = JSONField(_('Recurrence rule'), null=True, blank=True)

    booking_errors = JSONField(default=dict)

# chrono - agendas system

# Copyright (C) 2021  Entr'ouvert

#

# This program is free software: you can redistribute it and/or modify it

# under the terms of the GNU Affero General Public License as published

# by the Free Software Foundation, either version 3 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

# GNU Affero General Public License for more details.

#

# You should have received a copy of the GNU Affero General Public License

# along with this program.  If not, see <http://www.gnu.org/licenses/>.

from django.db import connection

from django.db.migrations.operations.base import Operation

class EnsureJsonbType(Operation):

    reversible = True

    def __init__(self, model_name, field_name):

        self.model_name = model_name

        self.field_name = field_name

    def state_forwards(self, app_label, state):

        pass

    def database_forwards(self, app_label, schema_editor, from_state, to_state):

        if connection.vendor == 'postgresql':

            model = from_state.apps.get_model(app_label, self.model_name)

            table_name = model._meta.db_table

            field = model._meta.get_field(self.field_name)

            _, column_name = field.get_attname_column()

            with schema_editor.connection.cursor() as cursor:

                cursor.execute(

                    'ALTER TABLE {table} ALTER COLUMN {col} TYPE jsonb USING {col}::jsonb;'.format(

                        table=table_name, col=column_name

                    )

                )

    def database_backwards(self, app_label, schema_editor, from_state, to_state):

        pass

    def describe(self):

        return "Migrate to postgres jsonb type"

# -*- coding: utf-8 -*-

import pytest

from django.core.management import call_command

from django.db import connection

pytestmark = pytest.mark.django_db

@pytest.mark.skipif(connection.vendor != 'postgresql', reason='only postgresql is supported')

def test_ensure_jsonb_fields():

    json_fields = (

        'extra_data',

        'booking_errors',

        'recurrence_rule',

    )

    with connection.cursor() as cursor:

        query = '''SELECT table_name, column_name, data_type

                     FROM information_schema.columns

                    WHERE column_name IN %(json_fields)s'''

        cursor.execute(query, {'json_fields': json_fields})

        # make sure the data_type is correct

        for line in cursor.fetchall():

            assert line[2] == 'jsonb'

        # alter columns

        cursor.execute(

            '''ALTER TABLE agendas_booking

              ALTER COLUMN extra_data TYPE text USING extra_data::text'''

        )

        cursor.execute(

            '''ALTER TABLE agendas_eventcancellationreport

              ALTER COLUMN booking_errors TYPE text USING booking_errors::text'''

        )

        cursor.execute(

            '''ALTER TABLE agendas_event

              ALTER COLUMN recurrence_rule TYPE text USING recurrence_rule::text'''

        )

    call_command('ensure_jsonb')

    with connection.cursor() as cursor:

        query = '''SELECT table_name, column_name, data_type

                     FROM information_schema.columns

                    WHERE column_name IN %(json_fields)s'''

        cursor.execute(query, {'json_fields': json_fields})

        # check the data_type is correct

        for line in cursor.fetchall():

            assert line[2] == 'jsonb'