SQL式ステートメントでカスタム型を適用するにはどうすればよいですか？

Question

私はSQLAlchemy ORMマッパーを使ってアプリケーションにカスタムタイプを持っています。複雑なクエリの場合は、SQL式モジュールを使用する必要がありますが、これによりカスタムタイプの処理が不透明になります。 がORMを使用しないでない場合、マッピングに私のカスタムタイプを使用するようにSQLAlchemyに指示するにはどうすればいいですか？

以下は、この問題を示す簡単な例です。

最初のクエリが機能することに注意してください。ただし、カスタムタイプが定義されていても、最初にstrをPythonに、INETをPostgreSQLに手動でキャストしなければなりません。

私は、SQL式モジュールがORMの1つ上に定義されているため、カスタムタイプを認識していないことを理解します。しかし、どうやってそのカスタムタイプをSQLレイヤーに配線して、タイプと値の使用をより透明にする方法がないのだろうかと思います。また、どのタイプのSAが使用されていても、カスタムタイプで定義されている操作（クリーンアップなど）が一貫して適用されることを保証します。

from sqlalchemy.orm import sessionmaker 
from sqlalchemy.sql.expression import any_ 
from sqlalchemy.types import TypeDecorator 

Base = declarative_base() 


class PgIpInterface(TypeDecorator): 
    """ 
    A codec for :py:mod:`ipaddress` interfaces. 
    """ 

    impl = INET 

    def process_bind_param(self, value, dialect): 
     return str(value) if value else None 

    def process_result_value(self, value, dialect): 
     return ip_interface(value) if value else None 

    def process_literal_param(self, value, dialect): 
     raise NotImplementedError('Not yet implemented') 


class Network(Base): 
    __tablename__ = 'example_table' 
    cidr = Column(PgIpInterface, primary_key=True) 


def execute(query): 
    import logging 
    LOG = logging.getLogger() 
    try: 
     print(query) 
     print(query.all()) 
    except: 
     LOG.exception('!!! failed') 


engine = create_engine('postgresql://malbert@/malbert') 
Base.metadata.create_all(engine) 
Session = sessionmaker(bind=engine) 

session = Session() 

ranges = [ 
    ip_interface('192.168.1.0/24'), 
    ip_interface('192.168.3.0/24'), 
] 


# Query with manual casting 
print(' Manual Casting via "str" '.center(80, '-')) 
arr = array([cast(str(_), INET) for _ in ranges]) 
query1 = session.query(Network).filter(Network.cidr.op("<<=")(any_(arr))) 
execute(query1) 


print(' Manual Casting '.center(80, '-')) 
arr = array([cast(_, INET) for _ in ranges]) 
query2 = session.query(Network).filter(Network.cidr.op("<<=")(any_(arr))) 
execute(query2) 


# Query without casting 
print(' No Casting '.center(80, '-')) 
query3 = session.query(Network).filter(Network.cidr.op("<<=")(any_(ranges))) 
execute(query3) 

Answer 1

単にカスタム型にキャスト、あなたの2番目のクエリを動作させるために：あなたの第三問合せ作業を行うには

arr = array([cast(_, PgIpInterface) for _ in ranges]) 

、あなたはpsycopg2で、より深い1つのレベルを移動する必要があります。 psycopg2はipaddressタイプの場合はbuiltin supportですが、残念ながら不完全なようです。 （ipaddress型は明示的なキャストせずに文字列に変換されます。）

register_ipaddress() # register ipaddress handling globally 
arr = [ip_interface('192.168.1.0/24'), ip_interface('192.168.3.0/24')] 
session.query(Foo).filter(Foo.cidr.op("<<=")(any_(arr))).all() 

これはoperator does not exist: inet <<= textエラーで失敗し

WHERE foo.cidr <<= ANY (ARRAY['192.168.1.0/24', '192.168.3.0/24']) 

のようなものをレンダリングします。幸いにも、それを修正するのは簡単です。私達はちょうどregister_ipaddressに自分自身を書き換えます：

import ipaddress 

from psycopg2.extensions import (
    AsIs, 
    new_array_type, 
    new_type, 
    register_adapter, 
    register_type 
) 


def register_ipaddress(): 
    def cast_interface(s, cur=None): 
     if s is None: 
      return None 
     return ipaddress.ip_interface(s) 
    inet = new_type((869,), 'INET', cast_interface) 
    ainet = new_array_type((1041,), 'INET[]', inet) 

    def cast_network(s, cur=None): 
     if s is None: 
      return None 
     return ipaddress.ip_network(s) 
    cidr = new_type((650,), 'CIDR', cast_network) 
    acidr = new_array_type((651,), 'CIDR[]', cidr) 

    for caster in [inet, ainet, cidr, acidr]: 
     register_type(caster) 

    def adapt_interface(obj): 
     return AsIs("'{}'::inet".format(obj)) 
    for t in [ipaddress.IPv4Interface, ipaddress.IPv6Interface]: 
     register_adapter(t, adapt_interface) 

    def adapt_network(obj): 
     return AsIs("'{}'::cidr".format(obj)) 
    for t in [ipaddress.IPv4Network, ipaddress.IPv6Network]: 
     register_adapter(t, adapt_network) 

これは

WHERE foo.cidr <<= ANY (ARRAY['192.168.1.0/24'::inet, '192.168.3.0/24'::inet]) 

ノートで

arr = array([ip_interface...]) 

と

arr = [ip_interface...] 

を使用しての違いのようなクエリをレンダリングします前者の場合、配列はSQLAlchemyによって処理されるため、nの項目のリストにはnのバインドされたパラメータがあります。後者の場合、配列はpsycopg2で処理されるため、配列全体に対して1つのバインドされたパラメータが得られます。