Skip to content

Instantly share code, notes, and snippets.

@kelvneo
Created May 28, 2017 11:57
Show Gist options
  • Save kelvneo/f81d39095e3a70510b31010366c43dc0 to your computer and use it in GitHub Desktop.
Save kelvneo/f81d39095e3a70510b31010366c43dc0 to your computer and use it in GitHub Desktop.
This gist is to handle multiple files writes that will create new files on a daily basis.
#!/usr/bin/env python
import os
import time
import logging
logging.basicConfig(level=logging.DEBUG)
import random
class DayBasedMultiFileHandler():
""" Create the new object. This class only supports `with DayBasedMultiFileHandler(...) as f:` creation.
"""
def __init__(self, prepends, path='.', append='', mode='a', headers='', footers=''):
"""Create the object
Arguments:
prepends: The list of prependable strings for each file.
path: The path to store all the files
append: The suffix to append at the end of the file (extensions).
mode: The file modes for each file.
headers: Additional content for each file every time a new file is created.
footers: Additional content for each file before the file is closed per day.
"""
self._prepends = prepends
self._path = path
self._append = append
self._mode = mode
self._day = time.localtime().tm_mday
self._headers = headers
self._footers = footers
def _create_file(self, prepend):
""" Create the daily file with a prefix.
"""
name = self._format_name(prepend)
logging.info('Creating new file: %s', name)
_file = open(name, self._mode)
if self._headers:
_file.write(self._headers)
return _file
def _format_name(self, prepend):
""" Get a formatted name for the file based on the prefix, date, and suffix.
"""
return os.path.join(self._path, "{}{}{}".format(prepend, time.strftime('%y%m%d'), self._append))
def _day_changed(self):
""" Check if day has changed.
"""
return self._day != time.localtime().tm_mday
def __enter__(self):
self._files = {key: self._create_file(key) for key in self._prepends}
return self
def write(self, prepend, *args):
""" Write to a specific file.
"""
if self._day_changed(): # Check if the day has changed.
# If day has changed, close all old files and create new files.
for _file in self._files:
if self._footers:
self._files[_file].write(self._footers)
self._files[_file].close()
self._files = {key: self._create_file(key) for key in self._prepends}
# Attempt to write the data given by the user to the specified file.
return getattr(self._files[prepend], 'write')(*args)
def __exit__(self, *args):
for _file in self._files:
ret = getattr(self._files[_file], '__exit__')(*args)
if not ret:
return ret
return True
if __name__ == '__main__':
# For testing purposes only.
file_pre = [str(i) + '-' for i in range(10)]
count = 0
with DayBasedMultiFileHandler(file_pre, mode='w', append='.txt', headers='TESTING\n') as f:
while count != 1000000:
num = random.randrange(1000000000)
f.write(file_pre[num % 10], str(num) + '\n')
count += 1
if count % 10000 == 0:
logging.info('%d lines recorded!', count)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment