如何格式化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 需要一些配置来处理来自 Pythonipaddress模块的类型；这可以在第一次连接到数据库时触发的事件侦听器中完成。

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