Несколько дней назад я начинаю рефакторинг и переопределение крошечного репозитория JSON, который я создаю на работе, основной причиной была быстрая реализация без базы данных и хранить несколько объектов.
Я думаю, что реализация может быть улучшенной, но она работает, и для первого аромата она была действительной, в следующие месяцы я буду продолжать работать над этим.
Реализация довольно проста, существует репозиторий, который использует контекст для хранения данных, в конце поста есть пример использования, и при следующих постах я подробно описываю другие способы использования.
import os
import json
import uuid
import threading
from ..errors.entity_not_found import EntityNotFound
class JsonContext(object):
def __init__(self, enity_name, filepath="./", key_field="id", key_generate_func=lambda: str(uuid.uuid4())):
self.file_path = os.path.join(filepath, "{0}s.json".format(enity_name))
self.entity_name = enity_name
self.session_values = []
if not os.path.exists(self.file_path):
self.commit()
self.key_field = key_field
self.key_generate_func = key_generate_func
self.lock = threading.Lock()
def open(self):
self.lock.acquire()
with open(self.file_path) as f:
data = json.load(f)
self.session_values = data["values"]
def close(self):
self.lock.release()
def __enter__(self):
self.open()
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.close()
def exists(self, identifier):
matches = list(filter(
lambda x: x[self.key_field] == identifier,
self.session_values))
return len(matches) > 0
def add(self, entity):
if (self.key_field in entity and self.exists(entity[self.key_field])):
self.delete(entity)
else:
entity[self.key_field] = self.key_generate_func()
self.session_values.append(entity)
return entity
def delete(self, entity):
self.session_values = list(filter(
lambda x: x[self.key_field] != entity[self.key_field],
self.session_values))
def commit(self):
with open(self.file_path, 'w+') as f:
f.write(json.dumps({"name": self.entity_name, "values" : self.session_values}))
def find(self, query_function):
return list(filter(query_function, self.session_values))
def get(self, identifier):
if not self.exists(identifier):
raise EntityNotFound("identifier not found")
return self.find(lambda x: x[self.key_field] == identifier)[0]
def get_all(self):
return self.session_values
from ..context.json_context import JsonContext
class BaseJsonRepository(object):
def __init__(self, entity_name, filepath="./"):
self.context = JsonContext(entity_name, filepath=filepath)
def __enter__(self):
self.context.open()
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.context.close()
def insert(self, entity):
return self.context.add(entity)
def update(self, entity):
return self.context.add(entity)
def delete(self, entity):
self.context.delete(entity)
def get_all(self):
return self.context.get_all()
def get(self, identifier):
return self.context.get(identifier)
def find(self, function):
return self.context.find(function)
Примеры … посмотрите на Тесты 🙂
from json_repository.repositories.base_json_repository import BaseJsonRepository
from json_repository.errors.entity_not_found import EntityNotFound
class FoobarRepository(BaseJsonRepository):
def __init__(self):
super(FoobarRepository, self).__init__("foo")
with FoobarRepository() as repo:
value = repo.insert({
"foo": "a foo value",
"bar":" a bar value"
})
repo.context.commit()
Любые отзывы об этой «игрушке» приветствуются, оставьте комментарий ниже:).
Оригинал: “https://dev.to/mandrewcito/creating-a-json-repository-1l14”