faust.tables

class faust.tables.Collection(app: faust.types.app.AppT, *, name: str = None, default: Callable[Any] = None, store: Union[str, yarl.URL] = None, key_type: Union[Type[faust.types.models.ModelT], Type[bytes], Type[str]] = None, value_type: Union[Type[faust.types.models.ModelT], Type[bytes], Type[str]] = None, partitions: int = None, window: faust.types.windows.WindowT = None, changelog_topic: faust.types.topics.TopicT = None, help: str = None, on_recover: Callable[Awaitable[None]] = None, on_changelog_event: Callable[faust.types.events.EventT, Awaitable[None]] = None, recovery_buffer_size: int = 1000, standby_buffer_size: int = None, extra_topic_configs: Mapping[str, Any] = None, **kwargs) → None

Base class for changelog-backed data structures stored in Kafka.

data

on_recover(fun: Callable[Awaitable[None]]) → Callable[Awaitable[None]]

Add function as callback to be called on table recovery.

Return type:Callable[[], Awaitable[None]]

info() → Mapping[str, Any]

Return type:Mapping[str, Any]

persisted_offset(tp: faust.types.tuples.TP) → Optional[int]

Return type:Optional[int]

reset_state() → None

Return type:None

join(*fields) → faust.types.streams.StreamT

Return type:StreamT[+T_co]

left_join(*fields) → faust.types.streams.StreamT

Return type:StreamT[+T_co]

inner_join(*fields) → faust.types.streams.StreamT

Return type:StreamT[+T_co]

outer_join(*fields) → faust.types.streams.StreamT

Return type:StreamT[+T_co]

clone(**kwargs) → Any

Return type:Any

combine(*nodes, **kwargs) → faust.types.streams.StreamT

Return type:StreamT[+T_co]

contribute_to_stream(active: faust.types.streams.StreamT) → None

Return type:None

label

Label used for graphs. :rtype: str

shortlabel

Label used for logging. :rtype: str

coroutine call_recover_callbacks(self) → None

Return type:None

logger = <Logger faust.tables.base (WARNING)>

coroutine need_active_standby_for(self, tp: faust.types.tuples.TP) → bool

Return type:bool

coroutine on_changelog_event(self, event: faust.types.events.EventT) → None

Return type:None

coroutine on_rebalance(self, assigned: Set[faust.types.tuples.TP], revoked: Set[faust.types.tuples.TP], newly_assigned: Set[faust.types.tuples.TP]) → None

Return type:None

coroutine on_recovery_completed(self, active_tps: Set[faust.types.tuples.TP], standby_tps: Set[faust.types.tuples.TP]) → None

Return type:None

coroutine on_start(self) → None

Called every time before the service is started/restarted.

Return type:None

coroutine remove_from_stream(self, stream: faust.types.streams.StreamT) → None

Return type:None

changelog_topic

Return type:TopicT[]

apply_changelog_batch(batch: Iterable[faust.types.events.EventT]) → None

Return type:None

class faust.tables.CollectionT(app: faust.types.tables.AppT, *, name: str = None, default: Callable[Any] = None, store: Union[str, yarl.URL] = None, key_type: faust.types.tables.ModelArg = None, value_type: faust.types.tables.ModelArg = None, partitions: int = None, window: faust.types.windows.WindowT = None, changelog_topic: faust.types.topics.TopicT = None, help: str = None, on_recover: Callable[Awaitable[None]] = None, on_changelog_event: Callable[faust.types.events.EventT, Awaitable[None]] = None, recovery_buffer_size: int = 1000, standby_buffer_size: int = None, extra_topic_configs: Mapping[str, Any] = None, **kwargs) → None

changelog_topic

Return type:TopicT[]

apply_changelog_batch(batch: Iterable[faust.types.events.EventT]) → None

Return type:None

persisted_offset(tp: faust.types.tuples.TP) → Optional[int]

Return type:Optional[int]

