kev/deadsimplekv
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

added get_raw_json and fixed formatting, switched to setuptools

Browse Source
This commit is contained in:
Kevin Froman 2020-03-28 00:03:16 -05:00
parent a33a56e6ad
commit f6d3bac205
4 changed files with 72 additions and 50 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
CHANGELOG.md
View File

@ -2,6 +2,12 @@
This project uses Semantic Versioning
## 0.3.0
Added get_raw_json
Improved code formatting
Switched to setuptools
## 0.2.0
Added automatic creation of path for file writing

72
deadsimplekv/__init__.py
View File

@ -15,18 +15,22 @@
You should have received a copy of the GNU General Public License
along with this program. If not, see <https://www.gnu.org/licenses/>.
"""
import json, time, math, os, atexit
import json
import time
import math
import os
import atexit
def _is_serializable(data):
"""
Test if something is able to be in JSON format
"""
"""Test if something is able to be in JSON format"""
try:
json.dumps(data)
return True
except TypeError:
return False
class DeadSimpleKV:
def __init__(self, file_path=None, refresh_seconds=0, flush_seconds=0, flush_on_exit=True):
"""
@ -35,63 +39,61 @@ class DeadSimpleKV:
Refresh is an integer specifying how many seconds should pass before data is re-read from disk.
If refresh_seconds or flush_seconds are set to None they will not be done automatically (except where flush_on_exit applies)
"""
temp_time = DeadSimpleKV._get_epoch() # Initialization time
self.file_path = file_path # The file path where we write our data in JSON format
self.refresh_seconds = refresh_seconds # How often to refresh the
self.flush_seconds = flush_seconds # How often to flush out when setting
self._data = {} # The key store (in memory)
self._last_refresh = temp_time # time last refreshed from disk
self._last_flush = temp_time # time last flushed to disk
temp_time = DeadSimpleKV._get_epoch() # Initialization time
self.file_path = file_path # The file path where we write our data in JSON format
self.refresh_seconds = refresh_seconds # How often to refresh the
self.flush_seconds = flush_seconds # How often to flush out when setting
self._data = {} # The key store (in memory)
self._last_refresh = temp_time # time last refreshed from disk
self._last_flush = temp_time # time last flushed to disk
if not file_path is None:
abs_path = os.path.dirname(os.path.abspath(file_path))
if not os.path.exists(abs_path): os.makedirs(abs_path)
if not os.path.exists(abs_path):
os.makedirs(abs_path)
try:
if os.path.exists(file_path):
self.refresh()
except TypeError:
pass
if flush_on_exit:
atexit.register(self.flush)
def get(self, key):
"""
Accepts key, which must be json serializable
"""
"""Accepts key, which must be json serializable."""
self._do_auto_refresh()
try:
return self._data[key]
except KeyError:
return None
def get_raw_json(self) -> str:
"""Return raw json string of the data"""
return json.dumps(self._data)
def put(self, key, value):
"""Value setter. Will automatically flush if auto_flush is True and file_path is not None.
Will return value error if either key or value are not JSON serializable
"""
Value setter. Will automatically flush if auto_flush is True and file_path is not None
Will return value error if either key or value are not JSON serializable
"""
self._data[key] = value # Set the key
self._data[key] = value # Set the key
# Check if key and value can be put in JSON
if not _is_serializable(key):
raise ValueError('%s is not JSON serializable' % (key,))
if not _is_serializable(value):
raise ValueError('%s is not JSON serializable' % (value,))
raise ValueError('%s is not JSON serializable' % (value,))
self._do_auto_flush()
return (key, value)
def delete(self, key):
"""
Deletes value. Will automatically flush if auto_flush is True and file_path is not None
"""
"""Deletes value. Will automatically flush if auto_flush is True and file_path is not None."""
del self._data[key]
self._do_auto_flush()
def refresh(self):
"""
Refresh data and then mark time read. Can be manually called
"""
"""Refresh data and then mark time read. Can be manually called."""
try:
self._data = json.loads(DeadSimpleKV._read_in(self.file_path))
except FileNotFoundError:
@ -99,18 +101,16 @@ class DeadSimpleKV:
self._last_refresh = DeadSimpleKV._get_epoch()
def flush(self):
"""
Write out to file then mark time flushed. Can be manually called
"""
"""Write out to file then mark time flushed. Can be manually called."""
DeadSimpleKV._write_out(self.file_path, json.dumps(self._data))
self._last_flush = DeadSimpleKV._get_epoch()
def _do_auto_flush(self):
if self.flush_seconds is not None and self.file_path is not None:
# flush if automatic and it is time to do so
if DeadSimpleKV._get_epoch() - self._last_flush >= self.flush_seconds:
self.flush()
def _do_auto_refresh(self):
# Automatically flush if it is enabled and time to do so
if self.refresh_seconds is not None and self.file_path is not None:
@ -131,4 +131,4 @@ class DeadSimpleKV:
def _read_in(file_path):
"""Read in raw data"""
with open(file_path, 'r') as read_file:
return read_file.read()
return read_file.read()

