How to format a SQLAlchemy ORM Postgresql CIDR contains (>>) query

Question

I have a model defined with:

from app import db
from sqlalchemy.dialects import postgresql

class TableIpAddress(db.Model):
  __tablename__ = 'ipaddress'
  idipaddress = db.Column( postgresql.UUID, primary_key=True )
  ipaddress = db.Column( postgresql.CIDR, index=True, nullable=False )

I would like to do something like the following:

ip = '192.168.0.0/16'
db.session.query( TableIpAddress.ipaddress.op('<<')(ip) ).all()

This results in the key portion of the error message with:

sqlalchemy.exc.DBAPIError: 
    (ParameterError) could not pack parameter $1::pg_catalog.inet for transfer

The field is actually a CIDR field. It doesn't seem to know how to pack a CIDR. Is there a way to coerce the parameter to the appropriate type?

Answer 1

This can be done using SQLAlchemy and psycopg2. Psycopg2 needs some configuration to handle types from Python's ipaddress module; this can be done in an event listener that fires when the first connection is made to the database.

import ipaddress                                  
                           
from psycopg2.extensions import register_adapter, AsIs
from psycopg2.extras import register_ipaddress    
import sqlalchemy as sa    
from sqlalchemy import orm    
from sqlalchemy.dialects import postgresql
    
engine = sa.create_engine('postgresql+psycopg2:///test')    


@sa.event.listens_for(engine, 'first_connect')    
def on_connect(dbapi_connection, connection_record):
    # Configure adapters so that psycopg2 handles Python ipaddress types
    register_adapter(ipaddress.IPv4Address, lambda a: AsIs(f"'{a}'"))    
    register_ipaddress()    
    
nw = ipaddress.ip_network('192.168.1.0/24')    
ips = map(ipaddress.ip_address, ['192.168.1.5', '192.168.0.5'])    
    
with orm.Session(engine) as session:    
    session.add_all([TableIpAddress(ipaddress=ip) for ip in ips])    
    session.commit()    
    
    res = session.execute(    
sa.select(TableIpAddress).where(TableIpAddress.ipaddress.op('<<')(nw))    
    ).scalars().all()
    for obj in res:
        print(obj.ipaddress)

outputs

192.168.1.5/32