<s> from multicorn import ForeignDataWrapper <EOL> from multicorn . utils import log_to_postgres as log2pg <EOL> from functools import partial <EOL> import httplib <EOL> import json <EOL> import logging <EOL> class ElasticsearchFDW ( ForeignDataWrapper ) : <EOL> def __init__ ( self , options , columns ) : <EOL> super ( ElasticsearchFDW , self ) . __init__ ( options , columns ) <EOL> self . host = options . get ( '<STR_LIT:host>' , '<STR_LIT:localhost>' ) <EOL> self . port = int ( options . get ( '<STR_LIT:port>' , '<STR_LIT>' ) ) <EOL> self . node = options . get ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . index = options . get ( '<STR_LIT:index>' , '<STR_LIT>' ) <EOL> self . columns = columns <EOL> def get_rel_size ( self , quals , columns ) : <EOL> """<STR_LIT>""" <EOL> conn = httplib . HTTPConnection ( self . host , self . port ) <EOL> conn . request ( "<STR_LIT:GET>" , "<STR_LIT>" % ( self . node , self . index ) ) <EOL> resp = conn . getresponse ( ) <EOL> if not <NUM_LIT:200> == resp . status : <EOL> return ( <NUM_LIT:0> , <NUM_LIT:0> ) <EOL> raw = resp . read ( ) <EOL> data = json . loads ( raw ) <EOL> return ( data [ '<STR_LIT:count>' ] , len ( columns ) * <NUM_LIT:100> ) <EOL> def execute ( self , quals , columns ) : <EOL> conn = httplib . HTTPConnection ( self . host , self . port ) <EOL> conn . request ( "<STR_LIT:GET>" , "<STR_LIT>" % ( self . node , self . index ) ) <EOL> resp = conn . getresponse ( ) <EOL> if not <NUM_LIT:200> == resp . status : <EOL> return ( <NUM_LIT:0> , <NUM_LIT:0> ) <EOL> raw = resp . read ( ) <EOL> data = json . loads ( raw ) <EOL> out = [ ] <EOL> for hit in data [ '<STR_LIT>' ] [ '<STR_LIT>' ] : <EOL> row = { } <EOL> for col in columns : <EOL> if col == '<STR_LIT:id>' : <EOL> row [ col ] = hit [ '<STR_LIT>' ] <EOL> elif col in hit [ '<STR_LIT>' ] : <EOL> row [ col ] = hit [ '<STR_LIT>' ] [ col ] <EOL> out . append ( row ) <EOL> return out <EOL> @ property <EOL> def rowid_column ( self ) : <EOL> """<STR_LIT>""" <EOL> return '<STR_LIT:id>' ; <EOL> def es_index ( self , id , values ) : <EOL> content = json . dumps ( values ) <EOL> conn = httplib . HTTPConnection ( self . host , self . port ) <EOL> conn . request ( "<STR_LIT>" , "<STR_LIT>" % ( self . node , self . index , id ) , content ) <EOL> resp = conn . getresponse ( ) <EOL> if not <NUM_LIT:200> == resp . status : <EOL> return <EOL> raw = resp . read ( ) <EOL> data = json . loads ( raw ) <EOL> return data <EOL> def insert ( self , new_values ) : <EOL> log2pg ( '<STR_LIT>' % new_values , logging . DEBUG ) <EOL> if not '<STR_LIT:id>' in new_values : <EOL> log2pg ( '<STR_LIT>' % new_values , logging . ERROR ) <EOL> id = new_values [ '<STR_LIT:id>' ] <EOL> new_values . pop ( '<STR_LIT:id>' , None ) <EOL> return self . es_index ( id , new_values ) <EOL> def update ( self , id , new_values ) : <EOL> new_values . pop ( '<STR_LIT:id>' , None ) <EOL> return self . es_index ( id , new_values ) <EOL> def delete ( self , id ) : <EOL> conn = httplib . HTTPConnection ( self . host , self . port ) <EOL> conn . request ( "<STR_LIT>" , "<STR_LIT>" % ( self . node , self . index , id ) ) <EOL> resp = conn . getresponse ( ) <EOL> if not <NUM_LIT:200> == resp . status : <EOL> log2pg ( '<STR_LIT>' % resp . read ( ) , logging . ERROR ) <EOL> return <EOL> raw = resp . read ( ) <EOL> return json . loads ( raw ) </s>
