ich wollte heute meine kleine Lib, FileDB vorstellen. Ist nix großes, ich habe sie für meinen kleinen Blog erstellt und um überhaupt einmal ein kleines Projekt auf Github liegen zu haben. Vom Prinzip her wird auch nur via "pickle" ein dictionary in eine Binärdatei abgelegt und wieder geholt. dazu habe ich noch ein paar Methoden zum besseren handhaben erstellt.
Vielleicht gefällt es euch ja und der ein oder andere kann was mit anfangen. Vorrangig stell ich aber das Projekt hier vor um mir Tipps zu hohlen ob ich alles halbwegs korrekt gemacht habe und was man noch verbessern könnte.
Viel Spaß damit!
https://github.com/deadshox/FileDB
Code: Alles auswählen
# -*- coding: utf-8 -*-
"""
FileDB
------
FileDB is a small handler to use files as database.
Copyright (c) 2011 Daniel Brüggemann.
License: WTFPL2, see LICENSE.txt for details.
"""
__author__ = 'Daniel Brüggemann'
__version__ = '0.5'
__license__ = 'WTFPL2'
import os
import pickle
import zipfile
import time
import logging
import operator
import errno
import hashlib
class FileDB:
# -------------------------------------------------------------------------
def __init__(self, db, filebase='filebase', logname='filedb.log'):
self._filebase = filebase
self._path = os.path.join('.', filebase + os.sep)
self._bin = os.path.join(self._path, '{0}.bin'.format(db))
self.data = {}
self.checksum = ''
self.new_checksum = ''
# set logger
logging.basicConfig(filename=logname)
# create 'base' if not exist
try:
os.mkdir(self._path)
except OSError as e:
if e.errno == errno.EACCES:
logging.error('Permission denied to create Folder: "{0}".'\
.format(self._path))
raise
elif e.errno == errno.EEXIST:
logging.warning('Folder: "{0}" already exist.'\
.format(self._path))
else:
logging.warning('Create init path: "{0}".'.format(self._path))
self._bin = os.path.join(self._path, '{0}.bin'.format(db))
self.load()
# -------------------------------------------------------------------------
def load(self):
"""Load data from file."""
try:
with open(self._bin, 'rb') as handle:
self.data = pickle.load(handle)
except IOError as e:
if e.errno == errno.ENOENT:
try:
open(self._bin, 'w').close()
except IOError:
logging.error('Can\'t create new file in load(): "{0}".'\
.format(self._bin))
else:
self.checksum = self.get_checksum()
logging.warning('Create file: "{0}", given was not found.'\
.format(self._bin))
elif e.errno == errno.EACCES:
logging.error('Permission denied to given file: "{0}".'\
.format(self._bin))
except EOFError:
logging.warning('EOFError in given file: "{0}".'.format(self._bin))
else:
self.checksum = self.get_checksum()
# -------------------------------------------------------------------------
def reload(self):
"""Reload the data if checksum has changed."""
self.new_checksum = self.get_checksum()
if not self.checksum == self.new_checksum:
try:
with open(self._bin, 'rb') as handle:
self.data = pickle.load(handle)
except IOError:
logging.error('Can\'t open the given file in reload(): "{0}".'\
.format(self._bin))
# -------------------------------------------------------------------------
def save(self):
"""Save data to file."""
try:
with open(self._bin, 'wb') as handle:
pickle.dump(self.data, handle)
except IOError:
logging.error('Error while save file: "{0}" not found!'\
.format(self._bin))
raise
# -------------------------------------------------------------------------
def sort_by(self, sort='id'):
"""Sort data by key."""
l = []
for i in self.data:
l.append(self.data[i][sort])
l.sort()
l.reverse()
return l
# -------------------------------------------------------------------------
def get_checksum(self):
"""Get sha1 checksum of a given file."""
stat = os.stat(self._bin)
h = hashlib.sha1()
h.update('{0}{1}'.format(stat.st_size, stat.st_mtime))
return h.hexdigest()
# -------------------------------------------------------------------------
def create_id(self, data):
"""Creates an id as a hex digget from a integer."""
try:
return hex(int(data))[2:]
except (TypeError, ValueError):
logging.error('Given data must be a integer: "{0}".'.format(data))
# -------------------------------------------------------------------------
def del_db(self):
"""Delete the db."""
try:
os.remove(self._bin)
except OSError:
logging.error('Could not remove db: "{0}".'.format(self._bin))
# -------------------------------------------------------------------------
def del_base(self):
"""Delete the db base."""
try:
os.rmdir(self._path)
except OSError:
logging.error('Directory: "{0}" is not empty, please clear it.'\
.format(self._path))
# -------------------------------------------------------------------------
def backup(self, backup_path=''):
"""Backup the filebase or only a file."""
try:
os.mkdir(backup_path)
except OSError:
pass
else:
logging.warning('Backup path: "{0}" created.'.format(backup_path))
filename = time.strftime('%Y%m%d%H%M', time.localtime(time.time()))
f = os.path.join(backup_path, '{0}.zip'.format(filename))
with zipfile.ZipFile(f, 'w') as backup:
backup.write(self._filebase)
for bkp_file in os.listdir(self._path):
backup.write(os.path.join(self._filebase, bkp_file),
compress_type=zipfile.ZIP_DEFLATED)UPDATE: Version 0.2 - 18.07.2011
UPDATE: Version 0.3 - 18.07.2011
UPDATE: Version 0.4 - 26.07.2011
UPDATE: Version 0.5 - 28.07.2011
Aber danke für den Hinweis.