Fix the tests and make SQLAlchemy 1.2.8 compatible

.travis.yml

@@ -7,17 +7,14 @@ matrix:
  # One build pulls latest versions dynamically
  - python: 3.6
  env: CDH=cdh5 CDH_VERSION=5 PRESTO=RELEASE SQLALCHEMY=sqlalchemy
- # Others use pinned versions.
+ - python: 3.6
+ env: CDH=cdh5 CDH_VERSION=5.10.1 PRESTO=0.147 SQLALCHEMY=sqlalchemy==1.2.8
  - python: 3.5
- env: CDH=cdh5 CDH_VERSION=5.10.1 PRESTO=0.147 SQLALCHEMY=sqlalchemy==1.0.12
+ env: CDH=cdh5 CDH_VERSION=5.10.1 PRESTO=0.147 SQLALCHEMY=sqlalchemy==1.2.8
  - python: 3.4
- env: CDH=cdh5 CDH_VERSION=5.10.1 PRESTO=0.147 SQLALCHEMY=sqlalchemy==1.0.12
- - python: 2.7
- env: CDH=cdh5 CDH_VERSION=5.10.1 PRESTO=0.147 SQLALCHEMY=sqlalchemy==1.0.12
- # stale stuff we're still using / supporting
+ env: CDH=cdh5 CDH_VERSION=5.10.1 PRESTO=0.147 SQLALCHEMY=sqlalchemy==1.2.8
  - python: 2.7
- env: CDH=cdh5 CDH_VERSION=5.10.1 PRESTO=0.147 SQLALCHEMY=sqlalchemy==0.8.7
- # exclude: python 3 against old libries
+ env: CDH=cdh5 CDH_VERSION=5.10.1 PRESTO=0.147 SQLALCHEMY=sqlalchemy==1.2.8
 install:
  - ./scripts/travis-install.sh
  - pip install codecov

README.rst

@@ -1,5 +1,6 @@
 .. image:: https://travis-ci.org/dropbox/PyHive.svg?branch=master
  :target: https://travis-ci.org/dropbox/PyHive
+.. image:: https://img.shields.io/codecov/c/github/dropbox/PyHive.svg
 
 ======
 PyHive
@@ -100,9 +101,6 @@ PyHive works with
 - Python 2.7 / Python 3
 - For Presto: Presto install
 - For Hive: `HiveServer2 <https://cwiki.apache.org/confluence/display/Hive/Setting+up+HiveServer2>`_ daemon
-- For Python 3 + Hive + SASL, you currently need to install an unreleased version of ``thrift_sasl``
- (``pip install git+https://github.com/cloudera/thrift_sasl``).
- At the time of writing, the latest version of ``thrift_sasl`` was 0.2.1.
 
 Changelog
 =========
@@ -137,3 +135,10 @@ Run the following in an environment with Hive/Presto::
 
 WARNING: This drops/creates tables named ``one_row``, ``one_row_complex``, and ``many_rows``, plus a
 database called ``pyhive_test_database``.
+
+Updating TCLIService
+====================
+
+The TCLIService module is autogenerated using a ``TCLIService.thrift`` file. To update it, the
+``generate.py`` file can be used: ``python generate.py <TCLIServiceURL>``. When left blank, the
+version for Hive 2.3 will be downloaded.

dev_requirements.txt

@@ -11,7 +11,6 @@ pytest-timeout==1.2.0
 # actual dependencies: let things break if a package changes
 requests>=1.0.0
 sasl>=0.2.1
-sqlalchemy>=0.8.7
 thrift>=0.10.0
 #thrift_sasl>=0.1.0
 git+https://github.com/cloudera/thrift_sasl # Using master branch in order to get Python 3 SASL patches

generate.py

@@ -0,0 +1,52 @@
+"""
+This file can be used to generate a new version of the TCLIService package
+using a TCLIService.thrift URL.
+
+If no URL is specified, the file for Hive 2.3 will be downloaded.
+
+Usage:
+
+ python generate.py THRIFT_URL
+
+or
+
+ python generate.py
+"""
+import shutil
+import sys
+from os import path
+from urllib.request import urlopen
+import subprocess
+
+here = path.abspath(path.dirname(__file__))
+
+PACKAGE = 'TCLIService'
+GENERATED = 'gen-py'
+
+HIVE_SERVER2_URL = \
+ 'https://raw.githubusercontent.com/apache/hive/branch-2.3/service-rpc/if/TCLIService.thrift'
+
+
+def save_url(url):
+ data = urlopen(url).read()
+ file_path = path.join(here, url.rsplit('/', 1)[-1])
+ with open(file_path, 'wb') as f:
+ f.write(data)
+
+
+def main(hive_server2_url):
+ save_url(hive_server2_url)
+ hive_server2_path = path.join(here, hive_server2_url.rsplit('/', 1)[-1])
+
+ subprocess.call(['thrift', '-r', '--gen', 'py', hive_server2_path])
+ shutil.move(path.join(here, PACKAGE), path.join(here, PACKAGE + '.old'))
+ shutil.move(path.join(here, GENERATED, PACKAGE), path.join(here, PACKAGE))
+ shutil.rmtree(path.join(here, PACKAGE + '.old'))
+
+
+if __name__ == '__main__':
+ if len(sys.argv) > 1:
+ url = sys.argv[1]
+ else:
+ url = HIVE_SERVER2_URL
+ main(url)

pyhive/common.py

@@ -8,14 +8,14 @@
 from builtins import bytes
 from builtins import int
 from builtins import object
-from builtins import range
 from builtins import str
 from past.builtins import basestring
 from pyhive import exc
 import abc
 import collections
 import time
 from future.utils import with_metaclass
+from itertools import islice
 
 
 class DBAPICursor(with_metaclass(abc.ABCMeta, object)):
@@ -124,14 +124,7 @@ def fetchmany(self, size=None):
  """
  if size is None:
  size = self.arraysize
- result = []
- for _ in range(size):
- one = self.fetchone()
- if one is None:
- break
- else:
- result.append(one)
- return result
+ return list(islice(iter(self.fetchone, None), size))
 
  def fetchall(self):
  """Fetch all (remaining) rows of a query result, returning them as a sequence of sequences
@@ -140,14 +133,7 @@ def fetchall(self):
  An :py:class:`~pyhive.exc.Error` (or subclass) exception is raised if the previous call to
  :py:meth:`execute` did not produce any result set or no call was issued yet.
  """
- result = []
- while True:
- one = self.fetchone()
- if one is None:
- break
- else:
- result.append(one)
- return result
+ return list(iter(self.fetchone, None))
 
  @property
  def arraysize(self):

pyhive/hive.py

@@ -7,6 +7,11 @@
 
 from __future__ import absolute_import
 from __future__ import unicode_literals
+
+import datetime
+import re
+from decimal import Decimal
+
 from TCLIService import TCLIService
 from TCLIService import constants
 from TCLIService import ttypes
@@ -31,6 +36,31 @@
 
 _logger = logging.getLogger(__name__)
 
+_TIMESTAMP_PATTERN = re.compile(r'(\d+-\d+-\d+ \d+:\d+:\d+(\.\d{,6})?)')
+
+
+def _parse_timestamp(value):
+ if value:
+ match = _TIMESTAMP_PATTERN.match(value)
+ if match:
+ if match.group(2):
+ format = '%Y-%m-%d %H:%M:%S.%f'
+ # use the pattern to truncate the value
+ value = match.group()
+ else:
+ format = '%Y-%m-%d %H:%M:%S'
+ value = datetime.datetime.strptime(value, format)
+ else:
+ raise Exception(
+ 'Cannot convert "{}" into a datetime'.format(value))
+ else:
+ value = None
+ return value
+
+
+TYPES_CONVERTER = {"DECIMAL_TYPE": Decimal,
+ "TIMESTAMP_TYPE": _parse_timestamp}
+
 
 class HiveParamEscaper(common.ParamEscaper):
  def escape_string(self, item):
@@ -177,6 +207,14 @@ def sasl_factory():
  self._transport.close()
  raise
 
+ def __enter__(self):
+ """Transport should already be opened by __init__"""
+ return self
+
+ def __exit__(self, exc_type, exc_val, exc_tb):
+ """Call close"""
+ self.close()
+
  def close(self):
  """Close the underlying session and Thrift transport"""
  req = ttypes.TCloseSessionReq(sessionHandle=self._sessionHandle)
@@ -215,7 +253,7 @@ class Cursor(common.DBAPICursor):
  def __init__(self, connection, arraysize=1000):
  self._operationHandle = None
  super(Cursor, self).__init__()
- self.arraysize = arraysize
+ self._arraysize = arraysize
  self._connection = connection
 
  def _reset_state(self):
@@ -230,6 +268,19 @@ def _reset_state(self):
  finally:
  self._operationHandle = None
 
+ @property
+ def arraysize(self):
+ return self._arraysize
+
+ @arraysize.setter
+ def arraysize(self, value):
+ """Array size cannot be None, and should be an integer"""
+ default_arraysize = 1000
+ try:
+ self._arraysize = int(value) or default_arraysize
+ except TypeError:
+ self._arraysize = default_arraysize
+
  @property
  def description(self):
  """This read-only attribute is a sequence of 7-item sequences.
@@ -273,6 +324,12 @@ def description(self):
  ))
  return self._description
 
+ def __enter__(self):
+ return self
+
+ def __exit__(self, exc_type, exc_val, exc_tb):
+ self.close()
+
  def close(self):
  """Close the operation handle"""
  self._reset_state()
@@ -320,8 +377,10 @@ def _fetch_more(self):
  )
  response = self._connection.client.FetchResults(req)
  _check_status(response)
+ schema = self.description
  assert not response.results.rows, 'expected data in columnar format'
- columns = map(_unwrap_column, response.results.columns)
+ columns = [_unwrap_column(col, col_schema[1]) for col, col_schema in
+ zip(response.results.columns, schema)]
  new_data = list(zip(*columns))
  self._data += new_data
  # response.hasMoreRows seems to always be False, so we instead check the number of rows
@@ -402,7 +461,7 @@ def fetch_logs(self):
 #
 
 
-def _unwrap_column(col):
+def _unwrap_column(col, type_=None):
  """Return a list of raw values from a TColumn instance."""
  for attr, wrapper in iteritems(col.__dict__):
  if wrapper is not None:
@@ -414,6 +473,9 @@ def _unwrap_column(col):
  for b in range(8):
  if byte & (1 << b):
  result[i * 8 + b] = None
+ converter = TYPES_CONVERTER.get(type_, None)
+ if converter and type_:
+ result = [converter(row) if row else row for row in result]
  return result
  raise DataError("Got empty column value {}".format(col)) # pragma: no cover
 

