Skip to content

Instantly share code, notes, and snippets.

@AlmightyOatmeal
Last active May 8, 2018 22:43
Show Gist options
  • Select an option

  • Save AlmightyOatmeal/bc78ed034abbcdc79ea06131ba17ecf5 to your computer and use it in GitHub Desktop.

Select an option

Save AlmightyOatmeal/bc78ed034abbcdc79ea06131ba17ecf5 to your computer and use it in GitHub Desktop.
Fun with Python and Gzipped files! Works on Python 2.7.x.
import json
import gzip
import logging
import os
logger = logging.getLogger(os.path.splitext(os.path.basename(__file__))[0])
def write_gzip_file(path, data, overwrite_existing=False, compression_level=9):
"""Write a compressed gzip file.
:param path: Relative, or absolute, path of the file to write.
:type path: str or unicode
:param data: Data to write.
:type data: *
:param overwrite_existing: (optional) If the file, overwrite it if set to True otherwise do not write.
(default: False)
:param compression_level: (optional) Level of compression for gzip. (default: 9)
:type compression_level: int
:return: True if successful else False.
:rtype: bool
"""
path = os.path.abspath(path)
root_dir = os.path.split(path)[0]
if not os.path.exists(root_dir):
os.makedirs(root_dir)
if not path.lower().endswith('.gz'):
path = '{}.gz'.format(path)
if os.path.exists(path):
logger.warning('Output file already exists.')
if not overwrite_existing:
logger.warning('Overwrite existing is disabled.')
return False
try:
with gzip.open(path, 'w', compresslevel=compression_level) as f:
f.write(data)
logger.info('Successfully wrote "{}"'.format(os.path.split(path)[1]))
except Exception as err:
logger.exception(err)
return False
return True
def read_gzip_file(path, parse_json=False):
"""Read a compressed gzip file.
:param path: Relative, or absolute, path of the file to write.
:type path: str or unicode
:param parse_json: (optional) Whether or not to try to parse the JSON data.
:type parse_json: bool
:return: Data from the file or None
:rtype: *
"""
# logger = logging.getLogger('read_gzip_file')
path = os.path.abspath(path)
# if not os.path.exists(path):
# logger.critical('File not found: "{}"'.format(path))
# return None
try:
with gzip.open(path, 'rb') as f:
data = f.read().strip()
if parse_json:
return json.loads(data)
return data
except IOError:
if not path.lower().endswith('.gz'):
path = '{}.gz'.format(path)
return read_gzip_file(path, parse_json=parse_json)
except Exception as err:
logger.exception(err)
return None
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment