from json import JSONDecodeError, JSONDecoder
from typing import Optional, Type
import requests
from eventsourcing.application.notificationlog import (
AbstractNotificationLog,
LocalNotificationLog,
Section,
)
from eventsourcing.utils.transcoding import ObjectJSONDecoder, ObjectJSONEncoder
[docs]class RemoteNotificationLog(AbstractNotificationLog):
"""
Presents notification log sections retrieved an HTTP API
that presents notification log sections in JSON format,
for example by using a NotificationLogView.
"""
[docs] def __init__(
self, base_url: str, json_decoder_class: Optional[Type[JSONDecoder]] = None
):
"""
Initialises remote notification log object.
:param str base_url: A URL for the HTTP API.
:param JSONDecoder json_decoder_class: used to deserialize remote sections.
"""
self.base_url = base_url
json_decoder_class_ = json_decoder_class or ObjectJSONDecoder
self.json_decoder = json_decoder_class_()
self._section_size = -1
def json_loads(self, value: str) -> object:
try:
return self.json_decoder.decode(value)
except JSONDecodeError:
raise ValueError("Couldn't load JSON string: {}".format(value))
@property
def section_size(self) -> int:
"""
Size of section of notification log.
"""
if self._section_size == -1:
resource = self.get_resource(self.make_notification_log_url("section_size"))
self._section_size = self.deserialize_section_size(resource)
return self._section_size
def deserialize_section_size(self, resource: str) -> int:
try:
section_size = self.json_loads(resource)
except:
raise ValueError(
"Failed to decode JSON 'section_size' resource: {}".format(resource)
)
else:
if not isinstance(section_size, int):
raise TypeError(
"Section size is not an int: {}".format(type(section_size))
)
return section_size
[docs] def __getitem__(self, section_id: str) -> Section:
"""
Returns a section of notification log.
:param str section_id: ID of the section of the notification log.
:return: Identified section of notification log.
:rtype: Section
"""
section_json = self.get_json(section_id)
return self.deserialize_section(section_json)
def deserialize_section(self, section_json: str) -> Section:
try:
obj = self.json_loads(section_json)
assert isinstance(obj, dict)
section = Section(**obj)
except ValueError as e:
raise ValueError(
"Couldn't deserialize notification log section: "
"{}: {}".format(e, section_json)
)
return section
def get_json(self, section_id: str) -> str:
notification_log_url = self.make_notification_log_url(section_id)
return self.get_resource(notification_log_url)
def make_notification_log_url(self, section_id: str) -> str:
return "{}/{}/".format(self.base_url.strip("/"), section_id)
def get_resource(self, url: str) -> str:
return requests.get(url).content.decode("utf8")
[docs]class NotificationLogView(object):
"""
Presents sections of a notification log in JSON format.
Can be used to make an HTTP API that can be used
remotely, for example by a RemoteNotificationLog.
"""
[docs] def __init__(
self, notification_log: LocalNotificationLog, json_encoder: ObjectJSONEncoder
):
"""
Initialises notification log view object.
:param notification_log: A notification log object
:param json_encoder_class: JSON encoder class
"""
assert isinstance(notification_log, LocalNotificationLog), type(
notification_log
)
self.notification_log = notification_log
self.json_encoder = json_encoder
[docs] def present_resource(self, name: str) -> bytes:
"""
Returns a resource of the notification log in JSON format.
Supports returning section of the notification log.
Also supports returning section_size of notification
log, if section_id has special value 'section_size'.
:param name: Name of the resource, e.g. a section ID.
:return: Identified resource of notification log view in JSON format.
"""
if name == "section_size":
# Present the notification log's configured section size.
section_size = self.notification_log.section_size
resource = self.json_encoder.encode(section_size)
else:
# Default to assuming the resource is a section.
section = self.notification_log[name]
resource = self.json_encoder.encode(section.__dict__)
return resource