.travis.yml

@@ -7,17 +7,14 @@ matrix:
  # One build pulls latest versions dynamically
  - python: 3.6
  env: CDH=cdh5 CDH_VERSION=5 PRESTO=RELEASE SQLALCHEMY=sqlalchemy
- # Others use pinned versions.
+ - python: 3.6
+ env: CDH=cdh5 CDH_VERSION=5.10.1 PRESTO=0.147 SQLALCHEMY=sqlalchemy==1.2.8
  - python: 3.5
- env: CDH=cdh5 CDH_VERSION=5.10.1 PRESTO=0.147 SQLALCHEMY=sqlalchemy==1.0.12
+ env: CDH=cdh5 CDH_VERSION=5.10.1 PRESTO=0.147 SQLALCHEMY=sqlalchemy==1.2.8
  - python: 3.4
- env: CDH=cdh5 CDH_VERSION=5.10.1 PRESTO=0.147 SQLALCHEMY=sqlalchemy==1.0.12
- - python: 2.7
- env: CDH=cdh5 CDH_VERSION=5.10.1 PRESTO=0.147 SQLALCHEMY=sqlalchemy==1.0.12
- # stale stuff we're still using / supporting
+ env: CDH=cdh5 CDH_VERSION=5.10.1 PRESTO=0.147 SQLALCHEMY=sqlalchemy==1.2.8
  - python: 2.7
- env: CDH=cdh5 CDH_VERSION=5.10.1 PRESTO=0.147 SQLALCHEMY=sqlalchemy==0.8.7
- # exclude: python 3 against old libries
+ env: CDH=cdh5 CDH_VERSION=5.10.1 PRESTO=0.147 SQLALCHEMY=sqlalchemy==1.2.8
 install:
  - ./scripts/travis-install.sh
  - pip install codecov

README.rst

@@ -1,5 +1,6 @@
 .. image:: https://travis-ci.org/dropbox/PyHive.svg?branch=master
  :target: https://travis-ci.org/dropbox/PyHive
+.. image:: https://img.shields.io/codecov/c/github/dropbox/PyHive.svg
 
 ======
 PyHive
@@ -100,9 +101,6 @@ PyHive works with
 - Python 2.7 / Python 3
 - For Presto: Presto install
 - For Hive: `HiveServer2 <https://cwiki.apache.org/confluence/display/Hive/Setting+up+HiveServer2>`_ daemon
-- For Python 3 + Hive + SASL, you currently need to install an unreleased version of ``thrift_sasl``
- (``pip install git+https://github.com/cloudera/thrift_sasl``).
- At the time of writing, the latest version of ``thrift_sasl`` was 0.2.1.
 
 Changelog
 =========
@@ -137,3 +135,10 @@ Run the following in an environment with Hive/Presto::
 
 WARNING: This drops/creates tables named ``one_row``, ``one_row_complex``, and ``many_rows``, plus a
 database called ``pyhive_test_database``.
+
+Updating TCLIService
+====================
+
+The TCLIService module is autogenerated using a ``TCLIService.thrift`` file. To update it, the
+``generate.py`` file can be used: ``python generate.py <TCLIServiceURL>``. When left blank, the
+version for Hive 2.3 will be downloaded.

dev_requirements.txt

@@ -11,7 +11,6 @@ pytest-timeout==1.2.0
 # actual dependencies: let things break if a package changes
 requests>=1.0.0
 sasl>=0.2.1
-sqlalchemy>=0.8.7
 thrift>=0.10.0
 #thrift_sasl>=0.1.0
 git+https://github.com/cloudera/thrift_sasl # Using master branch in order to get Python 3 SASL patches

generate.py

@@ -0,0 +1,52 @@
+"""
+This file can be used to generate a new version of the TCLIService package
+using a TCLIService.thrift URL.
+
+If no URL is specified, the file for Hive 2.3 will be downloaded.
+
+Usage:
+
+ python generate.py THRIFT_URL
+
+or
+
+ python generate.py
+"""
+import shutil
+import sys
+from os import path
+from urllib.request import urlopen
+import subprocess
+
+here = path.abspath(path.dirname(__file__))
+
+PACKAGE = 'TCLIService'
+GENERATED = 'gen-py'
+
+HIVE_SERVER2_URL = \
+ 'https://raw.githubusercontent.com/apache/hive/branch-2.3/service-rpc/if/TCLIService.thrift'
+
+
+def save_url(url):
+ data = urlopen(url).read()
+ file_path = path.join(here, url.rsplit('/', 1)[-1])
+ with open(file_path, 'wb') as f:
+ f.write(data)
+
+
+def main(hive_server2_url):
+ save_url(hive_server2_url)
+ hive_server2_path = path.join(here, hive_server2_url.rsplit('/', 1)[-1])
+
+ subprocess.call(['thrift', '-r', '--gen', 'py', hive_server2_path])
+ shutil.move(path.join(here, PACKAGE), path.join(here, PACKAGE + '.old'))
+ shutil.move(path.join(here, GENERATED, PACKAGE), path.join(here, PACKAGE))
+ shutil.rmtree(path.join(here, PACKAGE + '.old'))
+
+
+if __name__ == '__main__':
+ if len(sys.argv) > 1:
+ url = sys.argv[1]
+ else:
+ url = HIVE_SERVER2_URL
+ main(url)

pyhive/common.py

@@ -8,14 +8,14 @@
 from builtins import bytes
 from builtins import int
 from builtins import object
-from builtins import range
 from builtins import str
 from past.builtins import basestring
 from pyhive import exc
 import abc
 import collections
 import time
 from future.utils import with_metaclass
+from itertools import islice
 
 
 class DBAPICursor(with_metaclass(abc.ABCMeta, object)):
@@ -124,14 +124,7 @@ def fetchmany(self, size=None):
  """
  if size is None:
  size = self.arraysize
- result = []
- for _ in range(size):
- one = self.fetchone()
- if one is None:
- break
- else:
- result.append(one)
- return result
+ return list(islice(iter(self.fetchone, None), size))
 
  def fetchall(self):
  """Fetch all (remaining) rows of a query result, returning them as a sequence of sequences
@@ -140,14 +133,7 @@ def fetchall(self):
  An :py:class:`~pyhive.exc.Error` (or subclass) exception is raised if the previous call to
  :py:meth:`execute` did not produce any result set or no call was issued yet.
  """
- result = []
- while True:
- one = self.fetchone()
- if one is None:
- break
- else:
- result.append(one)
- return result
+ return list(iter(self.fetchone, None))
 
  @property
  def arraysize(self):

pyhive/hive.py

@@ -7,6 +7,11 @@
 
 from __future__ import absolute_import
 from __future__ import unicode_literals
+
+import datetime
+import re
+from decimal import Decimal
+
 from TCLIService import TCLIService
 from TCLIService import constants
 from TCLIService import ttypes
@@ -31,6 +36,31 @@
 
 _logger = logging.getLogger(__name__)
 
+_TIMESTAMP_PATTERN = re.compile(r'(\d+-\d+-\d+ \d+:\d+:\d+(\.\d{,6})?)')
+
+
+def _parse_timestamp(value):
+ if value:
+ match = _TIMESTAMP_PATTERN.match(value)
+ if match:
+ if match.group(2):
+ format = '%Y-%m-%d %H:%M:%S.%f'
+ # use the pattern to truncate the value
+ value = match.group()
+ else:
+ format = '%Y-%m-%d %H:%M:%S'
+ value = datetime.datetime.strptime(value, format)
+ else:
+ raise Exception(
+ 'Cannot convert "{}" into a datetime'.format(value))
+ else:
+ value = None
+ return value
+
+
+TYPES_CONVERTER = {"DECIMAL_TYPE": Decimal,
+ "TIMESTAMP_TYPE": _parse_timestamp}
+
 
 class HiveParamEscaper(common.ParamEscaper):
  def escape_string(self, item):
@@ -177,6 +207,14 @@ def sasl_factory():
  self._transport.close()
  raise
 
+ def __enter__(self):
+ """Transport should already be opened by __init__"""
+ return self
+
+ def __exit__(self, exc_type, exc_val, exc_tb):
+ """Call close"""
+ self.close()
+
  def close(self):
  """Close the underlying session and Thrift transport"""
  req = ttypes.TCloseSessionReq(sessionHandle=self._sessionHandle)
@@ -215,7 +253,7 @@ class Cursor(common.DBAPICursor):
  def __init__(self, connection, arraysize=1000):
  self._operationHandle = None
  super(Cursor, self).__init__()
- self.arraysize = arraysize
+ self._arraysize = arraysize
  self._connection = connection
 
  def _reset_state(self):
@@ -230,6 +268,19 @@ def _reset_state(self):
  finally:
  self._operationHandle = None
 
+ @property
+ def arraysize(self):
+ return self._arraysize
+
+ @arraysize.setter
+ def arraysize(self, value):
+ """Array size cannot be None, and should be an integer"""
+ default_arraysize = 1000
+ try:
+ self._arraysize = int(value) or default_arraysize
+ except TypeError:
+ self._arraysize = default_arraysize
+
  @property
  def description(self):
  """This read-only attribute is a sequence of 7-item sequences.
@@ -273,6 +324,12 @@ def description(self):
  ))
  return self._description
 
+ def __enter__(self):
+ return self
+
+ def __exit__(self, exc_type, exc_val, exc_tb):
+ self.close()
+
  def close(self):
  """Close the operation handle"""
  self._reset_state()
@@ -320,8 +377,10 @@ def _fetch_more(self):
  )
  response = self._connection.client.FetchResults(req)
  _check_status(response)
+ schema = self.description
  assert not response.results.rows, 'expected data in columnar format'
- columns = map(_unwrap_column, response.results.columns)
+ columns = [_unwrap_column(col, col_schema[1]) for col, col_schema in
+ zip(response.results.columns, schema)]
  new_data = list(zip(*columns))
  self._data += new_data
  # response.hasMoreRows seems to always be False, so we instead check the number of rows
@@ -402,7 +461,7 @@ def fetch_logs(self):
 #
 
 
-def _unwrap_column(col):
+def _unwrap_column(col, type_=None):
  """Return a list of raw values from a TColumn instance."""
  for attr, wrapper in iteritems(col.__dict__):
  if wrapper is not None:
@@ -414,6 +473,9 @@ def _unwrap_column(col):
  for b in range(8):
  if byte & (1 << b):
  result[i * 8 + b] = None
+ converter = TYPES_CONVERTER.get(type_, None)
+ if converter and type_:
+ result = [converter(row) if row else row for row in result]
  return result
  raise DataError("Got empty column value {}".format(col)) # pragma: no cover
 

.travis.yml

@@ -7,17 +7,14 @@ matrix:
  # One build pulls latest versions dynamically
  - python: 3.6
  env: CDH=cdh5 CDH_VERSION=5 PRESTO=RELEASE SQLALCHEMY=sqlalchemy
- # Others use pinned versions.
+ - python: 3.6
+ env: CDH=cdh5 CDH_VERSION=5.10.1 PRESTO=0.147 SQLALCHEMY=sqlalchemy==1.2.8
  - python: 3.5
- env: CDH=cdh5 CDH_VERSION=5.10.1 PRESTO=0.147 SQLALCHEMY=sqlalchemy==1.0.12
+ env: CDH=cdh5 CDH_VERSION=5.10.1 PRESTO=0.147 SQLALCHEMY=sqlalchemy==1.2.8
  - python: 3.4
- env: CDH=cdh5 CDH_VERSION=5.10.1 PRESTO=0.147 SQLALCHEMY=sqlalchemy==1.0.12
- - python: 2.7
- env: CDH=cdh5 CDH_VERSION=5.10.1 PRESTO=0.147 SQLALCHEMY=sqlalchemy==1.0.12
- # stale stuff we're still using / supporting
+ env: CDH=cdh5 CDH_VERSION=5.10.1 PRESTO=0.147 SQLALCHEMY=sqlalchemy==1.2.8
  - python: 2.7
- env: CDH=cdh5 CDH_VERSION=5.10.1 PRESTO=0.147 SQLALCHEMY=sqlalchemy==0.8.7
- # exclude: python 3 against old libries
+ env: CDH=cdh5 CDH_VERSION=5.10.1 PRESTO=0.147 SQLALCHEMY=sqlalchemy==1.2.8
 install:
  - ./scripts/travis-install.sh
  - pip install codecov

README.rst

@@ -1,5 +1,6 @@
 .. image:: https://travis-ci.org/dropbox/PyHive.svg?branch=master
  :target: https://travis-ci.org/dropbox/PyHive
+.. image:: https://img.shields.io/codecov/c/github/dropbox/PyHive.svg
 
 ======
 PyHive
@@ -100,9 +101,6 @@ PyHive works with
 - Python 2.7 / Python 3
 - For Presto: Presto install
 - For Hive: `HiveServer2 <https://cwiki.apache.org/confluence/display/Hive/Setting+up+HiveServer2>`_ daemon
-- For Python 3 + Hive + SASL, you currently need to install an unreleased version of ``thrift_sasl``
- (``pip install git+https://github.com/cloudera/thrift_sasl``).
- At the time of writing, the latest version of ``thrift_sasl`` was 0.2.1.
 
 Changelog
 =========
@@ -137,3 +135,10 @@ Run the following in an environment with Hive/Presto::
 
 WARNING: This drops/creates tables named ``one_row``, ``one_row_complex``, and ``many_rows``, plus a
 database called ``pyhive_test_database``.
+
+Updating TCLIService
+====================
+
+The TCLIService module is autogenerated using a ``TCLIService.thrift`` file. To update it, the
+``generate.py`` file can be used: ``python generate.py <TCLIServiceURL>``. When left blank, the
+version for Hive 2.3 will be downloaded.

dev_requirements.txt

@@ -11,7 +11,6 @@ pytest-timeout==1.2.0
 # actual dependencies: let things break if a package changes
 requests>=1.0.0
 sasl>=0.2.1
-sqlalchemy>=0.8.7
 thrift>=0.10.0
 #thrift_sasl>=0.1.0
 git+https://github.com/cloudera/thrift_sasl # Using master branch in order to get Python 3 SASL patches

generate.py

@@ -0,0 +1,52 @@
+"""
+This file can be used to generate a new version of the TCLIService package
+using a TCLIService.thrift URL.
+
+If no URL is specified, the file for Hive 2.3 will be downloaded.
+
+Usage:
+
+ python generate.py THRIFT_URL
+
+or
+
+ python generate.py
+"""
+import shutil
+import sys
+from os import path
+from urllib.request import urlopen
+import subprocess
+
+here = path.abspath(path.dirname(__file__))
+
+PACKAGE = 'TCLIService'
+GENERATED = 'gen-py'
+
+HIVE_SERVER2_URL = \
+ 'https://raw.githubusercontent.com/apache/hive/branch-2.3/service-rpc/if/TCLIService.thrift'
+
+
+def save_url(url):
+ data = urlopen(url).read()
+ file_path = path.join(here, url.rsplit('/', 1)[-1])
+ with open(file_path, 'wb') as f:
+ f.write(data)
+
+
+def main(hive_server2_url):
+ save_url(hive_server2_url)
+ hive_server2_path = path.join(here, hive_server2_url.rsplit('/', 1)[-1])
+
+ subprocess.call(['thrift', '-r', '--gen', 'py', hive_server2_path])
+ shutil.move(path.join(here, PACKAGE), path.join(here, PACKAGE + '.old'))
+ shutil.move(path.join(here, GENERATED, PACKAGE), path.join(here, PACKAGE))
+ shutil.rmtree(path.join(here, PACKAGE + '.old'))
+
+
+if __name__ == '__main__':
+ if len(sys.argv) > 1:
+ url = sys.argv[1]
+ else:
+ url = HIVE_SERVER2_URL
+ main(url)

pyhive/common.py

@@ -8,14 +8,14 @@
 from builtins import bytes
 from builtins import int
 from builtins import object
-from builtins import range
 from builtins import str
 from past.builtins import basestring
 from pyhive import exc
 import abc
 import collections
 import time
 from future.utils import with_metaclass
+from itertools import islice
 
 
 class DBAPICursor(with_metaclass(abc.ABCMeta, object)):
@@ -124,14 +124,7 @@ def fetchmany(self, size=None):
  """
  if size is None:
  size = self.arraysize
- result = []
- for _ in range(size):
- one = self.fetchone()
- if one is None:
- break
- else:
- result.append(one)
- return result
+ return list(islice(iter(self.fetchone, None), size))
 
  def fetchall(self):
  """Fetch all (remaining) rows of a query result, returning them as a sequence of sequences
@@ -140,14 +133,7 @@ def fetchall(self):
  An :py:class:`~pyhive.exc.Error` (or subclass) exception is raised if the previous call to
  :py:meth:`execute` did not produce any result set or no call was issued yet.
  """
- result = []
- while True:
- one = self.fetchone()
- if one is None:
- break
- else:
- result.append(one)
- return result
+ return list(iter(self.fetchone, None))
 
  @property
  def arraysize(self):

pyhive/hive.py

@@ -7,6 +7,11 @@
 
 from __future__ import absolute_import
 from __future__ import unicode_literals
+
+import datetime
+import re
+from decimal import Decimal
+
 from TCLIService import TCLIService
 from TCLIService import constants
 from TCLIService import ttypes
@@ -31,6 +36,31 @@
 
 _logger = logging.getLogger(__name__)
 
+_TIMESTAMP_PATTERN = re.compile(r'(\d+-\d+-\d+ \d+:\d+:\d+(\.\d{,6})?)')
+
+
+def _parse_timestamp(value):
+ if value:
+ match = _TIMESTAMP_PATTERN.match(value)
+ if match:
+ if match.group(2):
+ format = '%Y-%m-%d %H:%M:%S.%f'
+ # use the pattern to truncate the value
+ value = match.group()
+ else:
+ format = '%Y-%m-%d %H:%M:%S'
+ value = datetime.datetime.strptime(value, format)
+ else:
+ raise Exception(
+ 'Cannot convert "{}" into a datetime'.format(value))
+ else:
+ value = None
+ return value
+
+
+TYPES_CONVERTER = {"DECIMAL_TYPE": Decimal,
+ "TIMESTAMP_TYPE": _parse_timestamp}
+
 
 class HiveParamEscaper(common.ParamEscaper):
  def escape_string(self, item):
@@ -177,6 +207,14 @@ def sasl_factory():
  self._transport.close()
  raise
 
+ def __enter__(self):
+ """Transport should already be opened by __init__"""
+ return self
+
+ def __exit__(self, exc_type, exc_val, exc_tb):
+ """Call close"""
+ self.close()
+
  def close(self):
  """Close the underlying session and Thrift transport"""
  req = ttypes.TCloseSessionReq(sessionHandle=self._sessionHandle)
@@ -215,7 +253,7 @@ class Cursor(common.DBAPICursor):
  def __init__(self, connection, arraysize=1000):
  self._operationHandle = None
  super(Cursor, self).__init__()
- self.arraysize = arraysize
+ self._arraysize = arraysize
  self._connection = connection
 
  def _reset_state(self):
@@ -230,6 +268,19 @@ def _reset_state(self):
  finally:
  self._operationHandle = None
 
+ @property
+ def arraysize(self):
+ return self._arraysize
+
+ @arraysize.setter
+ def arraysize(self, value):
+ """Array size cannot be None, and should be an integer"""
+ default_arraysize = 1000
+ try:
+ self._arraysize = int(value) or default_arraysize
+ except TypeError:
+ self._arraysize = default_arraysize
+
  @property
  def description(self):
  """This read-only attribute is a sequence of 7-item sequences.
@@ -273,6 +324,12 @@ def description(self):
  ))
  return self._description
 
+ def __enter__(self):
+ return self
+
+ def __exit__(self, exc_type, exc_val, exc_tb):
+ self.close()
+
  def close(self):
  """Close the operation handle"""
  self._reset_state()
@@ -320,8 +377,10 @@ def _fetch_more(self):
  )
  response = self._connection.client.FetchResults(req)
  _check_status(response)
+ schema = self.description
  assert not response.results.rows, 'expected data in columnar format'
- columns = map(_unwrap_column, response.results.columns)
+ columns = [_unwrap_column(col, col_schema[1]) for col, col_schema in
+ zip(response.results.columns, schema)]
  new_data = list(zip(*columns))
  self._data += new_data
  # response.hasMoreRows seems to always be False, so we instead check the number of rows
@@ -402,7 +461,7 @@ def fetch_logs(self):
 #
 
 
-def _unwrap_column(col):
+def _unwrap_column(col, type_=None):
  """Return a list of raw values from a TColumn instance."""
  for attr, wrapper in iteritems(col.__dict__):
  if wrapper is not None:
@@ -414,6 +473,9 @@ def _unwrap_column(col):
  for b in range(8):
  if byte & (1 << b):
  result[i * 8 + b] = None
+ converter = TYPES_CONVERTER.get(type_, None)
+ if converter and type_:
+ result = [converter(row) if row else row for row in result]
  return result
  raise DataError("Got empty column value {}".format(col)) # pragma: no cover
 

