Source code for faust.types.assignor

import abc
import typing
from typing import Any, List, MutableMapping, Set

from mode import ServiceT
from yarl import URL

from .tuples import TP

if typing.TYPE_CHECKING:
    from .app import AppT as _AppT
else:

    class _AppT: ...  # noqa


__all__ = [
    "TopicToPartitionMap",
    "HostToPartitionMap",
    "PartitionAssignorT",
    "LeaderAssignorT",
]

TopicToPartitionMap = MutableMapping[str, List[int]]
HostToPartitionMap = MutableMapping[str, TopicToPartitionMap]



[docs]class PartitionAssignorT(abc.ABC):
    replicas: int
    app: _AppT

    @abc.abstractmethod
    def __init__(self, app: _AppT, replicas: int = 0) -> None: ...


[docs]    @abc.abstractmethod
    def group_for_topic(self, topic: str) -> int: ...



[docs]    @abc.abstractmethod
    def assigned_standbys(self) -> Set[TP]: ...



[docs]    @abc.abstractmethod
    def assigned_actives(self) -> Set[TP]: ...



[docs]    @abc.abstractmethod
    def is_active(self, tp: TP) -> bool: ...



[docs]    @abc.abstractmethod
    def is_standby(self, tp: TP) -> bool: ...



[docs]    @abc.abstractmethod
    def key_store(self, topic: str, key: bytes) -> URL: ...



[docs]    @abc.abstractmethod
    def table_metadata(self, topic: str) -> HostToPartitionMap: ...



[docs]    @abc.abstractmethod
    def tables_metadata(self) -> HostToPartitionMap: ...




[docs]class LeaderAssignorT(ServiceT):
    app: _AppT

    @abc.abstractmethod
    def __init__(self, app: _AppT, **kwargs: Any) -> None: ...


[docs]    @abc.abstractmethod
    def is_leader(self) -> bool: ...