reset_state() → None

Return type:None

on_recover(fun: Callable[Awaitable[None]]) → Callable[Awaitable[None]]

Return type:Callable[[], Awaitable[None]]

coroutine call_recover_callbacks(self) → None

Return type:None

coroutine need_active_standby_for(self, tp: faust.types.tuples.TP) → bool

Return type:bool

coroutine on_changelog_event(self, event: faust.types.events.EventT) → None

Return type:None

coroutine on_rebalance(self, assigned: Set[faust.types.tuples.TP], revoked: Set[faust.types.tuples.TP], newly_assigned: Set[faust.types.tuples.TP]) → None

Return type:None

coroutine on_recovery_completed(self, active_tps: Set[faust.types.tuples.TP], standby_tps: Set[faust.types.tuples.TP]) → None

Return type:None

class faust.tables.TableManager(app: faust.types.app.AppT, **kwargs) → None

Manage tables used by Faust worker.

on_rebalance_start() → None

Return type:None

on_actives_ready() → None

Return type:None

on_standbys_ready() → None

Return type:None

changelog_topics

Return type:Set[str]

changelog_queue

Return type:ThrowableQueue

logger = <Logger faust.tables.manager (WARNING)>

coroutine on_rebalance(self, assigned: Set[faust.types.tuples.TP], revoked: Set[faust.types.tuples.TP], newly_assigned: Set[faust.types.tuples.TP]) → None

Return type:None

coroutine on_start(self) → None

Called every time before the service is started/restarted.

Return type:None

coroutine on_stop(self) → None

Called every time before the service is stopped/restarted.

Return type:None

recovery

Return type:Recovery[]

add(table: faust.types.tables.CollectionT) → faust.types.tables.CollectionT

Return type:CollectionT[]

on_partitions_revoked(revoked: Set[faust.types.tuples.TP]) → None

Return type:None

class faust.tables.TableManagerT(app: faust.types.tables.AppT, **kwargs) → None

add(table: faust.types.tables.CollectionT) → faust.types.tables.CollectionT

Return type:CollectionT[]

changelog_topics

Return type:Set[str]

coroutine on_rebalance(self, assigned: Set[faust.types.tuples.TP], revoked: Set[faust.types.tuples.TP], newly_assigned: Set[faust.types.tuples.TP]) → None

Return type:None

class faust.tables.Table(app: faust.types.app.AppT, *, name: str = None, default: Callable[Any] = None, store: Union[str, yarl.URL] = None, key_type: Union[Type[faust.types.models.ModelT], Type[bytes], Type[str]] = None, value_type: Union[Type[faust.types.models.ModelT], Type[bytes], Type[str]] = None, partitions: int = None, window: faust.types.windows.WindowT = None, changelog_topic: faust.types.topics.TopicT = None, help: str = None, on_recover: Callable[Awaitable[None]] = None, on_changelog_event: Callable[faust.types.events.EventT, Awaitable[None]] = None, recovery_buffer_size: int = 1000, standby_buffer_size: int = None, extra_topic_configs: Mapping[str, Any] = None, **kwargs) → None

Table (non-windowed).

class WindowWrapper(table: faust.types.tables.TableT, *, relative_to: Union[faust.types.tables.FieldDescriptorT, Callable[Optional[faust.types.events.EventT], Union[float, datetime.datetime]], datetime.datetime, float, None] = None, key_index: bool = False, key_index_table: faust.types.tables.TableT = None) → None

Windowed table wrapper.

A windowed table does not return concrete values when keys are accessed, instead WindowSet is returned so that the values can be further reduced to the wanted time period.

ValueType

alias of WindowSet

as_ansitable(title: str = '{table.name}', **kwargs) → str

Return type:str

clone(relative_to: Union[faust.types.tables.FieldDescriptorT, Callable[Optional[faust.types.events.EventT], Union[float, datetime.datetime]], datetime.datetime, float, None]) → faust.types.tables.WindowWrapperT