.travis.yml

@@ -7,17 +7,14 @@ matrix:
  # One build pulls latest versions dynamically
  - python: 3.6
  env: CDH=cdh5 CDH_VERSION=5 PRESTO=RELEASE SQLALCHEMY=sqlalchemy
- # Others use pinned versions.
+ - python: 3.6
+ env: CDH=cdh5 CDH_VERSION=5.10.1 PRESTO=0.147 SQLALCHEMY=sqlalchemy==1.2.8
  - python: 3.5
- env: CDH=cdh5 CDH_VERSION=5.10.1 PRESTO=0.147 SQLALCHEMY=sqlalchemy==1.0.12
+ env: CDH=cdh5 CDH_VERSION=5.10.1 PRESTO=0.147 SQLALCHEMY=sqlalchemy==1.2.8
  - python: 3.4
- env: CDH=cdh5 CDH_VERSION=5.10.1 PRESTO=0.147 SQLALCHEMY=sqlalchemy==1.0.12
- - python: 2.7
- env: CDH=cdh5 CDH_VERSION=5.10.1 PRESTO=0.147 SQLALCHEMY=sqlalchemy==1.0.12
- # stale stuff we're still using / supporting
+ env: CDH=cdh5 CDH_VERSION=5.10.1 PRESTO=0.147 SQLALCHEMY=sqlalchemy==1.2.8
  - python: 2.7
- env: CDH=cdh5 CDH_VERSION=5.10.1 PRESTO=0.147 SQLALCHEMY=sqlalchemy==0.8.7
- # exclude: python 3 against old libries
+ env: CDH=cdh5 CDH_VERSION=5.10.1 PRESTO=0.147 SQLALCHEMY=sqlalchemy==1.2.8
 install:
  - ./scripts/travis-install.sh
  - pip install codecov

README.rst

@@ -1,5 +1,6 @@
 .. image:: https://travis-ci.org/dropbox/PyHive.svg?branch=master
  :target: https://travis-ci.org/dropbox/PyHive
+.. image:: https://img.shields.io/codecov/c/github/dropbox/PyHive.svg
 
 ======
 PyHive
@@ -100,9 +101,6 @@ PyHive works with
 - Python 2.7 / Python 3
 - For Presto: Presto install
 - For Hive: `HiveServer2 <https://cwiki.apache.org/confluence/display/Hive/Setting+up+HiveServer2>`_ daemon
-- For Python 3 + Hive + SASL, you currently need to install an unreleased version of ``thrift_sasl``
- (``pip install git+https://github.com/cloudera/thrift_sasl``).
- At the time of writing, the latest version of ``thrift_sasl`` was 0.2.1.
 
 Changelog
 =========
@@ -137,3 +135,10 @@ Run the following in an environment with Hive/Presto::
 
 WARNING: This drops/creates tables named ``one_row``, ``one_row_complex``, and ``many_rows``, plus a
 database called ``pyhive_test_database``.
+
+Updating TCLIService
+====================
+
+The TCLIService module is autogenerated using a ``TCLIService.thrift`` file. To update it, the
+``generate.py`` file can be used: ``python generate.py <TCLIServiceURL>``. When left blank, the
+version for Hive 2.3 will be downloaded.

dev_requirements.txt

@@ -11,7 +11,6 @@ pytest-timeout==1.2.0
 # actual dependencies: let things break if a package changes
 requests>=1.0.0
 sasl>=0.2.1
-sqlalchemy>=0.8.7
 thrift>=0.10.0
 #thrift_sasl>=0.1.0
 git+https://github.com/cloudera/thrift_sasl # Using master branch in order to get Python 3 SASL patches

generate.py

@@ -0,0 +1,52 @@
+"""
+This file can be used to generate a new version of the TCLIService package
+using a TCLIService.thrift URL.
+
+If no URL is specified, the file for Hive 2.3 will be downloaded.
+
+Usage:
+
+ python generate.py THRIFT_URL
+
+or
+
+ python generate.py
+"""
+import shutil
+import sys
+from os import path
+from urllib.request import urlopen
+import subprocess
+
+here = path.abspath(path.dirname(__file__))
+
+PACKAGE = 'TCLIService'
+GENERATED = 'gen-py'
+
+HIVE_SERVER2_URL = \
+ 'https://raw.githubusercontent.com/apache/hive/branch-2.3/service-rpc/if/TCLIService.thrift'
+
+
+def save_url(url):
+ data = urlopen(url).read()
+ file_path = path.join(here, url.rsplit('/', 1)[-1])
+ with open(file_path, 'wb') as f:
+ f.write(data)
+
+
+def main(hive_server2_url):
+ save_url(hive_server2_url)
+ hive_server2_path = path.join(here, hive_server2_url.rsplit('/', 1)[-1])
+
+ subprocess.call(['thrift', '-r', '--gen', 'py', hive_server2_path])
+ shutil.move(path.join(here, PACKAGE), path.join(here, PACKAGE + '.old'))
+ shutil.move(path.join(here, GENERATED, PACKAGE), path.join(here, PACKAGE))
+ shutil.rmtree(path.join(here, PACKAGE + '.old'))
+
+
+if __name__ == '__main__':
+ if len(sys.argv) > 1:
+ url = sys.argv[1]
+ else:
+ url = HIVE_SERVER2_URL
+ main(url)

pyhive/common.py

@@ -8,14 +8,14 @@
 from builtins import bytes
 from builtins import int
 from builtins import object
-from builtins import range
 from builtins import str
 from past.builtins import basestring
 from pyhive import exc
 import abc
 import collections
 import time
 from future.utils import with_metaclass
+from itertools import islice
 
 
 class DBAPICursor(with_metaclass(abc.ABCMeta, object)):
@@ -124,14 +124,7 @@ def fetchmany(self, size=None):
  """
  if size is None:
  size = self.arraysize
- result = []
- for _ in range(size):
- one = self.fetchone()
- if one is None:
- break
- else:
- result.append(one)
- return result
+ return list(islice(iter(self.fetchone, None), size))
 
  def fetchall(self):
  """Fetch all (remaining) rows of a query result, returning them as a sequence of sequences
@@ -140,14 +133,7 @@ def fetchall(self):
  An :py:class:`~pyhive.exc.Error` (or subclass) exception is raised if the previous call to
  :py:meth:`execute` did not produce any result set or no call was issued yet.
  """
- result = []
- while True:
- one = self.fetchone()
- if one is None:
- break
- else:
- result.append(one)
- return result
+ return list(iter(self.fetchone, None))
 
  @property
  def arraysize(self):

pyhive/hive.py

@@ -7,6 +7,11 @@
 
 from __future__ import absolute_import
 from __future__ import unicode_literals
+
+import datetime
+import re
+from decimal import Decimal
+
 from TCLIService import TCLIService
 from TCLIService import constants
 from TCLIService import ttypes
@@ -31,6 +36,31 @@
 
 _logger = logging.getLogger(__name__)
 
+_TIMESTAMP_PATTERN = re.compile(r'(\d+-\d+-\d+ \d+:\d+:\d+(\.\d{,6})?)')
+
+
+def _parse_timestamp(value):
+ if value:
+ match = _TIMESTAMP_PATTERN.match(value)
+ if match:
+ if match.group(2):
+ format = '%Y-%m-%d %H:%M:%S.%f'
+ # use the pattern to truncate the value
+ value = match.group()
+ else:
+ format = '%Y-%m-%d %H:%M:%S'
+ value = datetime.datetime.strptime(value, format)
+ else:
+ raise Exception(
+ 'Cannot convert "{}" into a datetime'.format(value))
+ else:
+ value = None
+ return value
+
+
+TYPES_CONVERTER = {"DECIMAL_TYPE": Decimal,
+ "TIMESTAMP_TYPE": _parse_timestamp}
+
 
 class HiveParamEscaper(common.ParamEscaper):
  def escape_string(self, item):
@@ -177,6 +207,14 @@ def sasl_factory():
  self._transport.close()
  raise
 
+ def __enter__(self):
+ """Transport should already be opened by __init__"""
+ return self
+
+ def __exit__(self, exc_type, exc_val, exc_tb):
+ """Call close"""
+ self.close()
+
  def close(self):
  """Close the underlying session and Thrift transport"""
  req = ttypes.TCloseSessionReq(sessionHandle=self._sessionHandle)
@@ -215,7 +253,7 @@ class Cursor(common.DBAPICursor):
  def __init__(self, connection, arraysize=1000):
  self._operationHandle = None
  super(Cursor, self).__init__()
- self.arraysize = arraysize
+ self._arraysize = arraysize
  self._connection = connection
 
  def _reset_state(self):
@@ -230,6 +268,19 @@ def _reset_state(self):
  finally:
  self._operationHandle = None
 
+ @property
+ def arraysize(self):
+ return self._arraysize
+
+ @arraysize.setter
+ def arraysize(self, value):
+ """Array size cannot be None, and should be an integer"""
+ default_arraysize = 1000
+ try:
+ self._arraysize = int(value) or default_arraysize
+ except TypeError:
+ self._arraysize = default_arraysize
+
  @property
  def description(self):
  """This read-only attribute is a sequence of 7-item sequences.
@@ -273,6 +324,12 @@ def description(self):
  ))
  return self._description
 
+ def __enter__(self):
+ return self
+
+ def __exit__(self, exc_type, exc_val, exc_tb):
+ self.close()
+
  def close(self):
  """Close the operation handle"""
  self._reset_state()
@@ -320,8 +377,10 @@ def _fetch_more(self):
  )
  response = self._connection.client.FetchResults(req)
  _check_status(response)
+ schema = self.description
  assert not response.results.rows, 'expected data in columnar format'
- columns = map(_unwrap_column, response.results.columns)
+ columns = [_unwrap_column(col, col_schema[1]) for col, col_schema in
+ zip(response.results.columns, schema)]
  new_data = list(zip(*columns))
  self._data += new_data
  # response.hasMoreRows seems to always be False, so we instead check the number of rows
@@ -402,7 +461,7 @@ def fetch_logs(self):
 #
 
 
-def _unwrap_column(col):
+def _unwrap_column(col, type_=None):
  """Return a list of raw values from a TColumn instance."""
  for attr, wrapper in iteritems(col.__dict__):
  if wrapper is not None:
@@ -414,6 +473,9 @@ def _unwrap_column(col):
  for b in range(8):
  if byte & (1 << b):
  result[i * 8 + b] = None
+ converter = TYPES_CONVERTER.get(type_, None)
+ if converter and type_:
+ result = [converter(row) if row else row for row in result]
  return result
  raise DataError("Got empty column value {}".format(col)) # pragma: no cover
 

