-
Notifications
You must be signed in to change notification settings - Fork 2
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Permanently delete trashed objects #56
Labels
Comments
Draft: class Trashable:
"""
A trashed object is permanently deleted after some retention time.
.. attribute:: RETENTION
Duration until a trashed object will be permanently deleted.
"""
RETENTION: timedelta
def delete(self) -> None:
"""Subclass API: Permanently delete the object."""
# def trash(self):
# r.zadd('trash', {self.id: time() + Trashable.RETENTION})
# def restore(self):
# r.zrem('trash', self.id)
# @staticmethod
# async def _delete_trashed(self) -> None:
# while True:
# id = await bzpoptimed(self.r, 'trash')
# object = r.oget(id, expect=expect_type(Trashable))
# object.delete()
class Gone:
""":ref:`Object` that does not exist anymore."""
class Event:
"""
.. attribute:: object
:ref:`Gone` if the object does not exist anymore.
""" |
Weak collection draft (needed later): _O = TypeVar('_O', bound=Object)
_G = TypeVar('_G')
class Collection(Sequence[Union[_O, _G]]):
"""
.. attribute:: no_ref
Function of the form `no_ref(id: str) -> _G` which is responsible to produce a value to
return if *id* does not exist anymore. Defaults to :meth:`strong`.
"""
@staticmethod
def strong(id: str) -> NoReturn:
"""Ensure strong references, i.e. fail with a :exc:`ReferenceError`."""
# raise ReferenceError()
pass
@staticmethod
def weak(id: str) -> Gone:
"""Use :cls:`Gone` for broken references."""
# return Gone()
pass
def __init__(self, *, no_ref: Callable[[str], _G] = strong,
expect: ExpectFunc[_O] = expect_type(Object)) -> None:
pass
@overload
def __getitem__(self, key: Union[int, str]) -> Union[_O, _G]:
pass
@overload
def __getitem__(self, key: slice) -> List[Union[_O, _G]]:
pass
# def __getitem__(self, key: Union[int, slice, str]) -> Union[_O, _G, List[Union[_O, _G]]]:
# if isinstance(key, str):
# item = self.app.r.oget(key, expect=self.expect)
# return self.no_ref(key) if item is None else item Design discussion:
class Collection(Sequence[Union[_O, _G]]):
# https://github.com/python/mypy/issues/1020
@overload
@classmethod
def new(cls, *, weak: Literal[False] = False,
expect: ExpectFunc[_O] = expect_type(Object)) -> Collection[_O, NoReturn]:
pass
@overload
@classmethod
def new(cls, *, weak: Literal[True],
expect: ExpectFunc[_O] = expect_type(Object)) -> Collection[_O, Gone]:
pass
# @classmethod
# def new(cls, *, weak: bool = False,
# expect: ExpectFunc[_O] = expect_type(Object)) -> Union[Collection[_O, NoReturn], Collection[_O, Gone]]:
# return Collection(weak, expect)
def __init__(self, _weak: bool, _expect: ExpectFunc[_O]) -> None:
pass
@overload
def __getitem__(self, key: int) -> Union[_O, _G]:
pass
@overload
def __getitem__(self, key: slice) -> List[Union[_O, _G]]:
pass
# def __getitem__(self, key: Union[int, slice, str]) -> Union[_O, _G, List[Union[_O, _G]]]:
# if isinstance(key, str):
# item = self.app.r.oget(key, default=Gone() if self.weak else ReferenceError,
# expect=self.expect)
# # item is Union[_O, Gone] and _G is Union[NoReturn, Gone] ensured via new()
# return cast(Union[_O, _G], item)
_T = TypeVar('_T')
class Collection(Sequence[_T]):
@overload
@staticmethod
def fget(*, weak: Literal[False] = False,
expect: ExpectFunc[_O] = expect_type(Object)) -> Callable[[Union[int, slice, str], Application], _O]:
pass
@overload
@staticmethod
def fget(*, weak: Literal[True],
expect: ExpectFunc[_O] = expect_type(Object)) -> Callable[[Union[int, slice, str], Application], Union[_O, Gone]]:
pass
# @staticmethod
# def fget(*, weak: bool = False,
# expect: ExpectFunc[_O] = expect_type(Object)) -> Union[Callable[[Union[int, slice, str], Application], _O], Callable[[Union[int, slice, str], Application], Union[_O, Gone]]]:
# def _f(key: Union[int, slice, str], app: Application) -> Union[_O, Union[_O, Gone]]:
# if isinstance(key, str):
# return app.r.oget(key, default=Gone() if self.weak else ReferenceError,
# expect=expect)
# return _f
def __init__(self, *,
get: Callable[[Union[int, slice, str], Application], _T] = fget()) -> None:
pass
@overload
def __getitem__(self, key: Union[int, str]) -> _T:
pass
@overload
def __getitem__(self, key: slice) -> List[_T]:
pass
# def __getitem__(self, key: Union[int, slice, str]) -> Union[_T, List[_T]]:
# return self.get(key, self.app) |
Depends on #57. |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Permanently delete trashed objects after some retention time.
The text was updated successfully, but these errors were encountered: