import peewee as pw class BlockedPort(pw.Model): """ Port + protocol for blocking data """ port = pw.IntegerField(null=False) proto = pw.CharField( null=False, constraints=[pw.Check("proto in ('tcp', 'udp', 'all')")] ) comment = pw.CharField(null=True) class Meta: db_table = "blocked_port" indexes = ( # create an unique on port/proto (("port", "proto"), True), ) class IgnoredByPort(pw.Model): """ Ignored IPs for port + protocol """ port_proto = pw.ForeignKeyField( BlockedPort, null=False, on_delete="CASCADE", related_name="ips" ) ip = pw.CharField(null=False) comment = pw.CharField(null=True) class Meta: db_table = "ignored_by_port_proto" indexes = ( # create an unique on port/ip (("port_proto", "ip"), True), ) def migrate(migrator, database, fake=False, **kwargs): migrator.create_model(BlockedPort) migrator.create_model(IgnoredByPort) IPList = migrator.orm["iplist"] migrator.add_fields(IPList, full_access=pw.BooleanField(null=True)) def rollback(migrator, database, fake=False, **kwargs): BlockedPort = migrator.orm["blocked_port"] IgnoredByPort = migrator.orm["blocked_port_ip"] IPList = migrator.orm["iplist"] migrator.remove_model(BlockedPort) migrator.remove_model(IgnoredByPort) migrator.remove_fields(IPList, "full_access")