.travis.yml

@@ -7,17 +7,14 @@ matrix:
  # One build pulls latest versions dynamically
  - python: 3.6
  env: CDH=cdh5 CDH_VERSION=5 PRESTO=RELEASE SQLALCHEMY=sqlalchemy
- # Others use pinned versions.
+ - python: 3.6
+ env: CDH=cdh5 CDH_VERSION=5.10.1 PRESTO=0.147 SQLALCHEMY=sqlalchemy==1.2.8
  - python: 3.5
- env: CDH=cdh5 CDH_VERSION=5.10.1 PRESTO=0.147 SQLALCHEMY=sqlalchemy==1.0.12
+ env: CDH=cdh5 CDH_VERSION=5.10.1 PRESTO=0.147 SQLALCHEMY=sqlalchemy==1.2.8
  - python: 3.4
- env: CDH=cdh5 CDH_VERSION=5.10.1 PRESTO=0.147 SQLALCHEMY=sqlalchemy==1.0.12
- - python: 2.7
- env: CDH=cdh5 CDH_VERSION=5.10.1 PRESTO=0.147 SQLALCHEMY=sqlalchemy==1.0.12
- # stale stuff we're still using / supporting
+ env: CDH=cdh5 CDH_VERSION=5.10.1 PRESTO=0.147 SQLALCHEMY=sqlalchemy==1.2.8
  - python: 2.7
- env: CDH=cdh5 CDH_VERSION=5.10.1 PRESTO=0.147 SQLALCHEMY=sqlalchemy==0.8.7
- # exclude: python 3 against old libries
+ env: CDH=cdh5 CDH_VERSION=5.10.1 PRESTO=0.147 SQLALCHEMY=sqlalchemy==1.2.8
 install:
  - ./scripts/travis-install.sh
  - pip install codecov

README.rst

@@ -1,5 +1,6 @@
 .. image:: https://travis-ci.org/dropbox/PyHive.svg?branch=master
  :target: https://travis-ci.org/dropbox/PyHive
+.. image:: https://img.shields.io/codecov/c/github/dropbox/PyHive.svg
 
 ======
 PyHive
@@ -100,9 +101,6 @@ PyHive works with
 - Python 2.7 / Python 3
 - For Presto: Presto install
 - For Hive: `HiveServer2 <https://cwiki.apache.org/confluence/display/Hive/Setting+up+HiveServer2>`_ daemon
-- For Python 3 + Hive + SASL, you currently need to install an unreleased version of ``thrift_sasl``
- (``pip install git+https://github.com/cloudera/thrift_sasl``).
- At the time of writing, the latest version of ``thrift_sasl`` was 0.2.1.
 
 Changelog
 =========
@@ -137,3 +135,10 @@ Run the following in an environment with Hive/Presto::
 
 WARNING: This drops/creates tables named ``one_row``, ``one_row_complex``, and ``many_rows``, plus a
 database called ``pyhive_test_database``.
+
+Updating TCLIService
+====================
+
+The TCLIService module is autogenerated using a ``TCLIService.thrift`` file. To update it, the
+``generate.py`` file can be used: ``python generate.py <TCLIServiceURL>``. When left blank, the
+version for Hive 2.3 will be downloaded.

dev_requirements.txt

@@ -11,7 +11,6 @@ pytest-timeout==1.2.0
 # actual dependencies: let things break if a package changes
 requests>=1.0.0
 sasl>=0.2.1
-sqlalchemy>=0.8.7
 thrift>=0.10.0
 #thrift_sasl>=0.1.0
 git+https://github.com/cloudera/thrift_sasl # Using master branch in order to get Python 3 SASL patches

generate.py

@@ -0,0 +1,52 @@
+"""
+This file can be used to generate a new version of the TCLIService package
+using a TCLIService.thrift URL.
+
+If no URL is specified, the file for Hive 2.3 will be downloaded.
+
+Usage:
+
+ python generate.py THRIFT_URL
+
+or
+
+ python generate.py
+"""
+import shutil
+import sys
+from os import path
+from urllib.request import urlopen
+import subprocess
+
+here = path.abspath(path.dirname(__file__))
+
+PACKAGE = 'TCLIService'
+GENERATED = 'gen-py'
+
+HIVE_SERVER2_URL = \
+ 'https://raw.githubusercontent.com/apache/hive/branch-2.3/service-rpc/if/TCLIService.thrift'
+
+
+def save_url(url):
+ data = urlopen(url).read()
+ file_path = path.join(here, url.rsplit('/', 1)[-1])
+ with open(file_path, 'wb') as f:
+ f.write(data)
+
+
+def main(hive_server2_url):
+ save_url(hive_server2_url)
+ hive_server2_path = path.join(here, hive_server2_url.rsplit('/', 1)[-1])
+
+ subprocess.call(['thrift', '-r', '--gen', 'py', hive_server2_path])
+ shutil.move(path.join(here, PACKAGE), path.join(here, PACKAGE + '.old'))
+ shutil.move(path.join(here, GENERATED, PACKAGE), path.join(here, PACKAGE))
+ shutil.rmtree(path.join(here, PACKAGE + '.old'))
+
+
+if __name__ == '__main__':
+ if len(sys.argv) > 1:
+ url = sys.argv[1]
+ else:
+ url = HIVE_SERVER2_URL
+ main(url)

pyhive/common.py

@@ -8,14 +8,14 @@
 from builtins import bytes
 from builtins import int
 from builtins import object
-from builtins import range
 from builtins import str
 from past.builtins import basestring
 from pyhive import exc
 import abc
 import collections
 import time
 from future.utils import with_metaclass
+from itertools import islice
 
 
 class DBAPICursor(with_metaclass(abc.ABCMeta, object)):
@@ -124,14 +124,7 @@ def fetchmany(self, size=None):
  """
  if size is None:
  size = self.arraysize
- result = []
- for _ in range(size):
- one = self.fetchone()
- if one is None:
- break
- else:
- result.append(one)
- return result
+ return list(islice(iter(self.fetchone, None), size))
 
  def fetchall(self):
  """Fetch all (remaining) rows of a query result, returning them as a sequence of sequences
@@ -140,14 +133,7 @@ def fetchall(self):
  An :py:class:`~pyhive.exc.Error` (or subclass) exception is raised if the previous call to
  :py:meth:`execute` did not produce any result set or no call was issued yet.
  """
- result = []
- while True:
- one = self.fetchone()
- if one is None:
- break
- else:
- result.append(one)
- return result
+ return list(iter(self.fetchone, None))
 
  @property
  def arraysize(self):

pyhive/hive.py

@@ -7,6 +7,11 @@
 
 from __future__ import absolute_import
 from __future__ import unicode_literals
+
+import datetime
+import re
+from decimal import Decimal
+
 from TCLIService import TCLIService
 from TCLIService import constants
 from TCLIService import ttypes
@@ -31,6 +36,31 @@
 
 _logger = logging.getLogger(__name__)
 
+_TIMESTAMP_PATTERN = re.compile(r'(\d+-\d+-\d+ \d+:\d+:\d+(\.\d{,6})?)')
+
+
+def _parse_timestamp(value):
+ if value:
+ match = _TIMESTAMP_PATTERN.match(value)
+ if match:
+ if match.group(2):
+ format = '%Y-%m-%d %H:%M:%S.%f'
+ # use the pattern to truncate the value
+ value = match.group()
+ else:
+ format = '%Y-%m-%d %H:%M:%S'
+ value = datetime.datetime.strptime(value, format)
+ else:
+ raise Exception(
+ 'Cannot convert "{}" into a datetime'.format(value))
+ else:
+ value = None
+ return value
+
+
+TYPES_CONVERTER = {"DECIMAL_TYPE": Decimal,
+ "TIMESTAMP_TYPE": _parse_timestamp}
+
 
 class HiveParamEscaper(common.ParamEscaper):
  def escape_string(self, item):
@@ -177,6 +207,14 @@ def sasl_factory():
  self._transport.close()
  raise
 
+ def __enter__(self):
+ """Transport should already be opened by __init__"""
+ return self
+
+ def __exit__(self, exc_type, exc_val, exc_tb):
+ """Call close"""
+ self.close()
+
  def close(self):
  """Close the underlying session and Thrift transport"""
  req = ttypes.TCloseSessionReq(sessionHandle=self._sessionHandle)
@@ -215,7 +253,7 @@ class Cursor(common.DBAPICursor):
  def __init__(self, connection, arraysize=1000):
  self._operationHandle = None
  super(Cursor, self).__init__()
- self.arraysize = arraysize
+ self._arraysize = arraysize
  self._connection = connection
 
  def _reset_state(self):
@@ -230,6 +268,19 @@ def _reset_state(self):
  finally:
  self._operationHandle = None
 
+ @property
+ def arraysize(self):
+ return self._arraysize
+
+ @arraysize.setter
+ def arraysize(self, value):
+ """Array size cannot be None, and should be an integer"""
+ default_arraysize = 1000
+ try:
+ self._arraysize = int(value) or default_arraysize
+ except TypeError:
+ self._arraysize = default_arraysize
+
  @property
  def description(self):
  """This read-only attribute is a sequence of 7-item sequences.
@@ -273,6 +324,12 @@ def description(self):
  ))
  return self._description
 
+ def __enter__(self):
+ return self
+
+ def __exit__(self, exc_type, exc_val, exc_tb):
+ self.close()
+
  def close(self):
  """Close the operation handle"""
  self._reset_state()
@@ -320,8 +377,10 @@ def _fetch_more(self):
  )
  response = self._connection.client.FetchResults(req)
  _check_status(response)
+ schema = self.description
  assert not response.results.rows, 'expected data in columnar format'
- columns = map(_unwrap_column, response.results.columns)
+ columns = [_unwrap_column(col, col_schema[1]) for col, col_schema in
+ zip(response.results.columns, schema)]
  new_data = list(zip(*columns))
  self._data += new_data
  # response.hasMoreRows seems to always be False, so we instead check the number of rows
@@ -402,7 +461,7 @@ def fetch_logs(self):
 #
 
 
-def _unwrap_column(col):
+def _unwrap_column(col, type_=None):
  """Return a list of raw values from a TColumn instance."""
  for attr, wrapper in iteritems(col.__dict__):
  if wrapper is not None:
@@ -414,6 +473,9 @@ def _unwrap_column(col):
  for b in range(8):
  if byte & (1 << b):
  result[i * 8 + b] = None
+ converter = TYPES_CONVERTER.get(type_, None)
+ if converter and type_:
+ result = [converter(row) if row else row for row in result]
  return result
  raise DataError("Got empty column value {}".format(col)) # pragma: no cover
 

