import ipaddress def migrate(migrator, database, fake=False, **kwargs): if fake: return IgnoreListNew = migrator.orm["ignore_list_new"] IgnoreList = migrator.orm["ignore_list"] try: from defence360agent.utils.validate import IP from im360.utils.net import pack_ip_network except ImportError: pass else: with database.atomic(): # FIXME: after migrating to peewee 3 add .iterator() ip_strings = [item["ip"] for item in IgnoreList.select().dicts()] ips = set() for item in ip_strings: try: ip = ipaddress.ip_network(item) except ValueError: # malformed ip continue ips.add(ip) for ip in ips: net, mask, version = pack_ip_network(ip) IgnoreListNew.create( ip=IP.ip_net_to_string(ip), network_address=net, netmask=mask, version=version, ) migrator.sql("DROP TABLE ignore_list") migrator.sql("ALTER TABLE ignore_list_new RENAME TO ignore_list") def rollback(migrator, database, fake=False, **kwargs): """Write your rollback migrations here.""" pass