Return type:WindowWrapperT[]

get_relative_timestamp

Return type:Optional[Callable[[Optional[EventT[]]], Union[float, datetime]]]

get_timestamp(event: faust.types.events.EventT = None) → float

Return type:float

items(event: faust.types.events.EventT = None) → ItemsView

Return type:ItemsView[~KT, +VT_co]

key_index = False

key_index_table = None

keys() → KeysView

Return type:KeysView[~KT]

name

Return type:str

on_del_key(key: Any) → None

Return type:None

on_recover(fun: Callable[Awaitable[None]]) → Callable[Awaitable[None]]

Return type:Callable[[], Awaitable[None]]

on_set_key(key: Any, value: Any) → None

Return type:None

relative_to(ts: Union[faust.types.tables.FieldDescriptorT, Callable[Optional[faust.types.events.EventT], Union[float, datetime.datetime]], datetime.datetime, float, None]) → faust.types.tables.WindowWrapperT

Return type:WindowWrapperT[]

relative_to_field(field: faust.types.models.FieldDescriptorT) → faust.types.tables.WindowWrapperT

Return type:WindowWrapperT[]

relative_to_now() → faust.types.tables.WindowWrapperT

Return type:WindowWrapperT[]

relative_to_stream() → faust.types.tables.WindowWrapperT

Return type:WindowWrapperT[]

values(event: faust.types.events.EventT = None) → ValuesView

Return type:ValuesView[+VT_co]

using_window(window: faust.types.windows.WindowT, *, key_index: bool = False) → faust.types.tables.WindowWrapperT

Return type:WindowWrapperT[]

hopping(size: Union[datetime.timedelta, float, str], step: Union[datetime.timedelta, float, str], expires: Union[datetime.timedelta, float, str] = None, key_index: bool = False) → faust.types.tables.WindowWrapperT

Return type:WindowWrapperT[]

tumbling(size: Union[datetime.timedelta, float, str], expires: Union[datetime.timedelta, float, str] = None, key_index: bool = False) → faust.types.tables.WindowWrapperT

Return type:WindowWrapperT[]

on_key_get(key: KT) → None

Handle that key is being retrieved.

Return type:None

on_key_set(key: KT, value: VT) → None

Handle that value for a key is being set.

Return type:None

on_key_del(key: KT) → None

Handle that a key is deleted.

Return type:None

as_ansitable(title: str = '{table.name}', **kwargs) → str

Return type:str

logger = <Logger faust.tables.table (WARNING)>

class faust.tables.TableT(app: faust.types.tables.AppT, *, name: str = None, default: Callable[Any] = None, store: Union[str, yarl.URL] = None, key_type: faust.types.tables.ModelArg = None, value_type: faust.types.tables.ModelArg = None, partitions: int = None, window: faust.types.windows.WindowT = None, changelog_topic: faust.types.topics.TopicT = None, help: str = None, on_recover: Callable[Awaitable[None]] = None, on_changelog_event: Callable[faust.types.events.EventT, Awaitable[None]] = None, recovery_buffer_size: int = 1000, standby_buffer_size: int = None, extra_topic_configs: Mapping[str, Any] = None, **kwargs) → None

using_window(window: faust.types.windows.WindowT, *, key_index: bool = False) → faust.types.tables.WindowWrapperT

Return type:WindowWrapperT[]

hopping(size: Union[datetime.timedelta, float, str], step: Union[datetime.timedelta, float, str], expires: Union[datetime.timedelta, float, str] = None, key_index: bool = False) → faust.types.tables.WindowWrapperT

Return type:WindowWrapperT[]

tumbling(size: Union[datetime.timedelta, float, str], expires: Union[datetime.timedelta, float, str] = None, key_index: bool = False) → faust.types.tables.WindowWrapperT

Return type:WindowWrapperT[]

as_ansitable(**kwargs) → str

Return type:str