.travis.yml

@@ -7,17 +7,14 @@ matrix:
  # One build pulls latest versions dynamically
  - python: 3.6
  env: CDH=cdh5 CDH_VERSION=5 PRESTO=RELEASE SQLALCHEMY=sqlalchemy
- # Others use pinned versions.
+ - python: 3.6
+ env: CDH=cdh5 CDH_VERSION=5.10.1 PRESTO=0.147 SQLALCHEMY=sqlalchemy==1.2.8
  - python: 3.5
- env: CDH=cdh5 CDH_VERSION=5.10.1 PRESTO=0.147 SQLALCHEMY=sqlalchemy==1.0.12
+ env: CDH=cdh5 CDH_VERSION=5.10.1 PRESTO=0.147 SQLALCHEMY=sqlalchemy==1.2.8
  - python: 3.4
- env: CDH=cdh5 CDH_VERSION=5.10.1 PRESTO=0.147 SQLALCHEMY=sqlalchemy==1.0.12
- - python: 2.7
- env: CDH=cdh5 CDH_VERSION=5.10.1 PRESTO=0.147 SQLALCHEMY=sqlalchemy==1.0.12
- # stale stuff we're still using / supporting
+ env: CDH=cdh5 CDH_VERSION=5.10.1 PRESTO=0.147 SQLALCHEMY=sqlalchemy==1.2.8
  - python: 2.7
- env: CDH=cdh5 CDH_VERSION=5.10.1 PRESTO=0.147 SQLALCHEMY=sqlalchemy==0.8.7
- # exclude: python 3 against old libries
+ env: CDH=cdh5 CDH_VERSION=5.10.1 PRESTO=0.147 SQLALCHEMY=sqlalchemy==1.2.8
 install:
  - ./scripts/travis-install.sh
  - pip install codecov

README.rst

@@ -1,5 +1,6 @@
 .. image:: https://travis-ci.org/dropbox/PyHive.svg?branch=master
  :target: https://travis-ci.org/dropbox/PyHive
+.. image:: https://img.shields.io/codecov/c/github/dropbox/PyHive.svg
 
 ======
 PyHive
@@ -100,9 +101,6 @@ PyHive works with
 - Python 2.7 / Python 3
 - For Presto: Presto install
 - For Hive: `HiveServer2 <https://cwiki.apache.org/confluence/display/Hive/Setting+up+HiveServer2>`_ daemon
-- For Python 3 + Hive + SASL, you currently need to install an unreleased version of ``thrift_sasl``
- (``pip install git+https://github.com/cloudera/thrift_sasl``).
- At the time of writing, the latest version of ``thrift_sasl`` was 0.2.1.
 
 Changelog
 =========
@@ -137,3 +135,10 @@ Run the following in an environment with Hive/Presto::
 
 WARNING: This drops/creates tables named ``one_row``, ``one_row_complex``, and ``many_rows``, plus a
 database called ``pyhive_test_database``.
+
+Updating TCLIService
+====================
+
+The TCLIService module is autogenerated using a ``TCLIService.thrift`` file. To update it, the
+``generate.py`` file can be used: ``python generate.py <TCLIServiceURL>``. When left blank, the
+version for Hive 2.3 will be downloaded.

dev_requirements.txt

@@ -11,7 +11,6 @@ pytest-timeout==1.2.0
 # actual dependencies: let things break if a package changes
 requests>=1.0.0
 sasl>=0.2.1
-sqlalchemy>=0.8.7
 thrift>=0.10.0
 #thrift_sasl>=0.1.0
 git+https://github.com/cloudera/thrift_sasl # Using master branch in order to get Python 3 SASL patches

generate.py

@@ -0,0 +1,52 @@
+"""
+This file can be used to generate a new version of the TCLIService package
+using a TCLIService.thrift URL.
+
+If no URL is specified, the file for Hive 2.3 will be downloaded.
+
+Usage:
+
+ python generate.py THRIFT_URL
+
+or
+
+ python generate.py
+"""
+import shutil
+import sys
+from os import path
+from urllib.request import urlopen
+import subprocess
+
+here = path.abspath(path.dirname(__file__))
+
+PACKAGE = 'TCLIService'
+GENERATED = 'gen-py'
+
+HIVE_SERVER2_URL = \
+ 'https://raw.githubusercontent.com/apache/hive/branch-2.3/service-rpc/if/TCLIService.thrift'
+
+
+def save_url(url):
+ data = urlopen(url).read()
+ file_path = path.join(here, url.rsplit('/', 1)[-1])
+ with open(file_path, 'wb') as f:
+ f.write(data)
+
+
+def main(hive_server2_url):
+ save_url(hive_server2_url)
+ hive_server2_path = path.join(here, hive_server2_url.rsplit('/', 1)[-1])
+
+ subprocess.call(['thrift', '-r', '--gen', 'py', hive_server2_path])
+ shutil.move(path.join(here, PACKAGE), path.join(here, PACKAGE + '.old'))
+ shutil.move(path.join(here, GENERATED, PACKAGE), path.join(here, PACKAGE))
+ shutil.rmtree(path.join(here, PACKAGE + '.old'))
+
+
+if __name__ == '__main__':
+ if len(sys.argv) > 1:
+ url = sys.argv[1]
+ else:
+ url = HIVE_SERVER2_URL
+ main(url)

pyhive/common.py

@@ -8,14 +8,14 @@
 from builtins import bytes
 from builtins import int
 from builtins import object
-from builtins import range
 from builtins import str
 from past.builtins import basestring
 from pyhive import exc
 import abc
 import collections
 import time
 from future.utils import with_metaclass
+from itertools import islice
 
 
 class DBAPICursor(with_metaclass(abc.ABCMeta, object)):
@@ -124,14 +124,7 @@ def fetchmany(self, size=None):
  """
  if size is None:
  size = self.arraysize
- result = []
- for _ in range(size):
- one = self.fetchone()
- if one is None:
- break
- else:
- result.append(one)
- return result
+ return list(islice(iter(self.fetchone, None), size))
 
  def fetchall(self):
  """Fetch all (remaining) rows of a query result, returning them as a sequence of sequences
@@ -140,14 +133,7 @@ def fetchall(self):
  An :py:class:`~pyhive.exc.Error` (or subclass) exception is raised if the previous call to
  :py:meth:`execute` did not produce any result set or no call was issued yet.
  """
- result = []
- while True:
- one = self.fetchone()
- if one is None:
- break
- else:
- result.append(one)
- return result
+ return list(iter(self.fetchone, None))
 
  @property
  def arraysize(self):

pyhive/hive.py

@@ -7,6 +7,11 @@
 
 from __future__ import absolute_import
 from __future__ import unicode_literals
+
+import datetime
+import re
+from decimal import Decimal
+
 from TCLIService import TCLIService
 from TCLIService import constants
 from TCLIService import ttypes
@@ -31,6 +36,31 @@
 
 _logger = logging.getLogger(__name__)
 
+_TIMESTAMP_PATTERN = re.compile(r'(\d+-\d+-\d+ \d+:\d+:\d+(\.\d{,6})?)')
+
+
+def _parse_timestamp(value):
+ if value:
+ match = _TIMESTAMP_PATTERN.match(value)
+ if match:
+ if match.group(2):
+ format = '%Y-%m-%d %H:%M:%S.%f'
+ # use the pattern to truncate the value
+ value = match.group()
+ else:
+ format = '%Y-%m-%d %H:%M:%S'
+ value = datetime.datetime.strptime(value, format)
+ else:
+ raise Exception(
+ 'Cannot convert "{}" into a datetime'.format(value))
+ else:
+ value = None
+ return value
+
+
+TYPES_CONVERTER = {"DECIMAL_TYPE": Decimal,
+ "TIMESTAMP_TYPE": _parse_timestamp}
+
 
 class HiveParamEscaper(common.ParamEscaper):
  def escape_string(self, item):
@@ -177,6 +207,14 @@ def sasl_factory():
  self._transport.close()
  raise
 
+ def __enter__(self):
+ """Transport should already be opened by __init__"""
+ return self
+
+ def __exit__(self, exc_type, exc_val, exc_tb):
+ """Call close"""
+ self.close()
+
  def close(self):
  """Close the underlying session and Thrift transport"""
  req = ttypes.TCloseSessionReq(sessionHandle=self._sessionHandle)
@@ -215,7 +253,7 @@ class Cursor(common.DBAPICursor):
  def __init__(self, connection, arraysize=1000):
  self._operationHandle = None
  super(Cursor, self).__init__()
- self.arraysize = arraysize
+ self._arraysize = arraysize
  self._connection = connection
 
  def _reset_state(self):
@@ -230,6 +268,19 @@ def _reset_state(self):
  finally:
  self._operationHandle = None
 
+ @property
+ def arraysize(self):
+ return self._arraysize
+
+ @arraysize.setter
+ def arraysize(self, value):
+ """Array size cannot be None, and should be an integer"""
+ default_arraysize = 1000
+ try:
+ self._arraysize = int(value) or default_arraysize
+ except TypeError:
+ self._arraysize = default_arraysize
+
  @property
  def description(self):
  """This read-only attribute is a sequence of 7-item sequences.
@@ -273,6 +324,12 @@ def description(self):
  ))
  return self._description
 
+ def __enter__(self):
+ return self
+
+ def __exit__(self, exc_type, exc_val, exc_tb):
+ self.close()
+
  def close(self):
  """Close the operation handle"""
  self._reset_state()
@@ -320,8 +377,10 @@ def _fetch_more(self):
  )
  response = self._connection.client.FetchResults(req)
  _check_status(response)
+ schema = self.description
  assert not response.results.rows, 'expected data in columnar format'
- columns = map(_unwrap_column, response.results.columns)
+ columns = [_unwrap_column(col, col_schema[1]) for col, col_schema in
+ zip(response.results.columns, schema)]
  new_data = list(zip(*columns))
  self._data += new_data
  # response.hasMoreRows seems to always be False, so we instead check the number of rows
@@ -402,7 +461,7 @@ def fetch_logs(self):
 #
 
 
-def _unwrap_column(col):
+def _unwrap_column(col, type_=None):
  """Return a list of raw values from a TColumn instance."""
  for attr, wrapper in iteritems(col.__dict__):
  if wrapper is not None:
@@ -414,6 +473,9 @@ def _unwrap_column(col):
  for b in range(8):
  if byte & (1 << b):
  result[i * 8 + b] = None
+ converter = TYPES_CONVERTER.get(type_, None)
+ if converter and type_:
+ result = [converter(row) if row else row for row in result]
  return result
  raise DataError("Got empty column value {}".format(col)) # pragma: no cover
 

