how to do CIDR overlapping in SQL with SQLite / MySQL / SQLAlchemy

gistfile1.py

from sqlalchemy import event
from sqlalchemy import DDL


def _mysql_cidr_overlap(metadata):
    @event.listens_for(metadata, "after_create")
    def _create_mysql_proc(target, connection, **kw):
        if connection.engine.name != 'mysql':
            return

        if connection.scalar(
            "SELECT ROUTINE_NAME FROM INFORMATION_SCHEMA.ROUTINES "
            "WHERE ROUTINE_TYPE='FUNCTION' AND ROUTINE_SCHEMA=DATABASE() AND "
            "ROUTINE_NAME=%s",
            ("cidr_overlap", )
        ):
            connection.execute("DROP FUNCTION cidr_overlap")

        connection.execute(
            DDL("""
        CREATE FUNCTION cidr_overlap (cidr1 VARCHAR(30), cidr2 VARCHAR(30))
        RETURNS TINYINT
        BEGIN
        DECLARE bitmask INT;
        -- note - Mike is semi-guessing on the math here, needs tests!  don't stick
        -- into production pls :)
        SET bitmask = pow(
            2,
            (32 - least(
                cast(substring_index(cidr1, '/', -1) as integer),
                cast(substring_index(cidr2, '/', -1) as integer)
            ))
        ) - 1;

        RETURN
            inet_aton(substring_index(cidr1, '/', 1)) & ~bitmask =
            inet_aton(substring_index(cidr2, '/', 1)) & ~bitmask;
        END
        """)
        )


def _sqlite_cidr_overlap(engine):
    import ipaddr

    def python_cidr_overlap(n1, n2):
        n1 = ipaddr.IPNetwork(n1)
        n2 = ipaddr.IPNetwork(n2)
        return n1.overlaps(n2)

    @event.listens_for(engine, "connect")
    def connect(dbapi_connection, connection_record):
        if e.name == 'sqlite':
            dbapi_connection.create_function(
                "cidr_overlap", 2, python_cidr_overlap)


def cidr_overlap(engine, metadata):
    if engine.name == 'mysql':
        _mysql_cidr_overlap(metadata)
    elif engine.name == 'sqlite':
        _sqlite_cidr_overlap(engine)

if __name__ == '__main__':
    from sqlalchemy import Column, Integer, String, create_engine, func
    from sqlalchemy.ext.declarative import declarative_base
    from sqlalchemy.orm import Session, aliased

    Base = declarative_base()

    class A(Base):
        __tablename__ = 'a'
        id = Column(Integer, primary_key=True)
        subnet = Column(String(30))

    for url in [
        "mysql://scott:tiger@localhost/test",
        "sqlite://"
    ]:
        e = create_engine(url, echo=True)

        cidr_overlap(e, Base.metadata)

        Base.metadata.drop_all(e)
        Base.metadata.create_all(e)

        s = Session(e)

        s.add_all([
            A(subnet='192.168.1.0/24'),
            A(subnet='192.168.2.0/24'),
            A(subnet='192.168.2.0/25')
        ])
        s.commit()

        a1, a2 = aliased(A), aliased(A)

        # return all non-overlapping CIDR pairs
        for a, b in s.query(a1.subnet, a2.subnet).\
            filter(~func.cidr_overlap(a1.subnet, a2.subnet)).\
                filter(a1.id > a2.id):
                print a, b

Hi Mr.Mike

Thanks for reply my comment. Actuallty, it run normally with both "single quotes" and "double quotes". And the resulf is very good However, maybe it should change from "integer" to "signed integer".

Could I ask you a question. Following the process of this code. We will create some subnets (192.168.1.0/24, 192.168.2.0/24.. ) then filter CIDR, which was overlap during show information. Do you have suggestion to prevent overlapIP before creating into DB?

However, maybe it should change from "integer" to "signed integer".

hm, that integer is the part to the right of the slash, e.g. "22" in 192.168.100.0/22, can that number be negative?

We will create some subnets (192.168.1.0/24, 192.168.2.0/24.. ) then filter CIDR, which was overlap during show information. Do you have suggestion to prevent overlapIP before creating into DB?

I think we talked about this at summit, and a strong way to do this would be to create a trigger (a CHECK constraint would work also by MySQL doesnt support CHECK contraints). Let me see if I can make a proof of concept.

here you go, works great https://gist.github.com/zzzeek/3cf63d88dbc5246dcfafe9828ab92e3e we can make similar triggers in Postgresql and possibly SQLite.

Oh. Great. Many thanks for your help. I am running it and it's ok. I will try to apply this to OpenStack to see what will happen. Thanks again, Mr.Mike.

Following your support. I have pushed new patch set which follow your suggestion. So could you please review it [1].
By the way, I would like to add you as "Co-Authored-By" to thank for your support me. What do you think?

[1] https://review.openstack.org/#/c/314054/