Skip to content

Commit

Permalink
Added a class that implements read only of a file.
Browse files Browse the repository at this point in the history
The file is expected to contain a set of information pieces. One per line.
The class acts like a list, except that it's non-mutable..
  • Loading branch information
rohe committed Feb 10, 2024
1 parent 195e0c2 commit 7b6c921
Show file tree
Hide file tree
Showing 3 changed files with 122 additions and 1 deletion.
2 changes: 1 addition & 1 deletion example/flask_op/static/jwks.json
Original file line number Diff line number Diff line change
@@ -1 +1 @@
{"keys": [{"kty": "RSA", "use": "sig", "kid": "bXNmZXROQ3N2dDI2SWY5VlNWTG5yOXZqYlpLenVsalhwUWR5RW9BMHNCaw", "n": "uGVI-b6qr-OTc2knp7bpmDtiCQoWFXZ8mUV-SX0rCMtcc_IRmc_J7AfNEfnYk3dv0cKQK_Dgv3vicoeuf4KQ9ZZY-xI3bnRl9_HnhRpz_cJScDirkNKlsv8aQuYBO_gIiHp8B32YC0nx3BUQV5I6QGEiyG-lZT9PmXsUO1uKPPhny_vtQ6cUpvtuLySBu2ZYpaTDQqCv5Y6EKC49NYWhBB4B6f6TNKCoQTaxA8ZoM3lh7kFbu5DPEXKFAtuNiOtUNP7Ei9KfBtyBYSaZQBY8VkwAm1yKCA2sfv1mBwx0dT53MPJlNkoltf89mv1NM2OJPQAgGE6ygwGS2fyBLAn_bQ", "e": "AQAB"}, {"kty": "EC", "use": "sig", "kid": "U0pLNmFBRE4waDYyZG9ZdjNPb2pTZXAwZzdrbmpZdG0ya3lpaFJwZU9ncw", "crv": "P-256", "x": "DYUyBfiD53SEtUuKLjFCFpIkqyhbmBppAMjOat9qiY0", "y": "-SUSvVeOv7EA84qHLLEkDP24iZree-fomICuA4baeeA"}]}
{"keys": [{"kty": "RSA", "use": "sig", "kid": "bXNmZXROQ3N2dDI2SWY5VlNWTG5yOXZqYlpLenVsalhwUWR5RW9BMHNCaw", "e": "AQAB", "n": "uGVI-b6qr-OTc2knp7bpmDtiCQoWFXZ8mUV-SX0rCMtcc_IRmc_J7AfNEfnYk3dv0cKQK_Dgv3vicoeuf4KQ9ZZY-xI3bnRl9_HnhRpz_cJScDirkNKlsv8aQuYBO_gIiHp8B32YC0nx3BUQV5I6QGEiyG-lZT9PmXsUO1uKPPhny_vtQ6cUpvtuLySBu2ZYpaTDQqCv5Y6EKC49NYWhBB4B6f6TNKCoQTaxA8ZoM3lh7kFbu5DPEXKFAtuNiOtUNP7Ei9KfBtyBYSaZQBY8VkwAm1yKCA2sfv1mBwx0dT53MPJlNkoltf89mv1NM2OJPQAgGE6ygwGS2fyBLAn_bQ"}, {"kty": "EC", "use": "sig", "kid": "U0pLNmFBRE4waDYyZG9ZdjNPb2pTZXAwZzdrbmpZdG0ya3lpaFJwZU9ncw", "crv": "P-256", "x": "DYUyBfiD53SEtUuKLjFCFpIkqyhbmBppAMjOat9qiY0", "y": "-SUSvVeOv7EA84qHLLEkDP24iZree-fomICuA4baeeA"}]}
94 changes: 94 additions & 0 deletions src/idpyoidc/storage/listfile.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,94 @@
import logging
import os
import time

from filelock import FileLock

logger = logging.getLogger(__name__)


class ReadOnlyListFile(object):

def __init__(self, file_name):
self.file_name = file_name
self.fmtime = 0
self.lst = None

if not os.path.exists(file_name):
fp = open(file_name, "x")
fp.close()


def __getitem__(self, item):
if self.is_changed(self.file_name):
self.lst = self._read_info(self.file_name)
return self.lst[item]

def __len__(self):
if self.is_changed(self.file_name):
self.lst = self._read_info(self.file_name)
if self.lst is None or self.lst == []:
return 0

return len(self.lst)

@staticmethod
def get_mtime(fname):
"""
Find the time this file was last modified.
:param fname: File name
:return: The last time the file was modified.
"""
try:
mtime = os.stat(fname).st_mtime_ns
except OSError:
# The file might be right in the middle of being written
# so sleep
time.sleep(1)
mtime = os.stat(fname).st_mtime_ns

return mtime

def is_changed(self, fname):
"""
Find out if this file has been modified since last
:param fname: A file name
:return: True/False
"""
if os.path.isfile(fname):
mtime = self.get_mtime(fname)

if self.fmtime == 0:
self.fmtime = mtime
return True

if mtime > self.fmtime: # has changed
self.fmtime = mtime
return True
else:
return False
else:
logger.error("Could not access {}".format(fname))
raise FileNotFoundError()

def _read_info(self, fname):
if os.path.isfile(fname):
try:
lock = FileLock(f"{fname}.lock")
with lock:
fp = open(fname, "r")
info = [x.strip() for x in fp.readlines()]
lock.release()
if info == "":
return None
else:
return info
except Exception as err:
logger.error(err)
raise
else:
_msg = f"No such file: '{fname}'"
logger.error(_msg)
return None
27 changes: 27 additions & 0 deletions tests/test_14_read_only_list_file.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,27 @@
import os

from idpyoidc.storage.listfile import ReadOnlyListFile

BASEDIR = os.path.abspath(os.path.dirname(__file__))


def full_path(local_file):
return os.path.join(BASEDIR, local_file)

FILE_NAME = full_path("read_only")
def test_read_only_list_file():
if os.path.exists(FILE_NAME):
os.unlink(FILE_NAME)
os.unlink(f"{FILE_NAME}.lock")

_read_only = ReadOnlyListFile(FILE_NAME)
assert len(_read_only) == 0

with open(FILE_NAME, "w") as fp:
for item in ["one", "two", "three"]:
fp.write(item)
fp.write("\n")

assert len(_read_only) == 3
assert set(_read_only) == {"one", "two", "three"}
assert _read_only[-1] == "three"

0 comments on commit 7b6c921

Please sign in to comment.