.travis.yml

@@ -7,17 +7,14 @@ matrix:
  # One build pulls latest versions dynamically
  - python: 3.6
  env: CDH=cdh5 CDH_VERSION=5 PRESTO=RELEASE SQLALCHEMY=sqlalchemy
- # Others use pinned versions.
+ - python: 3.6
+ env: CDH=cdh5 CDH_VERSION=5.10.1 PRESTO=0.147 SQLALCHEMY=sqlalchemy==1.2.8
  - python: 3.5
- env: CDH=cdh5 CDH_VERSION=5.10.1 PRESTO=0.147 SQLALCHEMY=sqlalchemy==1.0.12
+ env: CDH=cdh5 CDH_VERSION=5.10.1 PRESTO=0.147 SQLALCHEMY=sqlalchemy==1.2.8
  - python: 3.4
- env: CDH=cdh5 CDH_VERSION=5.10.1 PRESTO=0.147 SQLALCHEMY=sqlalchemy==1.0.12
- - python: 2.7
- env: CDH=cdh5 CDH_VERSION=5.10.1 PRESTO=0.147 SQLALCHEMY=sqlalchemy==1.0.12
- # stale stuff we're still using / supporting
+ env: CDH=cdh5 CDH_VERSION=5.10.1 PRESTO=0.147 SQLALCHEMY=sqlalchemy==1.2.8
  - python: 2.7
- env: CDH=cdh5 CDH_VERSION=5.10.1 PRESTO=0.147 SQLALCHEMY=sqlalchemy==0.8.7
- # exclude: python 3 against old libries
+ env: CDH=cdh5 CDH_VERSION=5.10.1 PRESTO=0.147 SQLALCHEMY=sqlalchemy==1.2.8
 install:
  - ./scripts/travis-install.sh
  - pip install codecov

README.rst

@@ -1,5 +1,6 @@
 .. image:: https://travis-ci.org/dropbox/PyHive.svg?branch=master
  :target: https://travis-ci.org/dropbox/PyHive
+.. image:: https://img.shields.io/codecov/c/github/dropbox/PyHive.svg
 
 ======
 PyHive
@@ -100,9 +101,6 @@ PyHive works with
 - Python 2.7 / Python 3
 - For Presto: Presto install
 - For Hive: `HiveServer2 <https://cwiki.apache.org/confluence/display/Hive/Setting+up+HiveServer2>`_ daemon
-- For Python 3 + Hive + SASL, you currently need to install an unreleased version of ``thrift_sasl``
- (``pip install git+https://github.com/cloudera/thrift_sasl``).
- At the time of writing, the latest version of ``thrift_sasl`` was 0.2.1.
 
 Changelog
 =========
@@ -137,3 +135,10 @@ Run the following in an environment with Hive/Presto::
 
 WARNING: This drops/creates tables named ``one_row``, ``one_row_complex``, and ``many_rows``, plus a
 database called ``pyhive_test_database``.
+
+Updating TCLIService
+====================
+
+The TCLIService module is autogenerated using a ``TCLIService.thrift`` file. To update it, the
+``generate.py`` file can be used: ``python generate.py <TCLIServiceURL>``. When left blank, the
+version for Hive 2.3 will be downloaded.

dev_requirements.txt

@@ -11,7 +11,6 @@ pytest-timeout==1.2.0
 # actual dependencies: let things break if a package changes
 requests>=1.0.0
 sasl>=0.2.1
-sqlalchemy>=0.8.7
 thrift>=0.10.0
 #thrift_sasl>=0.1.0
 git+https://github.com/cloudera/thrift_sasl # Using master branch in order to get Python 3 SASL patches

generate.py

@@ -0,0 +1,52 @@
+"""
+This file can be used to generate a new version of the TCLIService package
+using a TCLIService.thrift URL.
+
+If no URL is specified, the file for Hive 2.3 will be downloaded.
+
+Usage:
+
+ python generate.py THRIFT_URL
+
+or
+
+ python generate.py
+"""
+import shutil
+import sys
+from os import path
+from urllib.request import urlopen
+import subprocess
+
+here = path.abspath(path.dirname(__file__))
+
+PACKAGE = 'TCLIService'
+GENERATED = 'gen-py'
+
+HIVE_SERVER2_URL = \
+ 'https://raw.githubusercontent.com/apache/hive/branch-2.3/service-rpc/if/TCLIService.thrift'
+
+
+def save_url(url):
+ data = urlopen(url).read()
+ file_path = path.join(here, url.rsplit('/', 1)[-1])
+ with open(file_path, 'wb') as f:
+ f.write(data)
+
+
+def main(hive_server2_url):
+ save_url(hive_server2_url)
+ hive_server2_path = path.join(here, hive_server2_url.rsplit('/', 1)[-1])
+
+ subprocess.call(['thrift', '-r', '--gen', 'py', hive_server2_path])
+ shutil.move(path.join(here, PACKAGE), path.join(here, PACKAGE + '.old'))
+ shutil.move(path.join(here, GENERATED, PACKAGE), path.join(here, PACKAGE))
+ shutil.rmtree(path.join(here, PACKAGE + '.old'))
+
+
+if __name__ == '__main__':
+ if len(sys.argv) > 1:
+ url = sys.argv[1]
+ else:
+ url = HIVE_SERVER2_URL
+ main(url)

pyhive/common.py

@@ -8,14 +8,14 @@
 from builtins import bytes
 from builtins import int
 from builtins import object
-from builtins import range
 from builtins import str
 from past.builtins import basestring
 from pyhive import exc
 import abc
 import collections
 import time
 from future.utils import with_metaclass
+from itertools import islice
 
 
 class DBAPICursor(with_metaclass(abc.ABCMeta, object)):
@@ -124,14 +124,7 @@ def fetchmany(self, size=None):
  """
  if size is None:
  size = self.arraysize
- result = []
- for _ in range(size):
- one = self.fetchone()
- if one is None:
- break
- else:
- result.append(one)
- return result
+ return list(islice(iter(self.fetchone, None), size))
 
  def fetchall(self):
  """Fetch all (remaining) rows of a query result, returning them as a sequence of sequences
@@ -140,14 +133,7 @@ def fetchall(self):
  An :py:class:`~pyhive.exc.Error` (or subclass) exception is raised if the previous call to
  :py:meth:`execute` did not produce any result set or no call was issued yet.
  """
- result = []
- while True:
- one = self.fetchone()
- if one is None:
- break
- else:
- result.append(one)
- return result
+ return list(iter(self.fetchone, None))
 
  @property
  def arraysize(self):

pyhive/hive.py

@@ -7,6 +7,11 @@
 
 from __future__ import absolute_import
 from __future__ import unicode_literals
+
+import datetime
+import re
+from decimal import Decimal
+
 from TCLIService import TCLIService
 from TCLIService import constants
 from TCLIService import ttypes
@@ -31,6 +36,31 @@
 
 _logger = logging.getLogger(__name__)
 
+_TIMESTAMP_PATTERN = re.compile(r'(\d+-\d+-\d+ \d+:\d+:\d+(\.\d{,6})?)')
+
+
+def _parse_timestamp(value):
+ if value:
+ match = _TIMESTAMP_PATTERN.match(value)
+ if match:
+ if match.group(2):
+ format = '%Y-%m-%d %H:%M:%S.%f'
+ # use the pattern to truncate the value
+ value = match.group()
+ else:
+ format = '%Y-%m-%d %H:%M:%S'
+ value = datetime.datetime.strptime(value, format)
+ else:
+ raise Exception(
+ 'Cannot convert "{}" into a datetime'.format(value))
+ else:
+ value = None
+ return value
+
+
+TYPES_CONVERTER = {"DECIMAL_TYPE": Decimal,
+ "TIMESTAMP_TYPE": _parse_timestamp}
+
 
 class HiveParamEscaper(common.ParamEscaper):
  def escape_string(self, item):
@@ -177,6 +207,14 @@ def sasl_factory():
  self._transport.close()
  raise
 
+ def __enter__(self):
+ """Transport should already be opened by __init__"""
+ return self
+
+ def __exit__(self, exc_type, exc_val, exc_tb):
+ """Call close"""
+ self.close()
+
  def close(self):
  """Close the underlying session and Thrift transport"""
  req = ttypes.TCloseSessionReq(sessionHandle=self._sessionHandle)
@@ -215,7 +253,7 @@ class Cursor(common.DBAPICursor):
  def __init__(self, connection, arraysize=1000):
  self._operationHandle = None
  super(Cursor, self).__init__()
- self.arraysize = arraysize
+ self._arraysize = arraysize
  self._connection = connection
 
  def _reset_state(self):
@@ -230,6 +268,19 @@ def _reset_state(self):
  finally:
  self._operationHandle = None
 
+ @property
+ def arraysize(self):
+ return self._arraysize
+
+ @arraysize.setter
+ def arraysize(self, value):
+ """Array size cannot be None, and should be an integer"""
+ default_arraysize = 1000
+ try:
+ self._arraysize = int(value) or default_arraysize
+ except TypeError:
+ self._arraysize = default_arraysize
+
  @property
  def description(self):
  """This read-only attribute is a sequence of 7-item sequences.
@@ -273,6 +324,12 @@ def description(self):
  ))
  return self._description
 
+ def __enter__(self):
+ return self
+
+ def __exit__(self, exc_type, exc_val, exc_tb):
+ self.close()
+
  def close(self):
  """Close the operation handle"""
  self._reset_state()
@@ -320,8 +377,10 @@ def _fetch_more(self):
  )
  response = self._connection.client.FetchResults(req)
  _check_status(response)
+ schema = self.description
  assert not response.results.rows, 'expected data in columnar format'
- columns = map(_unwrap_column, response.results.columns)
+ columns = [_unwrap_column(col, col_schema[1]) for col, col_schema in
+ zip(response.results.columns, schema)]
  new_data = list(zip(*columns))
  self._data += new_data
  # response.hasMoreRows seems to always be False, so we instead check the number of rows
@@ -402,7 +461,7 @@ def fetch_logs(self):
 #
 
 
-def _unwrap_column(col):
+def _unwrap_column(col, type_=None):
  """Return a list of raw values from a TColumn instance."""
  for attr, wrapper in iteritems(col.__dict__):
  if wrapper is not None:
@@ -414,6 +473,9 @@ def _unwrap_column(col):
  for b in range(8):
  if byte & (1 << b):
  result[i * 8 + b] = None
+ converter = TYPES_CONVERTER.get(type_, None)
+ if converter and type_:
+ result = [converter(row) if row else row for row in result]
  return result
  raise DataError("Got empty column value {}".format(col)) # pragma: no cover
 

