如何格式化SQLAlchemy ORM的Postgresql CIDR包含(>>)查询

Question

我有一个模型是这样定义的：

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 )

我想做类似下面的事情：

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

这导致了错误信息的关键部分是：

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

这个字段实际上是一个CIDR字段。它似乎不知道怎么处理CIDR。有没有办法把这个参数强制转换成合适的类型？

Answer 1

这可以通过使用SQLAlchemy和psycopg2来实现。psycopg2需要一些设置，以便能够处理Python的ipaddress模块中的数据类型；这些设置可以在第一次连接到数据库时通过一个事件监听器来完成。

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)

输出结果

192.168.1.5/32