class IstosRouter:
"""
A router to group Istos decorators.
Routes defined here will be applied to the main Istos app when
`istos.include_router(router)` is called.
"""
def __init__(self, prefix: str = ""):
self.prefix = prefix
self._actions: List[Callable[["Istos"], None]] = []
def _apply_prefix(self, prefix: str) -> str:
"""Combines the router's prefix with the endpoint's prefix."""
if self.prefix:
base = self.prefix.rstrip('/')
sub = prefix.lstrip('/')
return f"{base}/{sub}" if base and sub else (base or sub)
return prefix
def handle(self, prefix: str, serializer: Optional[Serialize] = None, retry: Optional[Union[int, RetryPolicy]] = None, durability: str = "at_most_once", authorizer: Optional[Authorizer] = None, http: Optional[Union[bool, str]] = None) -> Callable:
full_prefix = self._apply_prefix(prefix)
def decorator(func: Callable) -> Callable:
proxy = RouterProxy(func.__name__)
def action(app: "Istos"):
proxy._real_wrapper = app.handle(full_prefix, serializer=serializer, retry=retry, durability=durability, authorizer=authorizer, http=http)(func)
self._actions.append(action)
return proxy
return decorator
def query(self, prefix: str, timeout_s: float = 5.0, retry: Optional[Union[int, RetryPolicy]] = None, serializer: Optional[Serialize] = None, token: Optional[Union[bytes, str]] = None) -> Callable:
full_prefix = self._apply_prefix(prefix)
def decorator(func: Callable) -> Callable:
proxy = RouterProxy(func.__name__)
def action(app: "Istos"):
proxy._real_wrapper = app.query(full_prefix, timeout_s=timeout_s, retry=retry, serializer=serializer, token=token)(func)
self._actions.append(action)
return proxy
return decorator
def stream(self, prefix: str, serializer: Optional[Serialize] = None, authorizer: Optional[Authorizer] = None, http: Optional[Union[bool, str]] = None, http_timeout_s: float = 60.0) -> Callable:
full_prefix = self._apply_prefix(prefix)
def decorator(func: Callable) -> Callable:
proxy = RouterProxy(func.__name__)
def action(app: "Istos"):
proxy._real_wrapper = app.stream(full_prefix, serializer=serializer, authorizer=authorizer, http=http, http_timeout_s=http_timeout_s)(func)
self._actions.append(action)
return proxy
return decorator
def channel(self, prefix: str, serializer: Optional[Serialize] = None, authorizer: Optional[Authorizer] = None, ws: Optional[Union[bool, str]] = None, durable: bool = False) -> Callable:
full_prefix = self._apply_prefix(prefix)
def decorator(func: Callable) -> Callable:
proxy = RouterProxy(func.__name__)
def action(app: "Istos"):
proxy._real_wrapper = app.channel(full_prefix, serializer=serializer, authorizer=authorizer, ws=ws, durable=durable)(func)
self._actions.append(action)
return proxy
return decorator
def stream_client(self, prefix: str, serializer: Optional[Serialize] = None, timeout_s: float = 60.0, token: Optional[Union[bytes, str]] = None) -> Callable:
full_prefix = self._apply_prefix(prefix)
def decorator(func: Callable) -> Callable:
proxy = RouterProxy(func.__name__)
def action(app: "Istos"):
proxy._real_wrapper = app.stream_client(full_prefix, serializer=serializer, timeout_s=timeout_s, token=token)(func)
self._actions.append(action)
return proxy
return decorator
def channel_client(self, prefix: str, serializer: Optional[Serialize] = None, timeout_s: float = 5.0, token: Optional[Union[bytes, str]] = None) -> Callable:
full_prefix = self._apply_prefix(prefix)
def decorator(func: Callable) -> Callable:
proxy = RouterProxy(func.__name__)
def action(app: "Istos"):
proxy._real_wrapper = app.channel_client(full_prefix, serializer=serializer, timeout_s=timeout_s, token=token)(func)
self._actions.append(action)
return proxy
return decorator
def worker(self, prefix: str, concurrency: int = 1, poll_interval_s: float = 1.0, serializer: Optional[Serialize] = None, token: Optional[Union[bytes, str]] = None) -> Callable:
full_prefix = self._apply_prefix(prefix)
def decorator(func: Callable) -> Callable:
proxy = RouterProxy(func.__name__)
def action(app: "Istos"):
proxy._real_wrapper = app.worker(full_prefix, concurrency=concurrency, poll_interval_s=poll_interval_s, serializer=serializer, token=token)(func)
self._actions.append(action)
return proxy
return decorator
def publish(self, prefix: str, use_shm: bool = False, serializer: Optional[Serialize] = None, durable: bool = False, cache: int = 1000, heartbeat: float = 1.0, persist: Optional[str] = None) -> Callable:
full_prefix = self._apply_prefix(prefix)
def decorator(func: Callable) -> Callable:
proxy = RouterProxy(func.__name__)
def action(app: "Istos"):
proxy._real_wrapper = app.publish(full_prefix, use_shm=use_shm, serializer=serializer, durable=durable, cache=cache, heartbeat=heartbeat, persist=persist)(func)
self._actions.append(action)
return proxy
return decorator
def subscribe(self, prefix: str, retry: Optional[Union[int, RetryPolicy]] = None, serializer: Optional[Serialize] = None, durable: bool = False, replay: int = 1000, recover: bool = True, authorizer: Optional[Authorizer] = None, replay_persisted: bool = False, dedup: Union[bool, int] = False) -> Callable:
full_prefix = self._apply_prefix(prefix)
def decorator(func: Callable) -> Callable:
proxy = RouterProxy(func.__name__)
def action(app: "Istos"):
proxy._real_wrapper = app.subscribe(full_prefix, retry=retry, serializer=serializer, durable=durable, replay=replay, recover=recover, authorizer=authorizer, replay_persisted=replay_persisted, dedup=dedup)(func)
self._actions.append(action)
return proxy
return decorator
def on_liveliness(self, prefix: str) -> Callable:
full_prefix = self._apply_prefix(prefix)
def decorator(func: Callable) -> Callable:
proxy = RouterProxy(func.__name__)
def action(app: "Istos"):
proxy._real_wrapper = app.on_liveliness(full_prefix)(func)
self._actions.append(action)
return proxy
return decorator
def declare_liveliness(self, prefix: str) -> None:
full_prefix = self._apply_prefix(prefix)
def action(app: "Istos"):
app.declare_liveliness(full_prefix)
self._actions.append(action)