我有一个定义的模型:
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。有没有办法将参数强制为适当的类型?
这可以使用 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)
outputs
192.168.1.5/32
本文内容由网友自发贡献,版权归原作者所有,本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容,请联系:hwhale#tublm.com(使用前将#替换为@)