如何格式化 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