.travis.yml

@@ -7,17 +7,14 @@ matrix:
  # One build pulls latest versions dynamically
  - python: 3.6
  env: CDH=cdh5 CDH_VERSION=5 PRESTO=RELEASE SQLALCHEMY=sqlalchemy
- # Others use pinned versions.
+ - python: 3.6
+ env: CDH=cdh5 CDH_VERSION=5.10.1 PRESTO=0.147 SQLALCHEMY=sqlalchemy==1.2.8
  - python: 3.5
- env: CDH=cdh5 CDH_VERSION=5.10.1 PRESTO=0.147 SQLALCHEMY=sqlalchemy==1.0.12
+ env: CDH=cdh5 CDH_VERSION=5.10.1 PRESTO=0.147 SQLALCHEMY=sqlalchemy==1.2.8
  - python: 3.4
- env: CDH=cdh5 CDH_VERSION=5.10.1 PRESTO=0.147 SQLALCHEMY=sqlalchemy==1.0.12
- - python: 2.7
- env: CDH=cdh5 CDH_VERSION=5.10.1 PRESTO=0.147 SQLALCHEMY=sqlalchemy==1.0.12
- # stale stuff we're still using / supporting
+ env: CDH=cdh5 CDH_VERSION=5.10.1 PRESTO=0.147 SQLALCHEMY=sqlalchemy==1.2.8
  - python: 2.7
- env: CDH=cdh5 CDH_VERSION=5.10.1 PRESTO=0.147 SQLALCHEMY=sqlalchemy==0.8.7
- # exclude: python 3 against old libries
+ env: CDH=cdh5 CDH_VERSION=5.10.1 PRESTO=0.147 SQLALCHEMY=sqlalchemy==1.2.8
 install:
  - ./scripts/travis-install.sh
  - pip install codecov

README.rst

@@ -1,5 +1,6 @@
 .. image:: https://travis-ci.org/dropbox/PyHive.svg?branch=master
  :target: https://travis-ci.org/dropbox/PyHive
+.. image:: https://img.shields.io/codecov/c/github/dropbox/PyHive.svg
 
 ======
 PyHive
@@ -100,9 +101,6 @@ PyHive works with
 - Python 2.7 / Python 3
 - For Presto: Presto install
 - For Hive: `HiveServer2 <https://cwiki.apache.org/confluence/display/Hive/Setting+up+HiveServer2>`_ daemon
-- For Python 3 + Hive + SASL, you currently need to install an unreleased version of ``thrift_sasl``
- (``pip install git+https://github.com/cloudera/thrift_sasl``).
- At the time of writing, the latest version of ``thrift_sasl`` was 0.2.1.
 
 Changelog
 =========
@@ -137,3 +135,10 @@ Run the following in an environment with Hive/Presto::
 
 WARNING: This drops/creates tables named ``one_row``, ``one_row_complex``, and ``many_rows``, plus a
 database called ``pyhive_test_database``.
+
+Updating TCLIService
+====================
+
+The TCLIService module is autogenerated using a ``TCLIService.thrift`` file. To update it, the
+``generate.py`` file can be used: ``python generate.py <TCLIServiceURL>``. When left blank, the
+version for Hive 2.3 will be downloaded.

dev_requirements.txt

@@ -11,7 +11,6 @@ pytest-timeout==1.2.0
 # actual dependencies: let things break if a package changes
 requests>=1.0.0
 sasl>=0.2.1
-sqlalchemy>=0.8.7
 thrift>=0.10.0
 #thrift_sasl>=0.1.0
 git+https://github.com/cloudera/thrift_sasl # Using master branch in order to get Python 3 SASL patches

generate.py

@@ -0,0 +1,52 @@
+"""
+This file can be used to generate a new version of the TCLIService package
+using a TCLIService.thrift URL.
+
+If no URL is specified, the file for Hive 2.3 will be downloaded.
+
+Usage:
+
+ python generate.py THRIFT_URL
+
+or
+
+ python generate.py
+"""
+import shutil
+import sys
+from os import path
+from urllib.request import urlopen
+import subprocess
+
+here = path.abspath(path.dirname(__file__))
+
+PACKAGE = 'TCLIService'
+GENERATED = 'gen-py'
+
+HIVE_SERVER2_URL = \
+ 'https://raw.githubusercontent.com/apache/hive/branch-2.3/service-rpc/if/TCLIService.thrift'
+
+
+def save_url(url):
+ data = urlopen(url).read()
+ file_path = path.join(here, url.rsplit('/', 1)[-1])
+ with open(file_path, 'wb') as f:
+ f.write(data)
+
+
+def main(hive_server2_url):
+ save_url(hive_server2_url)
+ hive_server2_path = path.join(here, hive_server2_url.rsplit('/', 1)[-1])
+
+ subprocess.call(['thrift', '-r', '--gen', 'py', hive_server2_path])
+ shutil.move(path.join(here, PACKAGE), path.join(here, PACKAGE + '.old'))
+ shutil.move(path.join(here, GENERATED, PACKAGE), path.join(here, PACKAGE))
+ shutil.rmtree(path.join(here, PACKAGE + '.old'))
+
+
+if __name__ == '__main__':
+ if len(sys.argv) > 1:
+ url = sys.argv[1]
+ else:
+ url = HIVE_SERVER2_URL
+ main(url)

pyhive/common.py

@@ -8,14 +8,14 @@
 from builtins import bytes
 from builtins import int
 from builtins import object
-from builtins import range
 from builtins import str
 from past.builtins import basestring
 from pyhive import exc
 import abc
 import collections
 import time
 from future.utils import with_metaclass
+from itertools import islice
 
 
 class DBAPICursor(with_metaclass(abc.ABCMeta, object)):
@@ -124,14 +124,7 @@ def fetchmany(self, size=None):
  """
  if size is None:
  size = self.arraysize
- result = []
- for _ in range(size):
- one = self.fetchone()
- if one is None:
- break
- else:
- result.append(one)
- return result
+ return list(islice(iter(self.fetchone, None), size))
 
  def fetchall(self):
  """Fetch all (remaining) rows of a query result, returning them as a sequence of sequences
@@ -140,14 +133,7 @@ def fetchall(self):
  An :py:class:`~pyhive.exc.Error` (or subclass) exception is raised if the previous call to
  :py:meth:`execute` did not produce any result set or no call was issued yet.
  """
- result = []
- while True:
- one = self.fetchone()
- if one is None:
- break
- else:
- result.append(one)
- return result
+ return list(iter(self.fetchone, None))
 
  @property
  def arraysize(self):

pyhive/hive.py

@@ -7,6 +7,11 @@
 
 from __future__ import absolute_import
 from __future__ import unicode_literals
+
+import datetime
+import re
+from decimal import Decimal
+
 from TCLIService import TCLIService
 from TCLIService import constants
 from TCLIService import ttypes
@@ -31,6 +36,31 @@
 
 _logger = logging.getLogger(__name__)
 
+_TIMESTAMP_PATTERN = re.compile(r'(\d+-\d+-\d+ \d+:\d+:\d+(\.\d{,6})?)')
+
+
+def _parse_timestamp(value):
+ if value:
+ match = _TIMESTAMP_PATTERN.match(value)
+ if match:
+ if match.group(2):
+ format = '%Y-%m-%d %H:%M:%S.%f'
+ # use the pattern to truncate the value
+ value = match.group()
+ else:
+ format = '%Y-%m-%d %H:%M:%S'
+ value = datetime.datetime.strptime(value, format)
+ else:
+ raise Exception(
+ 'Cannot convert "{}" into a datetime'.format(value))
+ else:
+ value = None
+ return value
+
+
+TYPES_CONVERTER = {"DECIMAL_TYPE": Decimal,
+ "TIMESTAMP_TYPE": _parse_timestamp}
+
 
 class HiveParamEscaper(common.ParamEscaper):
  def escape_string(self, item):
@@ -177,6 +207,14 @@ def sasl_factory():
  self._transport.close()
  raise
 
+ def __enter__(self):
+ """Transport should already be opened by __init__"""
+ return self
+
+ def __exit__(self, exc_type, exc_val, exc_tb):
+ """Call close"""
+ self.close()
+
  def close(self):
  """Close the underlying session and Thrift transport"""
  req = ttypes.TCloseSessionReq(sessionHandle=self._sessionHandle)
@@ -215,7 +253,7 @@ class Cursor(common.DBAPICursor):
  def __init__(self, connection, arraysize=1000):
  self._operationHandle = None
  super(Cursor, self).__init__()
- self.arraysize = arraysize
+ self._arraysize = arraysize
  self._connection = connection
 
  def _reset_state(self):
@@ -230,6 +268,19 @@ def _reset_state(self):
  finally:
  self._operationHandle = None
 
+ @property
+ def arraysize(self):
+ return self._arraysize
+
+ @arraysize.setter
+ def arraysize(self, value):
+ """Array size cannot be None, and should be an integer"""
+ default_arraysize = 1000
+ try:
+ self._arraysize = int(value) or default_arraysize
+ except TypeError:
+ self._arraysize = default_arraysize
+
  @property
  def description(self):
  """This read-only attribute is a sequence of 7-item sequences.
@@ -273,6 +324,12 @@ def description(self):
  ))
  return self._description
 
+ def __enter__(self):
+ return self
+
+ def __exit__(self, exc_type, exc_val, exc_tb):
+ self.close()
+
  def close(self):
  """Close the operation handle"""
  self._reset_state()
@@ -320,8 +377,10 @@ def _fetch_more(self):
  )
  response = self._connection.client.FetchResults(req)
  _check_status(response)
+ schema = self.description
  assert not response.results.rows, 'expected data in columnar format'
- columns = map(_unwrap_column, response.results.columns)
+ columns = [_unwrap_column(col, col_schema[1]) for col, col_schema in
+ zip(response.results.columns, schema)]
  new_data = list(zip(*columns))
  self._data += new_data
  # response.hasMoreRows seems to always be False, so we instead check the number of rows
@@ -402,7 +461,7 @@ def fetch_logs(self):
 #
 
 
-def _unwrap_column(col):
+def _unwrap_column(col, type_=None):
  """Return a list of raw values from a TColumn instance."""
  for attr, wrapper in iteritems(col.__dict__):
  if wrapper is not None:
@@ -414,6 +473,9 @@ def _unwrap_column(col):
  for b in range(8):
  if byte & (1 << b):
  result[i * 8 + b] = None
+ converter = TYPES_CONVERTER.get(type_, None)
+ if converter and type_:
+ result = [converter(row) if row else row for row in result]
  return result
  raise DataError("Got empty column value {}".format(col)) # pragma: no cover