diff --git a/elasticsearch/__init__.py b/elasticsearch/__init__.py index a396d0f1..ec904dba 100644 --- a/elasticsearch/__init__.py +++ b/elasticsearch/__init__.py @@ -20,6 +20,6 @@ from .connection_pool import ConnectionPool, ConnectionSelector, \ RoundRobinSelector from .serializer import JSONSerializer from .connection import Connection, RequestsHttpConnection, \ - Urllib3HttpConnection, MemcachedConnection, ThriftConnection + Urllib3HttpConnection from .exceptions import * diff --git a/elasticsearch/connection/__init__.py b/elasticsearch/connection/__init__.py index 51a5cded..c2f484a1 100644 --- a/elasticsearch/connection/__init__.py +++ b/elasticsearch/connection/__init__.py @@ -1,5 +1,3 @@ from .base import Connection from .http_requests import RequestsHttpConnection from .http_urllib3 import Urllib3HttpConnection -from .memcached import MemcachedConnection -from .thrift import ThriftConnection, THRIFT_AVAILABLE diff --git a/elasticsearch/connection/esthrift/Rest-remote b/elasticsearch/connection/esthrift/Rest-remote deleted file mode 100755 index e229ea59..00000000 --- a/elasticsearch/connection/esthrift/Rest-remote +++ /dev/null @@ -1,88 +0,0 @@ -#!/usr/bin/env python -# -# Autogenerated by Thrift Compiler (0.9.0) -# -# DO NOT EDIT UNLESS YOU ARE SURE THAT YOU KNOW WHAT YOU ARE DOING -# -# options string: py:new_style=true,utf8strings=true -# - -import sys -import pprint -from urlparse import urlparse -from thrift.transport import TTransport -from thrift.transport import TSocket -from thrift.transport import THttpClient -from thrift.protocol import TBinaryProtocol - -import Rest -from ttypes import * - -if len(sys.argv) <= 1 or sys.argv[1] == '--help': - print '' - print 'Usage: ' + sys.argv[0] + ' [-h host[:port]] [-u url] [-f[ramed]] function [arg1 [arg2...]]' - print '' - print 'Functions:' - print ' RestResponse execute(RestRequest request)' - print '' - sys.exit(0) - -pp = pprint.PrettyPrinter(indent = 2) -host = 'localhost' -port = 9090 -uri = '' -framed = False -http = False -argi = 1 - -if sys.argv[argi] == '-h': - parts = sys.argv[argi+1].split(':') - host = parts[0] - if len(parts) > 1: - port = int(parts[1]) - argi += 2 - -if sys.argv[argi] == '-u': - url = urlparse(sys.argv[argi+1]) - parts = url[1].split(':') - host = parts[0] - if len(parts) > 1: - port = int(parts[1]) - else: - port = 80 - uri = url[2] - if url[4]: - uri += '?%s' % url[4] - http = True - argi += 2 - -if sys.argv[argi] == '-f' or sys.argv[argi] == '-framed': - framed = True - argi += 1 - -cmd = sys.argv[argi] -args = sys.argv[argi+1:] - -if http: - transport = THttpClient.THttpClient(host, port, uri) -else: - socket = TSocket.TSocket(host, port) - if framed: - transport = TTransport.TFramedTransport(socket) - else: - transport = TTransport.TBufferedTransport(socket) -protocol = TBinaryProtocol.TBinaryProtocol(transport) -client = Rest.Client(protocol) -transport.open() - -if cmd == 'execute': - if len(args) != 1: - print 'execute requires 1 args' - sys.exit(1) - pp.pprint(client.execute(eval(args[0]),)) - -else: - print 'Unrecognized method %s' % cmd - sys.exit(1) - -transport.close() diff --git a/elasticsearch/connection/esthrift/Rest.py b/elasticsearch/connection/esthrift/Rest.py deleted file mode 100644 index d395d783..00000000 --- a/elasticsearch/connection/esthrift/Rest.py +++ /dev/null @@ -1,223 +0,0 @@ -# -# Autogenerated by Thrift Compiler (0.9.0) -# -# DO NOT EDIT UNLESS YOU ARE SURE THAT YOU KNOW WHAT YOU ARE DOING -# -# options string: py:new_style=true,utf8strings=true -# - -from thrift.Thrift import TType, TMessageType, TException, TApplicationException -from ttypes import * -from thrift.Thrift import TProcessor -from thrift.transport import TTransport -from thrift.protocol import TBinaryProtocol, TProtocol -try: - from thrift.protocol import fastbinary -except: - fastbinary = None - - -class Iface(object): - def execute(self, request): - """ - Parameters: - - request - """ - pass - - -class Client(Iface): - def __init__(self, iprot, oprot=None): - self._iprot = self._oprot = iprot - if oprot is not None: - self._oprot = oprot - self._seqid = 0 - - def execute(self, request): - """ - Parameters: - - request - """ - self.send_execute(request) - return self.recv_execute() - - def send_execute(self, request): - self._oprot.writeMessageBegin('execute', TMessageType.CALL, self._seqid) - args = execute_args() - args.request = request - args.write(self._oprot) - self._oprot.writeMessageEnd() - self._oprot.trans.flush() - - def recv_execute(self, ): - (fname, mtype, rseqid) = self._iprot.readMessageBegin() - if mtype == TMessageType.EXCEPTION: - x = TApplicationException() - x.read(self._iprot) - self._iprot.readMessageEnd() - raise x - result = execute_result() - result.read(self._iprot) - self._iprot.readMessageEnd() - if result.success is not None: - return result.success - raise TApplicationException(TApplicationException.MISSING_RESULT, "execute failed: unknown result"); - - -class Processor(Iface, TProcessor): - def __init__(self, handler): - self._handler = handler - self._processMap = {} - self._processMap["execute"] = Processor.process_execute - - def process(self, iprot, oprot): - (name, type, seqid) = iprot.readMessageBegin() - if name not in self._processMap: - iprot.skip(TType.STRUCT) - iprot.readMessageEnd() - x = TApplicationException(TApplicationException.UNKNOWN_METHOD, 'Unknown function %s' % (name)) - oprot.writeMessageBegin(name, TMessageType.EXCEPTION, seqid) - x.write(oprot) - oprot.writeMessageEnd() - oprot.trans.flush() - return - else: - self._processMap[name](self, seqid, iprot, oprot) - return True - - def process_execute(self, seqid, iprot, oprot): - args = execute_args() - args.read(iprot) - iprot.readMessageEnd() - result = execute_result() - result.success = self._handler.execute(args.request) - oprot.writeMessageBegin("execute", TMessageType.REPLY, seqid) - result.write(oprot) - oprot.writeMessageEnd() - oprot.trans.flush() - - -# HELPER FUNCTIONS AND STRUCTURES - -class execute_args(object): - """ - Attributes: - - request - """ - - thrift_spec = ( - None, # 0 - (1, TType.STRUCT, 'request', (RestRequest, RestRequest.thrift_spec), None, ), # 1 - ) - - def __init__(self, request=None,): - self.request = request - - def read(self, iprot): - if iprot.__class__ == TBinaryProtocol.TBinaryProtocolAccelerated and isinstance(iprot.trans, TTransport.CReadableTransport) and self.thrift_spec is not None and fastbinary is not None: - fastbinary.decode_binary(self, iprot.trans, (self.__class__, self.thrift_spec)) - return - iprot.readStructBegin() - while True: - (fname, ftype, fid) = iprot.readFieldBegin() - if ftype == TType.STOP: - break - if fid == 1: - if ftype == TType.STRUCT: - self.request = RestRequest() - self.request.read(iprot) - else: - iprot.skip(ftype) - else: - iprot.skip(ftype) - iprot.readFieldEnd() - iprot.readStructEnd() - - def write(self, oprot): - if oprot.__class__ == TBinaryProtocol.TBinaryProtocolAccelerated and self.thrift_spec is not None and fastbinary is not None: - oprot.trans.write(fastbinary.encode_binary(self, (self.__class__, self.thrift_spec))) - return - oprot.writeStructBegin('execute_args') - if self.request is not None: - oprot.writeFieldBegin('request', TType.STRUCT, 1) - self.request.write(oprot) - oprot.writeFieldEnd() - oprot.writeFieldStop() - oprot.writeStructEnd() - - def validate(self): - if self.request is None: - raise TProtocol.TProtocolException(message='Required field request is unset!') - return - - - def __repr__(self): - L = ['%s=%r' % (key, value) - for key, value in self.__dict__.iteritems()] - return '%s(%s)' % (self.__class__.__name__, ', '.join(L)) - - def __eq__(self, other): - return isinstance(other, self.__class__) and self.__dict__ == other.__dict__ - - def __ne__(self, other): - return not (self == other) - -class execute_result(object): - """ - Attributes: - - success - """ - - thrift_spec = ( - (0, TType.STRUCT, 'success', (RestResponse, RestResponse.thrift_spec), None, ), # 0 - ) - - def __init__(self, success=None,): - self.success = success - - def read(self, iprot): - if iprot.__class__ == TBinaryProtocol.TBinaryProtocolAccelerated and isinstance(iprot.trans, TTransport.CReadableTransport) and self.thrift_spec is not None and fastbinary is not None: - fastbinary.decode_binary(self, iprot.trans, (self.__class__, self.thrift_spec)) - return - iprot.readStructBegin() - while True: - (fname, ftype, fid) = iprot.readFieldBegin() - if ftype == TType.STOP: - break - if fid == 0: - if ftype == TType.STRUCT: - self.success = RestResponse() - self.success.read(iprot) - else: - iprot.skip(ftype) - else: - iprot.skip(ftype) - iprot.readFieldEnd() - iprot.readStructEnd() - - def write(self, oprot): - if oprot.__class__ == TBinaryProtocol.TBinaryProtocolAccelerated and self.thrift_spec is not None and fastbinary is not None: - oprot.trans.write(fastbinary.encode_binary(self, (self.__class__, self.thrift_spec))) - return - oprot.writeStructBegin('execute_result') - if self.success is not None: - oprot.writeFieldBegin('success', TType.STRUCT, 0) - self.success.write(oprot) - oprot.writeFieldEnd() - oprot.writeFieldStop() - oprot.writeStructEnd() - - def validate(self): - return - - - def __repr__(self): - L = ['%s=%r' % (key, value) - for key, value in self.__dict__.iteritems()] - return '%s(%s)' % (self.__class__.__name__, ', '.join(L)) - - def __eq__(self, other): - return isinstance(other, self.__class__) and self.__dict__ == other.__dict__ - - def __ne__(self, other): - return not (self == other) diff --git a/elasticsearch/connection/esthrift/__init__.py b/elasticsearch/connection/esthrift/__init__.py deleted file mode 100644 index bbc702ed..00000000 --- a/elasticsearch/connection/esthrift/__init__.py +++ /dev/null @@ -1 +0,0 @@ -__all__ = ['ttypes', 'constants', 'Rest'] diff --git a/elasticsearch/connection/esthrift/constants.py b/elasticsearch/connection/esthrift/constants.py deleted file mode 100644 index 4e9d2082..00000000 --- a/elasticsearch/connection/esthrift/constants.py +++ /dev/null @@ -1,11 +0,0 @@ -# -# Autogenerated by Thrift Compiler (0.9.0) -# -# DO NOT EDIT UNLESS YOU ARE SURE THAT YOU KNOW WHAT YOU ARE DOING -# -# options string: py:new_style=true,utf8strings=true -# - -from thrift.Thrift import TType, TMessageType, TException, TApplicationException -from ttypes import * - diff --git a/elasticsearch/connection/esthrift/ttypes.py b/elasticsearch/connection/esthrift/ttypes.py deleted file mode 100644 index 2a7038df..00000000 --- a/elasticsearch/connection/esthrift/ttypes.py +++ /dev/null @@ -1,412 +0,0 @@ -# -# Autogenerated by Thrift Compiler (0.9.0) -# -# DO NOT EDIT UNLESS YOU ARE SURE THAT YOU KNOW WHAT YOU ARE DOING -# -# options string: py:new_style=true,utf8strings=true -# - -from thrift.Thrift import TType, TMessageType, TException, TApplicationException - -from thrift.transport import TTransport -from thrift.protocol import TBinaryProtocol, TProtocol -try: - from thrift.protocol import fastbinary -except: - fastbinary = None - - -class Method(object): - GET = 0 - PUT = 1 - POST = 2 - DELETE = 3 - HEAD = 4 - OPTIONS = 5 - - _VALUES_TO_NAMES = { - 0: "GET", - 1: "PUT", - 2: "POST", - 3: "DELETE", - 4: "HEAD", - 5: "OPTIONS", - } - - _NAMES_TO_VALUES = { - "GET": 0, - "PUT": 1, - "POST": 2, - "DELETE": 3, - "HEAD": 4, - "OPTIONS": 5, - } - -class Status(object): - CONT = 100 - SWITCHING_PROTOCOLS = 101 - OK = 200 - CREATED = 201 - ACCEPTED = 202 - NON_AUTHORITATIVE_INFORMATION = 203 - NO_CONTENT = 204 - RESET_CONTENT = 205 - PARTIAL_CONTENT = 206 - MULTI_STATUS = 207 - MULTIPLE_CHOICES = 300 - MOVED_PERMANENTLY = 301 - FOUND = 302 - SEE_OTHER = 303 - NOT_MODIFIED = 304 - USE_PROXY = 305 - TEMPORARY_REDIRECT = 307 - BAD_REQUEST = 400 - UNAUTHORIZED = 401 - PAYMENT_REQUIRED = 402 - FORBIDDEN = 403 - NOT_FOUND = 404 - METHOD_NOT_ALLOWED = 405 - NOT_ACCEPTABLE = 406 - PROXY_AUTHENTICATION = 407 - REQUEST_TIMEOUT = 408 - CONFLICT = 409 - GONE = 410 - LENGTH_REQUIRED = 411 - PRECONDITION_FAILED = 412 - REQUEST_ENTITY_TOO_LARGE = 413 - REQUEST_URI_TOO_LONG = 414 - UNSUPPORTED_MEDIA_TYPE = 415 - REQUESTED_RANGE_NOT_SATISFIED = 416 - EXPECTATION_FAILED = 417 - UNPROCESSABLE_ENTITY = 422 - LOCKED = 423 - FAILED_DEPENDENCY = 424 - INTERNAL_SERVER_ERROR = 500 - NOT_IMPLEMENTED = 501 - BAD_GATEWAY = 502 - SERVICE_UNAVAILABLE = 503 - GATEWAY_TIMEOUT = 504 - INSUFFICIENT_STORAGE = 506 - - _VALUES_TO_NAMES = { - 100: "CONT", - 101: "SWITCHING_PROTOCOLS", - 200: "OK", - 201: "CREATED", - 202: "ACCEPTED", - 203: "NON_AUTHORITATIVE_INFORMATION", - 204: "NO_CONTENT", - 205: "RESET_CONTENT", - 206: "PARTIAL_CONTENT", - 207: "MULTI_STATUS", - 300: "MULTIPLE_CHOICES", - 301: "MOVED_PERMANENTLY", - 302: "FOUND", - 303: "SEE_OTHER", - 304: "NOT_MODIFIED", - 305: "USE_PROXY", - 307: "TEMPORARY_REDIRECT", - 400: "BAD_REQUEST", - 401: "UNAUTHORIZED", - 402: "PAYMENT_REQUIRED", - 403: "FORBIDDEN", - 404: "NOT_FOUND", - 405: "METHOD_NOT_ALLOWED", - 406: "NOT_ACCEPTABLE", - 407: "PROXY_AUTHENTICATION", - 408: "REQUEST_TIMEOUT", - 409: "CONFLICT", - 410: "GONE", - 411: "LENGTH_REQUIRED", - 412: "PRECONDITION_FAILED", - 413: "REQUEST_ENTITY_TOO_LARGE", - 414: "REQUEST_URI_TOO_LONG", - 415: "UNSUPPORTED_MEDIA_TYPE", - 416: "REQUESTED_RANGE_NOT_SATISFIED", - 417: "EXPECTATION_FAILED", - 422: "UNPROCESSABLE_ENTITY", - 423: "LOCKED", - 424: "FAILED_DEPENDENCY", - 500: "INTERNAL_SERVER_ERROR", - 501: "NOT_IMPLEMENTED", - 502: "BAD_GATEWAY", - 503: "SERVICE_UNAVAILABLE", - 504: "GATEWAY_TIMEOUT", - 506: "INSUFFICIENT_STORAGE", - } - - _NAMES_TO_VALUES = { - "CONT": 100, - "SWITCHING_PROTOCOLS": 101, - "OK": 200, - "CREATED": 201, - "ACCEPTED": 202, - "NON_AUTHORITATIVE_INFORMATION": 203, - "NO_CONTENT": 204, - "RESET_CONTENT": 205, - "PARTIAL_CONTENT": 206, - "MULTI_STATUS": 207, - "MULTIPLE_CHOICES": 300, - "MOVED_PERMANENTLY": 301, - "FOUND": 302, - "SEE_OTHER": 303, - "NOT_MODIFIED": 304, - "USE_PROXY": 305, - "TEMPORARY_REDIRECT": 307, - "BAD_REQUEST": 400, - "UNAUTHORIZED": 401, - "PAYMENT_REQUIRED": 402, - "FORBIDDEN": 403, - "NOT_FOUND": 404, - "METHOD_NOT_ALLOWED": 405, - "NOT_ACCEPTABLE": 406, - "PROXY_AUTHENTICATION": 407, - "REQUEST_TIMEOUT": 408, - "CONFLICT": 409, - "GONE": 410, - "LENGTH_REQUIRED": 411, - "PRECONDITION_FAILED": 412, - "REQUEST_ENTITY_TOO_LARGE": 413, - "REQUEST_URI_TOO_LONG": 414, - "UNSUPPORTED_MEDIA_TYPE": 415, - "REQUESTED_RANGE_NOT_SATISFIED": 416, - "EXPECTATION_FAILED": 417, - "UNPROCESSABLE_ENTITY": 422, - "LOCKED": 423, - "FAILED_DEPENDENCY": 424, - "INTERNAL_SERVER_ERROR": 500, - "NOT_IMPLEMENTED": 501, - "BAD_GATEWAY": 502, - "SERVICE_UNAVAILABLE": 503, - "GATEWAY_TIMEOUT": 504, - "INSUFFICIENT_STORAGE": 506, - } - - -class RestRequest(object): - """ - Attributes: - - method - - uri - - parameters - - headers - - body - """ - - thrift_spec = ( - None, # 0 - (1, TType.I32, 'method', None, None, ), # 1 - (2, TType.STRING, 'uri', None, None, ), # 2 - (3, TType.MAP, 'parameters', (TType.STRING,None,TType.STRING,None), None, ), # 3 - (4, TType.MAP, 'headers', (TType.STRING,None,TType.STRING,None), None, ), # 4 - (5, TType.STRING, 'body', None, None, ), # 5 - ) - - def __init__(self, method=None, uri=None, parameters=None, headers=None, body=None,): - self.method = method - self.uri = uri - self.parameters = parameters - self.headers = headers - self.body = body - - def read(self, iprot): - if iprot.__class__ == TBinaryProtocol.TBinaryProtocolAccelerated and isinstance(iprot.trans, TTransport.CReadableTransport) and self.thrift_spec is not None and fastbinary is not None: - fastbinary.decode_binary(self, iprot.trans, (self.__class__, self.thrift_spec)) - return - iprot.readStructBegin() - while True: - (fname, ftype, fid) = iprot.readFieldBegin() - if ftype == TType.STOP: - break - if fid == 1: - if ftype == TType.I32: - self.method = iprot.readI32(); - else: - iprot.skip(ftype) - elif fid == 2: - if ftype == TType.STRING: - self.uri = iprot.readString().decode('utf-8') - else: - iprot.skip(ftype) - elif fid == 3: - if ftype == TType.MAP: - self.parameters = {} - (_ktype1, _vtype2, _size0 ) = iprot.readMapBegin() - for _i4 in xrange(_size0): - _key5 = iprot.readString().decode('utf-8') - _val6 = iprot.readString().decode('utf-8') - self.parameters[_key5] = _val6 - iprot.readMapEnd() - else: - iprot.skip(ftype) - elif fid == 4: - if ftype == TType.MAP: - self.headers = {} - (_ktype8, _vtype9, _size7 ) = iprot.readMapBegin() - for _i11 in xrange(_size7): - _key12 = iprot.readString().decode('utf-8') - _val13 = iprot.readString().decode('utf-8') - self.headers[_key12] = _val13 - iprot.readMapEnd() - else: - iprot.skip(ftype) - elif fid == 5: - if ftype == TType.STRING: - self.body = iprot.readString(); - else: - iprot.skip(ftype) - else: - iprot.skip(ftype) - iprot.readFieldEnd() - iprot.readStructEnd() - - def write(self, oprot): - if oprot.__class__ == TBinaryProtocol.TBinaryProtocolAccelerated and self.thrift_spec is not None and fastbinary is not None: - oprot.trans.write(fastbinary.encode_binary(self, (self.__class__, self.thrift_spec))) - return - oprot.writeStructBegin('RestRequest') - if self.method is not None: - oprot.writeFieldBegin('method', TType.I32, 1) - oprot.writeI32(self.method) - oprot.writeFieldEnd() - if self.uri is not None: - oprot.writeFieldBegin('uri', TType.STRING, 2) - oprot.writeString(self.uri.encode('utf-8')) - oprot.writeFieldEnd() - if self.parameters is not None: - oprot.writeFieldBegin('parameters', TType.MAP, 3) - oprot.writeMapBegin(TType.STRING, TType.STRING, len(self.parameters)) - for kiter14,viter15 in self.parameters.items(): - oprot.writeString(kiter14.encode('utf-8')) - oprot.writeString(viter15.encode('utf-8')) - oprot.writeMapEnd() - oprot.writeFieldEnd() - if self.headers is not None: - oprot.writeFieldBegin('headers', TType.MAP, 4) - oprot.writeMapBegin(TType.STRING, TType.STRING, len(self.headers)) - for kiter16,viter17 in self.headers.items(): - oprot.writeString(kiter16.encode('utf-8')) - oprot.writeString(viter17.encode('utf-8')) - oprot.writeMapEnd() - oprot.writeFieldEnd() - if self.body is not None: - oprot.writeFieldBegin('body', TType.STRING, 5) - oprot.writeString(self.body) - oprot.writeFieldEnd() - oprot.writeFieldStop() - oprot.writeStructEnd() - - def validate(self): - if self.method is None: - raise TProtocol.TProtocolException(message='Required field method is unset!') - if self.uri is None: - raise TProtocol.TProtocolException(message='Required field uri is unset!') - return - - - def __repr__(self): - L = ['%s=%r' % (key, value) - for key, value in self.__dict__.iteritems()] - return '%s(%s)' % (self.__class__.__name__, ', '.join(L)) - - def __eq__(self, other): - return isinstance(other, self.__class__) and self.__dict__ == other.__dict__ - - def __ne__(self, other): - return not (self == other) - -class RestResponse(object): - """ - Attributes: - - status - - headers - - body - """ - - thrift_spec = ( - None, # 0 - (1, TType.I32, 'status', None, None, ), # 1 - (2, TType.MAP, 'headers', (TType.STRING,None,TType.STRING,None), None, ), # 2 - (3, TType.STRING, 'body', None, None, ), # 3 - ) - - def __init__(self, status=None, headers=None, body=None,): - self.status = status - self.headers = headers - self.body = body - - def read(self, iprot): - if iprot.__class__ == TBinaryProtocol.TBinaryProtocolAccelerated and isinstance(iprot.trans, TTransport.CReadableTransport) and self.thrift_spec is not None and fastbinary is not None: - fastbinary.decode_binary(self, iprot.trans, (self.__class__, self.thrift_spec)) - return - iprot.readStructBegin() - while True: - (fname, ftype, fid) = iprot.readFieldBegin() - if ftype == TType.STOP: - break - if fid == 1: - if ftype == TType.I32: - self.status = iprot.readI32(); - else: - iprot.skip(ftype) - elif fid == 2: - if ftype == TType.MAP: - self.headers = {} - (_ktype19, _vtype20, _size18 ) = iprot.readMapBegin() - for _i22 in xrange(_size18): - _key23 = iprot.readString().decode('utf-8') - _val24 = iprot.readString().decode('utf-8') - self.headers[_key23] = _val24 - iprot.readMapEnd() - else: - iprot.skip(ftype) - elif fid == 3: - if ftype == TType.STRING: - self.body = iprot.readString(); - else: - iprot.skip(ftype) - else: - iprot.skip(ftype) - iprot.readFieldEnd() - iprot.readStructEnd() - - def write(self, oprot): - if oprot.__class__ == TBinaryProtocol.TBinaryProtocolAccelerated and self.thrift_spec is not None and fastbinary is not None: - oprot.trans.write(fastbinary.encode_binary(self, (self.__class__, self.thrift_spec))) - return - oprot.writeStructBegin('RestResponse') - if self.status is not None: - oprot.writeFieldBegin('status', TType.I32, 1) - oprot.writeI32(self.status) - oprot.writeFieldEnd() - if self.headers is not None: - oprot.writeFieldBegin('headers', TType.MAP, 2) - oprot.writeMapBegin(TType.STRING, TType.STRING, len(self.headers)) - for kiter25,viter26 in self.headers.items(): - oprot.writeString(kiter25.encode('utf-8')) - oprot.writeString(viter26.encode('utf-8')) - oprot.writeMapEnd() - oprot.writeFieldEnd() - if self.body is not None: - oprot.writeFieldBegin('body', TType.STRING, 3) - oprot.writeString(self.body) - oprot.writeFieldEnd() - oprot.writeFieldStop() - oprot.writeStructEnd() - - def validate(self): - if self.status is None: - raise TProtocol.TProtocolException(message='Required field status is unset!') - return - - - def __repr__(self): - L = ['%s=%r' % (key, value) - for key, value in self.__dict__.iteritems()] - return '%s(%s)' % (self.__class__.__name__, ', '.join(L)) - - def __eq__(self, other): - return isinstance(other, self.__class__) and self.__dict__ == other.__dict__ - - def __ne__(self, other): - return not (self == other) diff --git a/elasticsearch/connection/memcached.py b/elasticsearch/connection/memcached.py deleted file mode 100644 index 3adc89f7..00000000 --- a/elasticsearch/connection/memcached.py +++ /dev/null @@ -1,83 +0,0 @@ -import time -try: - import simplejson as json -except ImportError: - import json - -from ..exceptions import TransportError, ConnectionError, ImproperlyConfigured -from ..compat import urlencode -from .pooling import PoolingConnection - -class MemcachedConnection(PoolingConnection): - """ - Client using the `pylibmc` python library to communicate with elasticsearch - using the memcached protocol. Requires plugin in the cluster. - - See https://github.com/elasticsearch/elasticsearch-transport-memcached for more details. - """ - transport_schema = 'memcached' - - method_map = { - 'PUT': 'set', - 'POST': 'set', - 'DELETE': 'delete', - 'HEAD': 'get', - 'GET': 'get', - } - - def __init__(self, host='localhost', port=11211, **kwargs): - try: - import pylibmc - except ImportError: - raise ImproperlyConfigured("You need to install pylibmc to use the MemcachedConnection class.") - super(MemcachedConnection, self).__init__(host=host, port=port, **kwargs) - self._make_connection = lambda: pylibmc.Client(['%s:%s' % (host, port)], behaviors={"tcp_nodelay": True}) - - def perform_request(self, method, url, params=None, body=None, timeout=None, ignore=()): - mc = self._get_connection() - url = self.url_prefix + url - if params: - url = '%s?%s' % (url, urlencode(params or {})) - full_url = self.host + url - - mc_method = self.method_map.get(method, 'get') - - start = time.time() - try: - status = 200 - if mc_method == 'set': - # no response from set commands - response = '' - if not json.dumps(mc.set(url, body)): - status = 500 - else: - response = mc.get(url) - - duration = time.time() - start - if response: - response = response.decode('utf-8') - except Exception as e: - self.log_request_fail(method, full_url, body, time.time() - start, exception=e) - raise ConnectionError('N/A', str(e), e) - finally: - self._release_connection(mc) - - # try not to load the json every time - if response and response[0] == '{' and ('"status"' in response or '"error"' in response): - data = json.loads(response) - if 'status' in data and isinstance(data['status'], int): - status = data['status'] - elif 'error' in data: - raise TransportError('N/A', data['error']) - - if not (200 <= status < 300) and status not in ignore: - self.log_request_fail(method, url, body, duration, status) - self._raise_error(status, response) - - self.log_request_success(method, full_url, url, body, status, - response, duration) - - return status, {}, response - - - diff --git a/elasticsearch/connection/pooling.py b/elasticsearch/connection/pooling.py index 764046e4..a7721207 100644 --- a/elasticsearch/connection/pooling.py +++ b/elasticsearch/connection/pooling.py @@ -6,6 +6,12 @@ from .base import Connection class PoolingConnection(Connection): + """ + Base connection class for connections that use libraries without thread + safety and no capacity for connection pooling. To use this just implement a + ``_make_connection`` method that constructs a new connection and returns + it. + """ def __init__(self, *args, **kwargs): self._free_connections = queue.Queue() super(PoolingConnection, self).__init__(*args, **kwargs) diff --git a/elasticsearch/connection/thrift.py b/elasticsearch/connection/thrift.py deleted file mode 100644 index 172868c8..00000000 --- a/elasticsearch/connection/thrift.py +++ /dev/null @@ -1,101 +0,0 @@ -from __future__ import absolute_import -from socket import timeout as SocketTimeout -from socket import error as SocketError -import time -import logging - -try: - from .esthrift import Rest - from .esthrift.ttypes import Method, RestRequest - - from thrift.transport import TTransport, TSocket, TSSLSocket - from thrift.protocol import TBinaryProtocol - from thrift.Thrift import TException - THRIFT_AVAILABLE = True -except ImportError: - THRIFT_AVAILABLE = False - -from ..exceptions import ConnectionError, ImproperlyConfigured, ConnectionTimeout -from .pooling import PoolingConnection - -logger = logging.getLogger('elasticsearch') - -class ThriftConnection(PoolingConnection): - """ - This connection class is deprecated and may be removed in future versions. - - Connection using the `thrift` protocol to communicate with elasticsearch. - - See https://github.com/elasticsearch/elasticsearch-transport-thrift for additional info. - """ - transport_schema = 'thrift' - - def __init__(self, host='localhost', port=9500, framed_transport=False, use_ssl=False, **kwargs): - """ - :arg framed_transport: use `TTransport.TFramedTransport` instead of - `TTransport.TBufferedTransport` - """ - if not THRIFT_AVAILABLE: - raise ImproperlyConfigured("Thrift is not available.") - - super(ThriftConnection, self).__init__(host=host, port=port, **kwargs) - self._framed_transport = framed_transport - self._tsocket_class = TSocket.TSocket - if use_ssl: - self._tsocket_class = TSSLSocket.TSSLSocket - self._tsocket_args = (host, port) - - def _make_connection(self): - socket = self._tsocket_class(*self._tsocket_args) - socket.setTimeout(self.timeout * 1000.0) - if self._framed_transport: - transport = TTransport.TFramedTransport(socket) - else: - transport = TTransport.TBufferedTransport(socket) - - protocol = TBinaryProtocol.TBinaryProtocolAccelerated(transport) - client = Rest.Client(protocol) - client.transport = transport - transport.open() - return client - - def perform_request(self, method, url, params=None, body=None, timeout=None, ignore=()): - request = RestRequest(method=Method._NAMES_TO_VALUES[method.upper()], uri=url, - parameters=params, body=body) - - start = time.time() - tclient = None - try: - tclient = self._get_connection() - response = tclient.execute(request) - duration = time.time() - start - except SocketTimeout as e: - self.log_request_fail(method, url, body, time.time() - start, exception=e) - raise ConnectionTimeout('TIMEOUT', str(e), e) - except (TException, SocketError) as e: - self.log_request_fail(method, url, body, time.time() - start, exception=e) - if tclient: - try: - # try closing transport socket - tclient.transport.close() - except Exception as e: - logger.warning( - 'Exception %s occured when closing a failed thrift connection.', - e, exc_info=True - ) - raise ConnectionError('N/A', str(e), e) - - self._release_connection(tclient) - - if not (200 <= response.status < 300) and response.status not in ignore: - self.log_request_fail(method, url, body, duration, response.status) - self._raise_error(response.status, response.body) - - self.log_request_success(method, url, url, body, response.status, - response.body, duration) - - headers = {} - if response.headers: - headers = dict((k.lower(), v) for k, v in response.headers.items()) - return response.status, headers, response.body or '' - diff --git a/test_elasticsearch/test_connection.py b/test_elasticsearch/test_connection.py index 7795b541..9900c4ae 100644 --- a/test_elasticsearch/test_connection.py +++ b/test_elasticsearch/test_connection.py @@ -6,29 +6,10 @@ from requests.auth import AuthBase from elasticsearch.exceptions import TransportError, ConflictError, RequestError, NotFoundError from elasticsearch.connection import RequestsHttpConnection, \ - Urllib3HttpConnection, THRIFT_AVAILABLE, ThriftConnection + Urllib3HttpConnection -from .test_cases import TestCase, SkipTest +from .test_cases import TestCase -class TestThriftConnection(TestCase): - def setUp(self): - if not THRIFT_AVAILABLE: - raise SkipTest('Thrift is not available.') - super(TestThriftConnection, self).setUp() - - def test_use_ssl_uses_ssl_socket(self): - from thrift.transport import TSSLSocket - con = ThriftConnection(use_ssl=True) - self.assertIs(con._tsocket_class, TSSLSocket.TSSLSocket) - - def test_use_normal_tsocket_by_default(self): - from thrift.transport import TSocket - con = ThriftConnection() - self.assertIs(con._tsocket_class, TSocket.TSocket) - - def test_timeout_set(self): - con = ThriftConnection(timeout=42) - self.assertEquals(42, con.timeout) class TestUrllib3Connection(TestCase): def test_timeout_set(self): diff --git a/test_elasticsearch/test_server/test_memcached.py b/test_elasticsearch/test_server/test_memcached.py deleted file mode 100644 index d08f1183..00000000 --- a/test_elasticsearch/test_server/test_memcached.py +++ /dev/null @@ -1,42 +0,0 @@ -# -*- coding: utf-8 -*- -from __future__ import unicode_literals - -from elasticsearch import Elasticsearch, MemcachedConnection, NotFoundError -from elasticsearch.transport import ADDRESS_RE - -from . import ElasticsearchTestCase -from ..test_cases import SkipTest - -class TestMemcachedConnection(ElasticsearchTestCase): - def setUp(self): - try: - import pylibmc - except ImportError: - raise SkipTest("No pylibmc.") - super(TestMemcachedConnection, self).setUp() - nodes = self.client.nodes.info() - for node_id, node_info in nodes["nodes"].items(): - if 'memcached_address' in node_info: - connection_info = ADDRESS_RE.search(node_info['memcached_address']).groupdict() - self.mc_client = Elasticsearch( - [connection_info], - connection_class=MemcachedConnection - ) - break - else: - raise SkipTest("No memcached plugin.") - - def test_index(self): - self.mc_client.index("test_index", "test_type", {"answer": 42}, id=1) - self.assertTrue(self.client.exists("test_index", doc_type="test_type", id=1)) - - def test_get(self): - self.client.index("test_index", "test_type", {"answer": 42}, id=1) - self.assertEquals({"answer": 42}, self.mc_client.get("test_index", doc_type="test_type", id=1)["_source"]) - - def test_unicode(self): - self.mc_client.index("test_index", "test_type", {"answer": "你好"}, id="你好") - self.assertEquals({"answer": "你好"}, self.mc_client.get("test_index", doc_type="test_type", id="你好")["_source"]) - - def test_missing(self): - self.assertRaises(NotFoundError, self.mc_client.get, "test_index", doc_type="test_type", id=42) diff --git a/test_elasticsearch/test_transport.py b/test_elasticsearch/test_transport.py index eec7d855..c6d4bd36 100644 --- a/test_elasticsearch/test_transport.py +++ b/test_elasticsearch/test_transport.py @@ -3,7 +3,7 @@ from __future__ import unicode_literals import time from elasticsearch.transport import Transport, get_host_info -from elasticsearch.connection import Connection, ThriftConnection +from elasticsearch.connection import Connection from elasticsearch.connection_pool import DummyConnectionPool from elasticsearch.exceptions import ConnectionError, ImproperlyConfigured @@ -61,7 +61,6 @@ class TestTransport(TestCase): def test_host_with_scheme_different_from_connection_fails(self): self.assertRaises(ImproperlyConfigured, Transport, [{'host': 'localhost', 'scheme': 'thrift'}]) - self.assertRaises(ImproperlyConfigured, Transport, [{'host': 'localhost', 'scheme': 'http'}], connection_class=ThriftConnection) def test_request_timeout_extracted_from_params_and_passed(self): t = Transport([{}], connection_class=DummyConnection)