faust.transport.drivers.aiokafka

Message transport using aiokafka.

class faust.transport.drivers.aiokafka.Consumer(*args, **kwargs) → None[source]

Kafka consumer using aiokafka.

logger = <Logger faust.transport.drivers.aiokafka (WARNING)>
RebalanceListener

alias of ConsumerRebalanceListener

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[source]
Return type

None

coroutine on_stop(self) → None[source]

Service is being stopped/restarted.

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)>
allow_headers = True
key_partition(topic: str, key: bytes) → faust.types.tuples.TP[source]
Return type

TP

supports_headers() → bool[source]
Return type

bool

coroutine abort_transaction(self, transactional_id: str) → None[source]
Return type

None

coroutine begin_transaction(self, transactional_id: str) → None[source]
Return type

None

coroutine commit_transaction(self, transactional_id: str) → None[source]
Return type

None

coroutine commit_transactions(self, tid_to_offset_map: Mapping[str, Mapping[faust.types.tuples.TP, int]], group_id: str, start_new_transaction: bool = True) → None[source]
Return type

None

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 flush(self) → None[source]
Return type

None

coroutine maybe_begin_transaction(self, transactional_id: str) → None[source]
Return type

None

coroutine on_start(self) → None[source]

Service is starting.

Return type

None

coroutine on_stop(self) → None[source]

Service is being stopped/restarted.

Return type

None

coroutine send(self, topic: str, key: Optional[bytes], value: Optional[bytes], partition: Optional[int], timestamp: Optional[float], headers: Union[List[Tuple[str, bytes]], Mapping[str, bytes], None], *, transactional_id: str = None) → 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], timestamp: Optional[float], headers: Union[List[Tuple[str, bytes]], Mapping[str, bytes], None], *, transactional_id: str = None) → faust.types.tuples.RecordMetadata[source]
Return type

RecordMetadata

coroutine stop_transaction(self, transactional_id: str) → None[source]
Return type

None

class faust.transport.drivers.aiokafka.Transport(*args, **kwargs) → None[source]

Kafka transport using aiokafka.

class Consumer(*args, **kwargs) → None

Kafka consumer using aiokafka.

RebalanceListener

alias of ConsumerRebalanceListener

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

logger = <Logger faust.transport.drivers.aiokafka (WARNING)>
coroutine on_stop(self) → None

Service is being stopped/restarted.

Return type

None

class Producer(transport: faust.types.transports.TransportT, loop: asyncio.events.AbstractEventLoop = None, **kwargs) → None

Kafka producer using aiokafka.

coroutine abort_transaction(self, transactional_id: str) → None
Return type

None

allow_headers = True
coroutine begin_transaction(self, transactional_id: str) → None
Return type

None

coroutine commit_transaction(self, transactional_id: str) → None
Return type

None

coroutine commit_transactions(self, tid_to_offset_map: Mapping[str, Mapping[faust.types.tuples.TP, int]], group_id: str, start_new_transaction: bool = True) → None
Return type

None

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

key_partition(topic: str, key: bytes) → faust.types.tuples.TP
Return type

TP

logger = <Logger faust.transport.drivers.aiokafka (WARNING)>
coroutine maybe_begin_transaction(self, transactional_id: str) → None
Return type

None

coroutine on_start(self) → None

Service is starting.

Return type

None

coroutine on_stop(self) → None

Service is being stopped/restarted.

Return type

None

coroutine send(self, topic: str, key: Optional[bytes], value: Optional[bytes], partition: Optional[int], timestamp: Optional[float], headers: Union[List[Tuple[str, bytes]], Mapping[str, bytes], None], *, transactional_id: str = None) → 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], timestamp: Optional[float], headers: Union[List[Tuple[str, bytes]], Mapping[str, bytes], None], *, transactional_id: str = None) → faust.types.tuples.RecordMetadata
Return type

RecordMetadata

coroutine stop_transaction(self, transactional_id: str) → None
Return type

None

supports_headers() → bool
Return type

bool

default_port = 9092
driver_version = 'aiokafka=1.0.3'