faust.transport.drivers.aiokafka
¶
Message transport using aiokafka.
-
class
faust.transport.drivers.aiokafka.
Consumer
(transport: faust.types.transports.TransportT, callback: Callable[faust.types.tuples.Message, Awaitable], on_partitions_revoked: Callable[Set[faust.types.tuples.TP], Awaitable[None]], on_partitions_assigned: Callable[Set[faust.types.tuples.TP], Awaitable[None]], *, commit_interval: float = None, commit_livelock_soft_timeout: float = None, loop: asyncio.events.AbstractEventLoop = None, **kwargs) → None[source]¶ Kafka consumer using aiokafka.
-
logger
= <Logger faust.transport.drivers.aiokafka (WARNING)>¶
-
RebalanceListener
¶ alias of
ConsumerRebalanceListener
-
fetch_timeout
= 10.0¶
-
consumer_stopped_errors
= (<class 'aiokafka.errors.ConsumerStoppedError'>,)¶
-
flow_active
= True¶
-
coroutine
create_topic
(self, topic: str, partitions: int, replication: int, *, config: Mapping[str, Any] = None, timeout: Union[datetime.timedelta, float, str] = 30.0, retention: Union[datetime.timedelta, float, str] = None, compacting: bool = None, deleting: bool = None, ensure_created: bool = False) → None[source]¶ Return type: None
-
coroutine
earliest_offsets
(self, *partitions) → MutableMapping[faust.types.tuples.TP, int][source]¶ Return type: MutableMapping
[TP
,int
]
-
getmany
(timeout: float) → AsyncIterator[Tuple[faust.types.tuples.TP, faust.types.tuples.Message]][source]¶ Return type: AsyncIterator
[Tuple
[TP
,Message
]]
-
coroutine
highwaters
(self, *partitions) → MutableMapping[faust.types.tuples.TP, int][source]¶ Return type: MutableMapping
[TP
,int
]
-
coroutine
on_restart
(self) → None[source]¶ Called every time when the service is restarted.
Return type: None
-
coroutine
on_start
(self) → None[source]¶ Called every time before the service is started/restarted.
Return type: None
-
coroutine
on_stop
(self) → None[source]¶ Called every time before the service is stopped/restarted.
Return type: None
-
coroutine
position
(self, tp: faust.types.tuples.TP) → Optional[int][source]¶ Return type: Optional
[int
]
-
coroutine
seek
(self, partition: faust.types.tuples.TP, offset: int) → None[source]¶ Return type: None
-
coroutine
seek_wait
(self, partitions: Mapping[faust.types.tuples.TP, int]) → None[source]¶ Return type: None
-
-
class
faust.transport.drivers.aiokafka.
Producer
(transport: faust.types.transports.TransportT, loop: asyncio.events.AbstractEventLoop = None, **kwargs) → None[source]¶ Kafka producer using aiokafka.
-
logger
= <Logger faust.transport.drivers.aiokafka (WARNING)>¶
-
coroutine
create_topic
(self, topic: str, partitions: int, replication: int, *, config: Mapping[str, Any] = None, timeout: Union[datetime.timedelta, float, str] = 20.0, retention: Union[datetime.timedelta, float, str] = None, compacting: bool = None, deleting: bool = None, ensure_created: bool = False) → None[source]¶ Return type: None
-
coroutine
on_restart
(self) → None[source]¶ Called every time when the service is restarted.
Return type: None
-
coroutine
on_start
(self) → None[source]¶ Called every time before the service is started/restarted.
Return type: None
-
coroutine
on_stop
(self) → None[source]¶ Called every time before the service is stopped/restarted.
Return type: None
-
coroutine
send
(self, topic: str, key: Optional[bytes], value: Optional[bytes], partition: Optional[int]) → Awaitable[faust.types.tuples.RecordMetadata][source]¶ Return type: Awaitable
[RecordMetadata
]
-
coroutine
send_and_wait
(self, topic: str, key: Optional[bytes], value: Optional[bytes], partition: Optional[int]) → faust.types.tuples.RecordMetadata[source]¶ Return type: RecordMetadata
-
-
class
faust.transport.drivers.aiokafka.
Transport
(*args, **kwargs) → None[source]¶ Kafka transport using aiokafka.
-
class
Consumer
(transport: faust.types.transports.TransportT, callback: Callable[faust.types.tuples.Message, Awaitable], on_partitions_revoked: Callable[Set[faust.types.tuples.TP], Awaitable[None]], on_partitions_assigned: Callable[Set[faust.types.tuples.TP], Awaitable[None]], *, commit_interval: float = None, commit_livelock_soft_timeout: float = None, loop: asyncio.events.AbstractEventLoop = None, **kwargs) → None¶ Kafka consumer using aiokafka.
-
RebalanceListener
¶ alias of
ConsumerRebalanceListener
-
close
() → None¶ Return type: None
-
consumer_stopped_errors
= (<class 'aiokafka.errors.ConsumerStoppedError'>,)¶
-
coroutine
create_topic
(self, topic: str, partitions: int, replication: int, *, config: Mapping[str, Any] = None, timeout: Union[datetime.timedelta, float, str] = 30.0, retention: Union[datetime.timedelta, float, str] = None, compacting: bool = None, deleting: bool = None, ensure_created: bool = False) → None¶ Return type: None
-
coroutine
earliest_offsets
(self, *partitions) → MutableMapping[faust.types.tuples.TP, int]¶ Return type: MutableMapping
[TP
,int
]
-
fetch_timeout
= 10.0¶
-
flow_active
= True¶
-
getmany
(timeout: float) → AsyncIterator[Tuple[faust.types.tuples.TP, faust.types.tuples.Message]]¶ Return type: AsyncIterator
[Tuple
[TP
,Message
]]
-
coroutine
highwaters
(self, *partitions) → MutableMapping[faust.types.tuples.TP, int]¶ Return type: MutableMapping
[TP
,int
]
-
logger
= <Logger faust.transport.drivers.aiokafka (WARNING)>¶
-
on_init
() → None¶ Return type: None
-
coroutine
on_restart
(self) → None¶ Called every time when the service is restarted.
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
-
pause_partitions
(tps: Iterable[faust.types.tuples.TP]) → None¶ Return type: None
-
coroutine
perform_seek
(self) → None¶ Return type: None
-
resume_flow
() → None¶ Return type: None
-
resume_partitions
(tps: Iterable[faust.types.tuples.TP]) → None¶ Return type: None
-
coroutine
seek
(self, partition: faust.types.tuples.TP, offset: int) → None¶ Return type: None
-
coroutine
seek_wait
(self, partitions: Mapping[faust.types.tuples.TP, int]) → None¶ Return type: None
-
stop_flow
() → None¶ Return type: None
-
coroutine
subscribe
(self, topics: Iterable[str]) → None¶ Return type: None
-
coroutine
threadsafe_partitions_assigned
(self, receiver_loop: asyncio.events.AbstractEventLoop, assigned: Set[faust.types.tuples.TP]) → None¶ Return type: None
-
coroutine
threadsafe_partitions_revoked
(self, receiver_loop: asyncio.events.AbstractEventLoop, revoked: Set[faust.types.tuples.TP]) → None¶ Return type: None
-
-
class
Producer
(transport: faust.types.transports.TransportT, loop: asyncio.events.AbstractEventLoop = None, **kwargs) → None¶ Kafka producer using aiokafka.
-
coroutine
create_topic
(self, topic: str, partitions: int, replication: int, *, config: Mapping[str, Any] = None, timeout: Union[datetime.timedelta, float, str] = 20.0, retention: Union[datetime.timedelta, float, str] = None, compacting: bool = None, deleting: bool = None, ensure_created: bool = False) → None¶ Return type: None
-
coroutine
flush
(self) → None¶ Return type: None
-
logger
= <Logger faust.transport.drivers.aiokafka (WARNING)>¶
-
on_init
() → None¶ Return type: None
-
coroutine
on_restart
(self) → None¶ Called every time when the service is restarted.
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
-
coroutine
send
(self, topic: str, key: Optional[bytes], value: Optional[bytes], partition: Optional[int]) → Awaitable[faust.types.tuples.RecordMetadata]¶ Return type: Awaitable
[RecordMetadata
]
-
coroutine
send_and_wait
(self, topic: str, key: Optional[bytes], value: Optional[bytes], partition: Optional[int]) → faust.types.tuples.RecordMetadata¶ Return type: RecordMetadata
-
coroutine
-
default_port
= 9092¶
-
driver_version
= 'aiokafka=0.4.19'¶
-
class