22

✏️ Editing: 152_add_listname_to_primary_key.py

from peewee import (
    BooleanField,
    CharField,
    Check,
    CompositeKey,
    ForeignKeyField,
    IntegerField,
    Model,
)
import time

def migrate(migrator, database, fake=False, **kwargs):
    orm_IPList = migrator.orm["iplist"]
    Country = migrator.orm["country"]

class TMP_IPList(Model):
        """'iplist' db table."""

#: field name
        ACTION_TYPE = "action_type"
        #: available list names
        IP_LISTS = [WHITE, BLACK, GRAY, GRAY_SPLASHSCREEN] = (
            "WHITE",
            "BLACK",
            "GRAY",
            "GRAY_SPLASHSCREEN",
        )
        SCOPE_LOCAL, SCOPE_GROUP = "local", "group"
        ip = CharField(null=False)
        listname = CharField(
            null=False,
            constraints=[
                Check("listname in ('{}')".format("','".join(IP_LISTS)))
            ],
        )

# null=True to be consistent
        # with previously used create table sql
        expiration = IntegerField(
            default=0, null=True  # 0 - never
        )  # null - the same :(

imported_from = CharField(null=True)
        ctime = IntegerField(
            null=True, default=lambda: int(time.time())  # those
        )  # are OK

deep = IntegerField(null=True)
        comment = CharField(null=True)
        country = ForeignKeyField(Country, null=True)

# actual for not manually whitelisted ips only
        # should be ignored for others
        captcha_passed = BooleanField(null=False, default=False)

# available only for graylist
        manual = BooleanField(null=False, default=True)

# available only for whitelist
        full_access = BooleanField(null=True)
        # was IP autowhitelisted with `--remote-addr` flag or not
        auto_whitelisted = BooleanField(null=True, default=False)

network_address = IntegerField(null=False)
        netmask = IntegerField(null=False)
        version = IntegerField(null=False)
        scope = CharField(
            null=True,
            constraints=[
                Check("scope in ('%s','%s')" % (SCOPE_LOCAL, SCOPE_GROUP))
            ],
        )

class Meta:
            db_table = "tmpiplist"
            primary_key = CompositeKey(
                "network_address", "netmask", "version", "listname"
            )

migrator.create_model(TMP_IPList)

# we can't use migrator.rename_table due to bug in peewee_migrate
    # https://github.com/klen/peewee_migrate/pull/158
    #
    # Also, sqlite could mix fields in command
    # insert into table select * from other_table
    # https://stackoverflow.com/questions/56682520/copy-sqlite-table-with-mixed-column-order
    #
    fields = [
        name
        for name in TMP_IPList._meta.sorted_field_names
        if name != "country"
    ] + ["country_id"]
    migrator.sql(
        "INSERT INTO tmpiplist ({fields}) SELECT {fields} FROM iplist".format(
            fields=",".join(fields)
        )
    )
    migrator.sql("DROP TABLE iplist")
    migrator.sql("ALTER TABLE tmpiplist RENAME TO iplist")
    migrator.add_index(orm_IPList, "listname")
    migrator.add_index(orm_IPList, "expiration")
    migrator.add_index(orm_IPList, "ip")

def rollback(migrator, database, fake=False, **kwargs):
    pass

MR SKB