43 lines
1.5 KiB
Python
43 lines
1.5 KiB
Python
from typing import Tuple, Optional, Iterable
|
|
|
|
from v25.messaging.message import Message
|
|
from v25.messaging.subject import Subject
|
|
from v25.storage.storage import EventStorage
|
|
|
|
|
|
class SecureStorage(EventStorage):
|
|
def __init__(self, storage: EventStorage, subject: Subject):
|
|
self.storage = storage
|
|
self.subject = subject
|
|
|
|
def asrt(self):
|
|
return self.subject.dumps()
|
|
|
|
def check(self, subject: Subject) -> dict:
|
|
assert self.subject == subject, self.asrt()
|
|
return self.storage.check(subject)
|
|
|
|
def push(self, m: Message) -> None:
|
|
assert self.subject == m.sfrom, self.asrt()
|
|
return self.storage.push(m)
|
|
|
|
def edit(self, old: Message, new: Message):
|
|
assert self.subject == old.sfrom, self.asrt()
|
|
return self.storage.edit(old, new)
|
|
|
|
def delete(self, m: Message):
|
|
assert self.subject in m.pair, self.asrt()
|
|
return self.storage.delete(m)
|
|
|
|
def pull(self, pair: Tuple[Subject, Subject], params: Optional[dict] = None) -> Iterable[Message]:
|
|
assert self.subject in pair, self.asrt()
|
|
return self.storage.pull(pair, params)
|
|
|
|
def flags(self, m: Message, flags: str):
|
|
assert self.subject in m.pair, self.asrt()
|
|
return self.storage.flags(m, flags)
|
|
|
|
def events(self, sfrom: Subject, sto: Subject, after: Optional[bytes]) -> Iterable[Tuple[bytes, bytes]]:
|
|
assert self.subject == sto, self.asrt()
|
|
return self.storage.events(sfrom, sto, after)
|