12
setup.py
View File

@ -1,10 +1,16 @@
from distutils.core import setup
from setuptools import setup, find_packages
setup(name='deadsimplekv',
version='0.2.0',
version='0.3.0',
description='Very simple key-value store for Python',
author='Kevin Froman',
author_email='beardog@mailbox.org',
url='https://github.com/beardog108/deadsimplekv',
packages=['deadsimplekv'],
packages=find_packages(exclude=['contrib', 'docs', 'tests']),
install_requires=[],
classifiers=[
"Programming Language :: Python :: 3",
"License :: OSI Approved :: GNU General Public License v3 (GPLv3)",
"Operating System :: OS Independent",
],
)

32
tests/test_basic.py
View File

@ -26,17 +26,27 @@ class TestInit(unittest.TestCase):
def test_init(self):
kv = deadsimplekv.DeadSimpleKV(get_test_id())
def test_get(self):
test_id = get_test_id()
with open(test_id , 'w') as test_file:
test_file.write('{"my_key": "test"}')
# assert we can get written data
kv = deadsimplekv.DeadSimpleKV(test_id)
self.assertEqual(kv.get('my_key'), 'test')
def test_get_raw_json(self):
test_id = get_test_id()
with open(test_id , 'w') as test_file:
test_file.write('{"my_key": "test"}')
# assert we can get written data
kv = deadsimplekv.DeadSimpleKV(test_id)
self.assertEqual(kv.get_raw_json(), '{"my_key": "test"}')
def test_delete(self):
# test key deletion
test_id = get_test_id()
@ -52,10 +62,10 @@ class TestInit(unittest.TestCase):
kv.put('meme', 'doge')
kv2 = deadsimplekv.DeadSimpleKV(test_id)
self.assertIsNone(kv2.get('meme'))
def test_manual_refresh(self):
# Test manual refreshing
test_id = get_test_id()
test_id = get_test_id()
kv = deadsimplekv.DeadSimpleKV(test_id, refresh_seconds=None)
kv2 = deadsimplekv.DeadSimpleKV(test_id)
@ -68,7 +78,7 @@ class TestInit(unittest.TestCase):
kv = deadsimplekv.DeadSimpleKV(get_test_id())
# assert non existant value returns none
self.assertIsNone(kv.get('my_key'))
def test_invalid(self):
kv = deadsimplekv.DeadSimpleKV(get_test_id(), flush_on_exit=False)
try:
@ -83,7 +93,7 @@ class TestInit(unittest.TestCase):
pass
else:
self.fail()
def test_put(self):
# Assert data gets written successfully as json
test_id = get_test_id()
@ -93,9 +103,9 @@ class TestInit(unittest.TestCase):
with open(test_id, 'r') as test_file:
test_data = json.loads(test_file.read())
self.assertEqual(test_data['my_key'], 'test')
def test_flush_no_path(self):
# assert data gets flushed when the path is not made yet
@ -106,9 +116,9 @@ class TestInit(unittest.TestCase):
with open(test_id, 'r') as test_file:
test_data = json.loads(test_file.read())
self.assertEqual(test_data['my_key'], 'test2')
def test_time(self):
self.assertEqual(deadsimplekv.DeadSimpleKV._get_epoch(), math.floor(time.time()))