### Setup Kafka and Produce Decorator Example Source: https://github.com/alm0ra/mockafka-py/blob/main/README.md Demonstrates setting up Kafka topics using @setup_kafka and then producing messages to a specific topic with @produce. ```python from mockafka import setup_kafka, produce @setup_kafka(topics=[{"topic": "test_topic", "partition": 16}]) @produce(topic='test_topic', partition=5, key='test_', value='test_value1') def test_produce_with_kafka_setup_decorator(): # Your test logic here pass ``` -------------------------------- ### Async Producer Example Source: https://github.com/alm0ra/mockafka-py/blob/main/_autodocs/errors.md Shows how to start an asynchronous FakeAIOKafkaProducer, send a message to a topic, and then stop the producer. ```python from mockafka.aiokafka import FakeAIOKafkaProducer producer = FakeAIOKafkaProducer() try: await producer.start() await producer.send_and_wait( topic='events', value=b'event_data', partition=0 ) except ValueError as e: print(f"Invalid message type: {e}") finally: await producer.stop() ``` -------------------------------- ### Setup Kafka with Topics and Clean Option Source: https://github.com/alm0ra/mockafka-py/blob/main/docs/decorators.md Use the @setup_kafka decorator to initialize Mockafka with specified topics and partitions. Set clean=True to start with an empty Kafka environment. ```python from mockafka import setup_kafka @setup_kafka(topics=[{'topic': 'test_topic', 'partition': 5}], clean=True) def test_function(): # Your test logic here pass ``` -------------------------------- ### Async Consumer Example Source: https://github.com/alm0ra/mockafka-py/blob/main/_autodocs/errors.md Illustrates starting an asynchronous FakeAIOKafkaConsumer, subscribing to topics, getting a message, and committing it. ```python from aiokafka.errors import ConsumerStoppedError from mockafka.aiokafka import FakeAIOKafkaConsumer consumer = FakeAIOKafkaConsumer() try: await consumer.start() consumer.subscribe(topics=['test']) message = await consumer.getone() if message: await consumer.commit() except ConsumerStoppedError: print("Consumer not started") except Exception as e: print(f"Error: {e}") finally: await consumer.stop() ``` -------------------------------- ### Install Mockafka-py Source: https://github.com/alm0ra/mockafka-py/blob/main/README.md Install the mockafka-py library using pip or poetry. ```bash pip install mockafka-py ``` ```bash poetry add mockafka-py ``` -------------------------------- ### Full Decorator Chain Example Source: https://github.com/alm0ra/mockafka-py/blob/main/README.md An example showcasing a comprehensive use of decorators including @setup_kafka, multiple @produce, and @consume for complex testing scenarios. ```python from mockafka import setup_kafka, produce, consume @setup_kafka(topics=[{"topic": "test_topic", "partition": 16}]) @produce(topic='test_topic', partition=5, key='test_', value='test_value1') @produce(topic='test_topic', partition=5, key='test_', value='test_value1') @consume(topics=['test_topic']) def test_consumer_decorator(message: Message = None): if message is None: return # Your test logic for processing the consumed message here pass ``` -------------------------------- ### Test Consumer Decorator with Setup, Produce, and Consume Source: https://github.com/alm0ra/mockafka-py/blob/main/docs/decorators.md This example demonstrates using `@setup_kafka`, multiple `@produce`, and `@consume` decorators to set up a Kafka topic, produce messages, and define a consumer function to process them. The `@setup_kafka` decorator initializes the topic, `@produce` sends messages, and `@consume` defines the function that will receive these messages. ```python from mockafka import setup_kafka, produce, consume, Message @setup_kafka(topics=[{"topic": "test_topic", "partition": 16}]) @produce(topic='test_topic', partition=5, key='test_', value='test_value1') @produce(topic='test_topic', partition=5, key='test_', value='test_value1') @consume(topics=['test_topic']) def test_consumer_decorator(message: Message = None): if message is None: return # Your test logic for processing the consumed message here pass ``` -------------------------------- ### Install Mockafka Python Source: https://github.com/alm0ra/mockafka-py/blob/main/docs/index.md Install the mockafka-py library using pip. ```bash pip install mockafka-py ``` -------------------------------- ### start() Source: https://github.com/alm0ra/mockafka-py/blob/main/_autodocs/api-reference/fake-aiokafka-admin.md Starts the admin client. In this mock implementation, this method is a no-operation and serves primarily to fulfill the interface contract. ```APIDOC ## start() ### Description Starts the admin client. In this mock implementation, this method is a no-operation and serves primarily to fulfill the interface contract. ### Signature ```python async def start(self) -> None: ``` ### Example ```python admin = FakeAIOKafkaAdmin() await admin.start() ``` ``` -------------------------------- ### start() Source: https://github.com/alm0ra/mockafka-py/blob/main/_autodocs/api-reference/fake-aiokafka-consumer.md Starts the consumer and resets its internal state, preparing it for message consumption. ```APIDOC ## start() ### Description Starts the consumer and resets internal state. ### Method async def start(self) -> None: ### Request Example ```python consumer = FakeAIOKafkaConsumer() await consumer.start() ``` ``` -------------------------------- ### Initialize and Use FakeAIOKafkaConsumer Source: https://github.com/alm0ra/mockafka-py/blob/main/docs/async-fake-aiokafka-consumer.md Demonstrates how to create an instance of FakeAIOKafkaConsumer, subscribe to topics, start the consumer, and retrieve a message. Ensure you have the mockafka library installed. ```python from mockafka.aiokafka import FakeAIOKafkaConsumer # Create an instance of FakeAIOKafkaConsumer fake_consumer = FakeAIOKafkaConsumer() # Subscribe to topics fake_consumer.subscribe(topics=['sample_topic1', 'sample_topic2']) # start consumer await fake_consumer.start() # Get one message message = await fake_consumer.getone() ``` -------------------------------- ### start() Source: https://github.com/alm0ra/mockafka-py/blob/main/_autodocs/api-reference/fake-aiokafka-producer.md Starts the producer. In the mock implementation, this method is a no-operation (no-op). ```APIDOC ## start() ### Description Starts the producer. Mock implementation is a no-op. ### Signature ```python async def start(self) -> None: ``` ### Example ```python producer = FakeAIOKafkaProducer() await producer.start() ``` ``` -------------------------------- ### KafkaStore Singleton Instance Example Source: https://github.com/alm0ra/mockafka-py/blob/main/_autodocs/api-reference/kafka-store.md Demonstrates how to get a shared instance of KafkaStore. Calling with `clean=True` provides a fresh instance that resets the global state. ```python from mockafka.kafka_store import KafkaStore # Get shared instance store = KafkaStore() # Get fresh instance (clears all data) store_clean = KafkaStore(clean=True) ``` -------------------------------- ### Start FakeAIOKafkaProducer Source: https://github.com/alm0ra/mockafka-py/blob/main/_autodocs/api-reference/fake-aiokafka-producer.md Starts the producer. The mock implementation is a no-operation (no-op). ```python producer = FakeAIOKafkaProducer() await producer.start() ``` -------------------------------- ### Example Usage of TopicConfig with @setup_kafka Source: https://github.com/alm0ra/mockafka-py/blob/main/_autodocs/types.md Illustrates how to configure topics using the TopicConfig structure for the `@setup_kafka` decorator. ```python from mockafka import setup_kafka topics = [ {'topic': 'orders', 'partition': 5}, {'topic': 'payments', 'partition': 3}, {'topic': 'events', 'partition': 10} ] @setup_kafka(topics=topics) def test_function(): pass ``` -------------------------------- ### Start FakeAIOKafkaAdmin Client Source: https://github.com/alm0ra/mockafka-py/blob/main/_autodocs/api-reference/fake-aiokafka-admin.md Starts the admin client. The mock implementation is a no-operation. ```python admin = FakeAIOKafkaAdmin() await admin.start() ``` -------------------------------- ### Initialize and Use KafkaStore Source: https://github.com/alm0ra/mockafka-py/blob/main/docs/kafka-store.md Demonstrates the basic setup and usage of KafkaStore, including creating topics, partitions, producing messages, and retrieving them. ```python from mockafka.kafka_store import KafkaStore from mockafka.message import Message # Create an instance of KafkaStore kafka_store = KafkaStore() # Create a topic and partitions kafka_store.create_topic('sample_topic') kafka_store.create_partition('sample_topic', 4) # Produce a message to a specific partition message = Message(content='Hello, Kafka!') kafka_store.produce(message, 'sample_topic', 1) # Get a message from a specific topic, partition, and offset retrieved_message = kafka_store.get_message('sample_topic', 1, 0) ``` -------------------------------- ### FakeAIOKafkaAdmin Start Source: https://github.com/alm0ra/mockafka-py/blob/main/docs/async-fake-aiokafka-admin-client.md Starts the FakeAIOKafkaAdmin client. This method prepares the client for operation. ```APIDOC #### `start(self)` - **Description:** Starts the admin client. ``` -------------------------------- ### Sync Consumer Example Source: https://github.com/alm0ra/mockafka-py/blob/main/_autodocs/errors.md Demonstrates how to subscribe to topics, poll for messages, and commit them using a synchronous FakeConsumer. ```python from mockafka import FakeConsumer consumer = FakeConsumer() consumer.subscribe(topics=['orders']) try: message = consumer.poll(timeout=100) if message: consumer.commit(message=message) except Exception as e: print(f"Consumption failed: {e}") consumer.close() ``` -------------------------------- ### Define TopicConfig for Setup Decorators Source: https://github.com/alm0ra/mockafka-py/blob/main/_autodocs/types.md Defines the dictionary structure for topic configuration when used with setup decorators like `@setup_kafka`. Both topic and partition fields are required. ```python from typing import TypedDict class TopicConfig(TypedDict): topic: str partition: int ``` -------------------------------- ### Setup Kafka and Produce Message Source: https://github.com/alm0ra/mockafka-py/blob/main/docs/decorators.md Combine @setup_kafka with @produce to first configure the Kafka environment with specific topics and then produce a message to one of those topics. ```python from mockafka import setup_kafka, produce @setup_kafka(topics=[{"topic": "test_topic", "partition": 16}]) @produce(topic='test_topic', partition=5, key='test_', value='test_value1') def test_produce_with_kafka_setup_decorator(): # Your test logic here pass ``` -------------------------------- ### Start FakeAIOKafkaConsumer Source: https://github.com/alm0ra/mockafka-py/blob/main/_autodocs/api-reference/fake-aiokafka-consumer.md Starts the consumer and resets its internal state. Ensure this is called before other operations. ```python consumer = FakeAIOKafkaConsumer() await consumer.start() ``` -------------------------------- ### Singleton Storage Example Source: https://github.com/alm0ra/mockafka-py/blob/main/_autodocs/index.md Demonstrates how multiple clients share a global KafkaStore instance for inter-client communication. Data produced by one client is accessible by another. ```python producer = FakeProducer() producer.produce(topic='test', value='data', partition=0) consumer = FakeConsumer() consumer.subscribe(topics=['test']) message = consumer.poll() # Gets message from producer above ``` -------------------------------- ### Kafka Decorator Example for Testing Source: https://github.com/alm0ra/mockafka-py/blob/main/_autodocs/index.md Shows how to use mockafka decorators to set up a Kafka environment, produce messages, and consume them within a test function. ```python from mockafka import setup_kafka, produce, consume @setup_kafka(topics=[{'topic': 'test', 'partition': 2}]) @produce(topic='test', value='msg1', partition=0) @produce(topic='test', value='msg2', partition=1) @consume(topics=['test']) def test_consume_produced_messages(message=None): if message is None: return print(f"Got message: {message.value()}") ``` -------------------------------- ### Example Usage of FakeAIOKafkaAdmin Source: https://github.com/alm0ra/mockafka-py/blob/main/docs/async-fake-aiokafka-admin-client.md Demonstrates how to instantiate FakeAIOKafkaAdmin, create new topics with specified partitions, and add more partitions to existing topics. ```python from aiokafka import NewTopic, NewPartitions from mockafka.aiokafka import FakeAIOKafkaAdmin # Create an instance of FakeAIOKafkaAdmin fake_admin = FakeAIOKafkaAdmin() # Define new topics new_topics = [NewTopic(name='topic1', num_partitions=3), NewTopic(name='topic2', num_partitions=5)] # Create topics await fake_admin.create_topics(new_topics=new_topics) # Define additional partitions for topics topic_partitions = {'topic1': NewPartitions(total_count=5), 'topic2': NewPartitions(total_count=7)} # Add partitions to topics await fake_admin.create_partitions(topic_partitions=topic_partitions) ``` -------------------------------- ### Initialize FakeAIOKafkaProducer Source: https://github.com/alm0ra/mockafka-py/blob/main/_autodocs/api-reference/fake-aiokafka-producer.md Initializes the FakeAIOKafkaProducer with in-memory message storage. This is the basic setup for using the mock producer. ```python from mockafka.aiokafka import FakeAIOKafkaProducer producer = FakeAIOKafkaProducer() ``` -------------------------------- ### Async Kafka Topic Setup with @asetup_kafka Source: https://github.com/alm0ra/mockafka-py/blob/main/docs/async-decorators.md Utilize the @asetup_kafka decorator to set up mock Kafka topics using FakeAIOKafkaAdminClient. It creates specified topics, with an option to clean existing ones first. ```python from mockafka import asetup_kafka @asetup_kafka(topics=[{'topic': 'test', 'partition': 1}]) async def test_function(): # test logic ``` -------------------------------- ### Async Test Setup with Decorator Source: https://github.com/alm0ra/mockafka-py/blob/main/_autodocs/index.md Demonstrates setting up an asynchronous test environment using pytest-asyncio and the @asetup_kafka decorator. This allows for testing async Kafka operations. ```python import pytest @pytest.mark.asyncio @asetup_kafka(topics=[{'topic': 'test', 'partition': 1}]) async def test_async(): # test code ``` -------------------------------- ### Initialize and Use KafkaStore Source: https://github.com/alm0ra/mockafka-py/blob/main/_autodocs/configuration.md Instantiate KafkaStore, which is a singleton. Use `clean=True` to reset the shared global state for all instances. Examples show creating topics and listing them. ```python from mockafka.kafka_store import KafkaStore # Get shared instance store1 = KafkaStore() store1.create_topic('test') # Get same instance store2 = KafkaStore() print(store2.topic_list()) # Output: ['test'] # Get fresh instance (clears all) store_clean = KafkaStore(clean=True) print(store_clean.topic_list()) # Output: [] ``` -------------------------------- ### Configure Kafka Topics with @setup_kafka Decorator Source: https://github.com/alm0ra/mockafka-py/blob/main/_autodocs/configuration.md Use the @setup_kafka decorator to configure topics for a test function. Set `clean=True` to clear existing topics before setup. ```python from mockafka import setup_kafka @setup_kafka( topics=[ {'topic': 'orders', 'partition': 5}, {'topic': 'payments', 'partition': 3} ], clean=False # Preserve existing topics ) def test_function(): pass ``` -------------------------------- ### Example Usage of MessageDict with @bulk_produce Source: https://github.com/alm0ra/mockafka-py/blob/main/_autodocs/types.md Demonstrates how to structure a list of messages using the MessageDict format for the `@bulk_produce` decorator. ```python from mockafka import bulk_produce messages = [ { 'topic': 'orders', 'value': 'order-123', 'key': 'customer-456', 'partition': 0 }, { 'topic': 'orders', 'value': 'order-124', 'key': 'customer-789', 'partition': 1, 'timestamp': 1234567890, 'headers': {'priority': b'high'} } ] @bulk_produce(list_of_messages=messages) def test_function(): pass ``` -------------------------------- ### Asynchronous Kafka Produce and Consume Example Source: https://github.com/alm0ra/mockafka-py/blob/main/_autodocs/index.md Illustrates asynchronous message production and consumption using FakeAIOKafkaProducer and FakeAIOKafkaConsumer with aiokafka. ```python import asyncio from mockafka.aiokafka import FakeAIOKafkaProducer, FakeAIOKafkaConsumer async def main(): # Produce producer = FakeAIOKafkaProducer() await producer.start() await producer.send(topic='events', value=b'data', partition=0) await producer.stop() # Consume consumer = FakeAIOKafkaConsumer() await consumer.start() consumer.subscribe(topics=['events']) record = await consumer.getone() print(f"Value: {record.value}") # b'data' await consumer.stop() asyncio.run(main()) ``` -------------------------------- ### Setup Kafka Topics with @asetup_kafka Decorator Source: https://github.com/alm0ra/mockafka-py/blob/main/_autodocs/configuration.md Use this decorator to configure Kafka topics before running a test. Set `clean=True` to clear existing topics beforehand. ```python from mockafka import asetup_kafka import pytest @pytest.mark.asyncio @asetup_kafka( topics=[{'topic': 'test', 'partition': 1}], clean=True # Clear before test ) async def test_function(): pass ``` -------------------------------- ### Async Decorator Stacking Example Source: https://github.com/alm0ra/mockafka-py/blob/main/_autodocs/api-reference/async-decorators.md Demonstrates stacking multiple asynchronous decorators (@aconsume, @aproduce, @asetup_kafka) to create a test pipeline. Decorators are applied in bottom-to-top order. ```python import pytest from mockafka import asetup_kafka, aproduce, aconsume @pytest.mark.asyncio @aconsume(topics=['output']) # Step 4: Consume output @aproduce(topic='output', value=b'result', partition=0) # Step 3: Produce result @aproduce(topic='input', value=b'data', partition=0) # Step 2: Produce input @asetup_kafka(topics=[ {'topic': 'input', 'partition': 1}, {'topic': 'output', 'partition': 1} ]) async def test_async_pipeline(message=None): if message is not None: assert message.value == b'result' ``` -------------------------------- ### FakeConsumer Basic Usage Example Source: https://github.com/alm0ra/mockafka-py/blob/main/docs/fake-consumer.md Demonstrates the basic lifecycle of using FakeConsumer, including subscription, consumption, committing offsets, unsubscription, and closing the consumer. This is useful for testing consumer logic without a live Kafka broker. ```python from mockafka import FakeConsumer # Create an instance of FakeConsumer fake_consumer = FakeConsumer() # Subscribe to topics fake_consumer.subscribe(topics=['sample_topic']) # Consume messages consumed_message = fake_consumer.consume() # Commit offsets fake_consumer.commit() # Unsubscribe from topics fake_consumer.unsubscribe(topics=['sample_topic']) # Close the consumer fake_consumer.close() ``` -------------------------------- ### Synchronous Kafka Produce and Consume Example Source: https://github.com/alm0ra/mockafka-py/blob/main/_autodocs/index.md Demonstrates basic usage of FakeProducer to send a message and FakeConsumer to receive and commit it using the confluent-kafka API. ```python from mockafka import FakeProducer, FakeConsumer # Create topic and producer producer = FakeProducer() producer.produce(topic='orders', value='data', key='key1', partition=0) # Consume consumer = FakeConsumer() consumer.subscribe(topics=['orders']) message = consumer.poll() print(f"Value: {message.value()}") # b'data' print(f"Key: {message.key()}") # b'key1' consumer.commit(message=message) ``` -------------------------------- ### Example Usage of TopicPartition with getmany Source: https://github.com/alm0ra/mockafka-py/blob/main/_autodocs/types.md Shows how to use `TopicPartition` objects to fetch messages from specific partitions using `FakeAIOKafkaConsumer.getmany()`. ```python from aiokafka.structs import TopicPartition from mockafka.aiokafka import FakeAIOKafkaConsumer consumer = FakeAIOKafkaConsumer() await consumer.start() # Fetch from specific partitions records = await consumer.getmany( TopicPartition(topic='orders', partition=0), TopicPartition(topic='orders', partition=1) ) for tp, messages in records.items(): print(f"{tp.topic}[{tp.partition}]: {len(messages)} messages") ``` -------------------------------- ### Example Usage of FakeAIOKafkaConsumer Source: https://github.com/alm0ra/mockafka-py/blob/main/_autodocs/types.md Demonstrates fetching a single record using `FakeAIOKafkaConsumer.getone()` and accessing its attributes. ```python from mockafka.aiokafka import FakeAIOKafkaConsumer consumer = FakeAIOKafkaConsumer() await consumer.start() consumer.subscribe(topics=['test']) record = await consumer.getone() if record: print(f"Topic: {record.topic}") print(f"Partition: {record.partition}") print(f"Offset: {record.offset}") print(f"Value: {record.value}") print(f"Key: {record.key}") ``` -------------------------------- ### Bulk Produce and Consume Decorator Example Source: https://github.com/alm0ra/mockafka-py/blob/main/README.md Illustrates using @bulk_produce along with @consume decorators to send multiple messages at once and then process them. ```python from mockafka import bulk_produce, consume @bulk_produce(list_of_messages=sample_for_bulk_produce) @consume(topics=['test']) def test_bulk_produce_and_consume_decorator(message): """ This test showcases the usage of both @bulk_produce and @consume decorators in a single test case. It does bulk produces messages to the 'test' topic and then consumes them to perform further logic. """ # Your test logic for processing the consumed message here pass ``` -------------------------------- ### Produce and Consume Decorator Example Source: https://github.com/alm0ra/mockafka-py/blob/main/README.md Demonstrates using both @produce and @consume decorators in a single test case to simulate message production and consumption. ```python from mockafka import produce, consume @produce(topic='test', key='test_key', value='test_value', partition=4) @consume(topics=['test']) def test_produce_and_consume_decorator(message): """ This test showcases the usage of both @produce and @consume decorators in a single test case. It produces a message to the 'test' topic and then consumes it to perform further logic. # Notice you may get message None """ # Your test logic for processing the consumed message here if not message: return pass ``` -------------------------------- ### Get First Offset of a Partition Source: https://github.com/alm0ra/mockafka-py/blob/main/_autodocs/api-reference/kafka-store.md Retrieves the starting offset value for a specific partition within a topic. ```python def get_partition_first_offset(self, topic: str, partition: int) -> int: ``` -------------------------------- ### Accessing Partition Metadata Details Source: https://github.com/alm0ra/mockafka-py/blob/main/_autodocs/api-reference/metadata.md Demonstrates how to create a PartitionMetadata object and access its attributes such as ID, leader, replicas, and in-sync replicas. It also shows how to get the number of replicas using len(). ```python from mockafka.partition_metadata import PartitionMetadata partition = PartitionMetadata(id=0) print(f"Partition ID: {partition.id}") print(f"Leader: {partition.leader}") print(f"Replicas: {partition.replicas}") print(f"In-sync replicas: {partition.isrs}") print(f"Number of replicas: {len(partition)}") ``` -------------------------------- ### Get All Topics Source: https://github.com/alm0ra/mockafka-py/blob/main/_autodocs/api-reference/fake-aiokafka-consumer.md Retrieves all topic names available in the mock Kafka cluster. Ensure the consumer is started before calling this method. ```python consumer = FakeAIOKafkaConsumer() await consumer.start() all_topics = await consumer.topics() # Returns: set of all available topic names ``` -------------------------------- ### Create Topics with FakeAIOKafkaAdmin Source: https://github.com/alm0ra/mockafka-py/blob/main/_autodocs/api-reference/fake-aiokafka-admin.md Creates multiple topics using a list of NewTopic objects. Ensure the admin client is started before calling this method. ```python from aiokafka.admin import NewTopic from mockafka.aiokafka import FakeAIOKafkaAdmin async def test_create_topics(): admin = FakeAIOKafkaAdmin() await admin.start() await admin.create_topics([ NewTopic( name='orders', num_partitions=3, replication_factor=1 ), NewTopic( name='events', num_partitions=5, replication_factor=1 ), ]) await admin.close() ``` -------------------------------- ### Async Kafka Setup Decorator Source: https://github.com/alm0ra/mockafka-py/blob/main/_autodocs/api-reference/async-decorators.md The @asetup_kafka decorator creates Kafka topics before executing an asynchronous function. It accepts a list of topic configurations and an option to clean existing topics. ```python from mockafka import asetup_kafka, aproduce import pytest @pytest.mark.asyncio @asetup_kafka( topics=[ {'topic': 'orders', 'partition': 3}, {'topic': 'payments', 'partition': 2}, ], clean=True ) @aproduce(topic='orders', value=b'data', partition=0) async def test_with_custom_topics(): # Topics 'orders' and 'payments' now exist with specified partitions from mockafka.aiokafka import FakeAIOKafkaConsumer consumer = FakeAIOKafkaConsumer() await consumer.start() consumer.subscribe(topics=['orders']) message = await consumer.getone() assert message.value == b'data' await consumer.stop() ``` -------------------------------- ### Get Number of Partitions for a Topic Source: https://github.com/alm0ra/mockafka-py/blob/main/_autodocs/api-reference/kafka-store.md A static method that returns the total number of partitions configured for a given topic. ```python @staticmethod def get_number_of_partition(topic: str) -> int: ``` -------------------------------- ### Get Multiple Messages Source: https://github.com/alm0ra/mockafka-py/blob/main/_autodocs/api-reference/fake-aiokafka-consumer.md Retrieves multiple messages at once from specified partitions, with options for maximum records. Returns a dictionary mapping TopicPartition to a list of ConsumerRecords. Raises ConsumerStoppedError if the consumer is not started. ```python consumer = FakeAIOKafkaConsumer() await consumer.start() consumer.subscribe(topics=['test']) records = await consumer.getmany(max_records=10) for tp, messages in records.items(): print(f"Topic: {tp.topic}, Partition: {tp.partition}, Messages: {len(messages)}") ``` -------------------------------- ### Decorator Stacking Example Source: https://github.com/alm0ra/mockafka-py/blob/main/_autodocs/api-reference/decorators.md Demonstrates the bottom-to-top application of decorators for a data pipeline. The innermost decorator executes first, and the outermost executes last. ```python @consume(topics=['output']) # Step 4: Consume messages @produce(topic='output', value='result', partition=0) # Step 3: Produce result @produce(topic='input', value='data', partition=0) # Step 2: Produce input @setup_kafka(topics=[{'topic': 'input', 'partition': 1}, {'topic': 'output', 'partition': 1}]) # Step 1: Setup def test_pipeline(message=None): if message is not None: assert message.value() == b'result' ``` -------------------------------- ### Get Single Message Source: https://github.com/alm0ra/mockafka-py/blob/main/_autodocs/api-reference/fake-aiokafka-consumer.md Retrieves the next available message from subscribed topics or specific partitions. Returns None if no messages are available or the consumer is exhausted. Raises ConsumerStoppedError if the consumer is not started. ```python from mockafka.aiokafka import FakeAIOKafkaConsumer consumer = FakeAIOKafkaConsumer() await consumer.start() consumer.subscribe(topics=['test']) message = await consumer.getone() if message: print(f"Key: {message.key}, Value: {message.value}") ``` -------------------------------- ### Async Context Manager for Consumer Lifecycle Source: https://github.com/alm0ra/mockafka-py/blob/main/_autodocs/api-reference/fake-aiokafka-consumer.md Utilize the consumer as an asynchronous context manager to automatically handle start and stop operations. Subscribe to topics within the context. ```python async with FakeAIOKafkaConsumer() as consumer: consumer.subscribe(topics=['test']) record = await consumer.getone() ``` -------------------------------- ### Multiple Produce Decorators Example Source: https://github.com/alm0ra/mockafka-py/blob/main/README.md Showcases the use of multiple @produce decorators within a single test function to send several messages. ```python from mockafka import produce @produce(topic='test', key='test_key', value='test_value', partition=4) @produce(topic='test', key='test_key1', value='test_value1', partition=0) def test_produce_twice(): # Your test logic here pass ``` -------------------------------- ### Instantiate and Print BrokerMetadata Source: https://github.com/alm0ra/mockafka-py/blob/main/_autodocs/api-reference/metadata.md Demonstrates how to create a BrokerMetadata instance and access its attributes. The output shows the default mock values for broker ID, host, and port. ```python from mockafka.broker_metadata import BrokerMetadata broker = BrokerMetadata() print(f"Broker ID: {broker.id}") print(f"Host: {broker.host}") print(f"Port: {broker.port}") print(f"Broker string: {broker}") # Output: "fakebroker:9091/1" ``` -------------------------------- ### Use FakeAIOKafkaProducer as Async Context Manager Source: https://github.com/alm0ra/mockafka-py/blob/main/_autodocs/api-reference/fake-aiokafka-producer.md Utilizes the FakeAIOKafkaProducer within an async context manager, automatically handling the start and stop operations. This simplifies producer lifecycle management. ```python async with FakeAIOKafkaProducer() as producer: await producer.send(topic='test', value=b'data', partition=0) ``` -------------------------------- ### Add Partitions to Topics with FakeAIOKafkaAdmin Source: https://github.com/alm0ra/mockafka-py/blob/main/_autodocs/api-reference/fake-aiokafka-admin.md Adds partitions to existing topics by specifying the topic name and the desired total partition count using NewPartitions. The admin client must be started prior to this operation. ```python from aiokafka.admin import NewPartitions from mockafka.aiokafka import FakeAIOKafkaAdmin async def test_add_partitions(): admin = FakeAIOKafkaAdmin() await admin.start() # First create topic with 1 partition from aiokafka.admin import NewTopic await admin.create_topics([ NewTopic(name='metrics', num_partitions=1, replication_factor=1) ]) # Later, expand to 5 partitions await admin.create_partitions({ 'metrics': NewPartitions(total_count=5) }) await admin.close() ``` -------------------------------- ### Accessing Topic Metadata and Partition Count Source: https://github.com/alm0ra/mockafka-py/blob/main/_autodocs/api-reference/metadata.md Shows how to instantiate ClusterMetadata for a specific topic and then access the TopicMetadata object to get the topic name and the number of partitions. It also iterates through partition details. ```python from mockafka.cluster_metadata import ClusterMetadata metadata = ClusterMetadata(topic='orders') topic_meta = metadata.topics['orders'] print(f"Topic: {topic_meta.topic}") print(f"Number of partitions: {len(topic_meta)}") for partition_id, partition_meta in topic_meta.partitions.items(): print(f" Partition {partition_id}: Leader {partition_meta.leader}") ``` -------------------------------- ### Initialize FakeAdminClientImpl Source: https://github.com/alm0ra/mockafka-py/blob/main/_autodocs/configuration.md Instantiate FakeAdminClientImpl. Set `clean=True` to clear all existing topics and offsets on initialization. ```python from mockafka import FakeAdminClientImpl # Create admin with existing data preserved admin = FakeAdminClientImpl() # Create admin and clear all data admin_fresh = FakeAdminClientImpl(clean=True) ``` -------------------------------- ### get_partition_first_offset Source: https://github.com/alm0ra/mockafka-py/blob/main/_autodocs/api-reference/kafka-store.md Retrieves the first offset (starting point) for a given topic partition. ```APIDOC ## get_partition_first_offset(topic: str, partition: int) ### Description Retrieves the first offset (starting point) for a given topic partition. ### Parameters #### Path Parameters None #### Query Parameters None #### Request Body None ### Parameters - **topic** (`str`) - Required - Description: Topic name - **partition** (`int`) - Required - Description: Partition number ### Returns - **int** - The first offset of the partition. ``` -------------------------------- ### @setup_kafka Source: https://github.com/alm0ra/mockafka-py/blob/main/_autodocs/api-reference/decorators.md Decorator that creates Kafka topics before executing the decorated function. ```APIDOC ## @setup_kafka ### Description Decorator that creates Kafka topics before executing the decorated function. ### Parameters #### Path Parameters None #### Query Parameters None #### Request Body None ### Parameters - **topics** (`list[TopicConfig]`) - Required - List of topic configurations - **clean** (`bool`) - Optional - If True, clears existing topics/offsets first Each topic in `topics` is a dict with: - `topic` (str): Topic name - `partition` (int): Number of partitions ### Request Example ```python from mockafka import setup_kafka, produce @setup_kafka( topics=[ {'topic': 'orders', 'partition': 3}, {'topic': 'payments', 'partition': 2}, ], clean=True ) def decorated_function(): ... ``` ### Response #### Success Response (200) None #### Response Example None ``` -------------------------------- ### Get Message by Offset Source: https://github.com/alm0ra/mockafka-py/blob/main/docs/kafka-store.md Retrieves a specific message from a topic, partition, and offset. ```APIDOC ## get_message(self, topic: str, partition: int, offset: int) -> Message ### Description Gets a message from a specific topic, partition, and offset. ### Parameters - `topic` (str): The name of the topic. - `partition` (int): The partition number. - `offset` (int): The offset of the message. ### Returns (Message) The requested message. ``` -------------------------------- ### Get Number of Partitions Source: https://github.com/alm0ra/mockafka-py/blob/main/docs/kafka-store.md Retrieves the total number of partitions for a specified topic. ```APIDOC ## get_number_of_partition(topic: str) -> int ### Description Gets the number of partitions in a topic. ### Parameters - `topic` (str): The name of the topic. ### Returns (int) The number of partitions in the topic. ``` -------------------------------- ### Get Partition List Source: https://github.com/alm0ra/mockafka-py/blob/main/_autodocs/api-reference/kafka-store.md Retrieves a list of all partition IDs for a given topic. ```python @staticmethod def partition_list(topic: str) -> list[int]: ``` -------------------------------- ### Initialize FakeAIOKafkaProducer Source: https://github.com/alm0ra/mockafka-py/blob/main/_autodocs/configuration.md Instantiate FakeAIOKafkaProducer. Additional arguments are accepted for API compatibility but are ignored. ```python from mockafka.aiokafka import FakeAIOKafkaProducer producer = FakeAIOKafkaProducer() # Arguments are ignored (for aiokafka compatibility) producer = FakeAIOKafkaProducer({'bootstrap.servers': 'localhost:9092'}) ``` -------------------------------- ### Get Messages in Partition Source: https://github.com/alm0ra/mockafka-py/blob/main/docs/kafka-store.md Retrieves all messages stored within a specific partition of a topic. ```APIDOC ## get_messages_in_partition(topic: str, partition: int) -> list[Message] ### Description Gets all messages in a specific partition. ### Parameters - `topic` (str): The name of the topic. - `partition` (int): The partition number. ### Returns (list[Message]) List of messages in the partition. ``` -------------------------------- ### Get Partition Next Offset Source: https://github.com/alm0ra/mockafka-py/blob/main/docs/kafka-store.md Retrieves the next available offset for a given partition. ```APIDOC ## get_partition_next_offset(self, topic: str, partition: int) -> int ### Description Gets the next offset for a partition. ### Parameters - `topic` (str): The name of the topic. - `partition` (int): The partition number. ### Returns (int) The next offset for the partition. ``` -------------------------------- ### Get Partition First Offset Source: https://github.com/alm0ra/mockafka-py/blob/main/docs/kafka-store.md Retrieves the first offset value for a given partition. ```APIDOC ## get_partition_first_offset(self, topic: str, partition: int) -> int ### Description Gets the first offset for a partition. ### Parameters - `topic` (str): The name of the topic. - `partition` (int): The partition number. ### Returns (int) The first offset for the partition. ``` -------------------------------- ### Length of Kafka Store (Not Implemented) Source: https://github.com/alm0ra/mockafka-py/blob/main/docs/fake-admin-client.md Gets the length of the Kafka store. This functionality is not implemented. ```APIDOC ## __len__(self, *args, **kwargs) ### Description Gets the length of the Kafka store (not implemented). ### Parameters - `args`: Unused parameters (not implemented). - `kwargs`: Unused parameters (not implemented). ``` -------------------------------- ### Get Topic List Source: https://github.com/alm0ra/mockafka-py/blob/main/_autodocs/api-reference/kafka-store.md Retrieves a list of all available topic names from the Kafka store. ```python @staticmethod def topic_list() -> list[str]: ``` -------------------------------- ### Mockafka Producer, Consumer, and Admin Client Usage Source: https://github.com/alm0ra/mockafka-py/blob/main/docs/index.md Demonstrates creating a topic, producing messages with a fake producer, and consuming messages with a fake consumer. Ensure you have the necessary imports. ```python from mockafka import FakeProducer, FakeConsumer, FakeAdminClientImpl from mockafka.admin_client import NewTopic from random import randint # Create topic admin = FakeAdminClientImpl() admin.create_topics([ NewTopic(topic='test', num_partitions=5) ]) # Produce messages producer = FakeProducer() for i in range(0, 10): producer.produce( topic='test', key=f'test_key{i}', value=f'test_value{i}', partition=randint(0, 4) ) # Subscribe consumer consumer = FakeConsumer() consumer.subscribe(topics=['test']) # Consume messages while True: message = consumer.poll() print(message) consumer.commit() if message is None: break ``` ```text """ None """ ``` -------------------------------- ### Get Current Subscription Source: https://github.com/alm0ra/mockafka-py/blob/main/_autodocs/api-reference/fake-aiokafka-consumer.md Retrieves a frozenset of the topic names the consumer is currently subscribed to. ```python consumer = FakeAIOKafkaConsumer() await consumer.start() consumer.subscribe(topics=['test1', 'test2']) subs = consumer.subscription() # Returns: frozenset({'test1', 'test2'}) ``` -------------------------------- ### Get Offset Store Key Source: https://github.com/alm0ra/mockafka-py/blob/main/docs/kafka-store.md Generates a unique key used for storing offset information. ```APIDOC ## get_offset_store_key(self, topic: str, partition: int) -> str ### Description Generates the key for offset storage. ### Parameters - `topic` (str): The name of the topic. - `partition` (int): The partition number. ### Returns (str) Offset store key. ``` -------------------------------- ### Get Leader Epoch Source: https://github.com/alm0ra/mockafka-py/blob/main/_autodocs/api-reference/message.md Retrieves the leader epoch for the message. Returns an integer or None if not set. ```python message = Message(leader_epoch=3) epoch = message.leader_epoch() # Returns: 3 ``` -------------------------------- ### Create Kafka topics before function execution Source: https://github.com/alm0ra/mockafka-py/blob/main/_autodocs/api-reference/decorators.md Utilize the @setup_kafka decorator to create specified Kafka topics and their partitions before executing the decorated function. Optionally, set 'clean=True' to clear existing topics and offsets first. ```python from mockafka import setup_kafka, produce @setup_kafka( topics=[ {'topic': 'orders', 'partition': 3}, {'topic': 'payments', 'partition': 2}, ], clean=True ) @produce(topic='orders', value='data', partition=0) def test_with_custom_topics(): # Topics 'orders' and 'payments' now exist pass ``` -------------------------------- ### Get Message Latency Source: https://github.com/alm0ra/mockafka-py/blob/main/_autodocs/api-reference/message.md Retrieves the latency of the message in milliseconds. Returns a float or None if not set. ```python message = Message(latency=15.5) latency = message.latency() # Returns: 15.5 ``` -------------------------------- ### Using FakeProducer to List Topics and Access Metadata Source: https://github.com/alm0ra/mockafka-py/blob/main/_autodocs/api-reference/metadata.md Demonstrates how to use FakeProducer to produce messages and then retrieve cluster and topic metadata. Shows how to access cluster ID, topic names, partition counts, and broker information. ```python from mockafka import FakeProducer producer = FakeProducer() producer.produce(topic='orders', value='data', partition=0) # Get metadata for all topics metadata = producer.list_topics() print(f"Cluster: {metadata.cluster_id}") print(f"Topics: {list(metadata.topics.keys())}") # Get metadata for specific topic topic_metadata = producer.list_topics(topic='orders') for name, tm in topic_metadata.topics.items(): print(f"Topic: {name}, Partitions: {len(tm.partitions)}") # Access broker info for broker_id, broker in metadata.brokers.items(): print(f"Broker {broker_id}: {broker.host}:{broker.port}") ``` -------------------------------- ### Get Message Key Source: https://github.com/alm0ra/mockafka-py/blob/main/_autodocs/api-reference/message.md Retrieves the key of the message. The key is returned as bytes or None if not set. ```python message = Message(key=b"order-123") key = message.key() # Returns: b"order-123" ``` -------------------------------- ### Async Iteration for Consuming Records Source: https://github.com/alm0ra/mockafka-py/blob/main/_autodocs/api-reference/fake-aiokafka-consumer.md Iterate asynchronously over consumed records. Subscribe to topics before starting the iteration. ```python consumer = FakeAIOKafkaConsumer() await consumer.start() consumer.subscribe(topics=['test']) async for record in consumer: print(f"Consumed: {record.value}") ``` -------------------------------- ### Constructor Source: https://github.com/alm0ra/mockafka-py/blob/main/_autodocs/api-reference/fake-admin-client.md Initializes a new instance of FakeAdminClientImpl. The 'clean' flag can be used to clear existing topics and offsets on initialization. ```APIDOC ## Constructor FakeAdminClientImpl ### Description Initializes a new instance of FakeAdminClientImpl. The 'clean' flag can be used to clear existing topics and offsets on initialization. ### Parameters #### Constructor Parameters - **clean** (bool) - Optional - Default: False - If True, clears all existing topics and offsets on initialization ### Example ```python from mockafka import FakeAdminClientImpl admin = FakeAdminClientImpl() # With clean flag to reset state admin_clean = FakeAdminClientImpl(clean=True) ``` ``` -------------------------------- ### Initialize FakeAdminClientImpl Source: https://github.com/alm0ra/mockafka-py/blob/main/docs/fake-admin-client.md Instantiates the FakeAdminClientImpl for testing. Use this to create a mock Kafka AdminClient. ```python from mockafka.admin_client import FakeAdminClientImpl, NewTopic, NewPartitions # Create an instance of FakeAdminClientImpl fake_admin_client = FakeAdminClientImpl() ``` -------------------------------- ### Get Message Offset Source: https://github.com/alm0ra/mockafka-py/blob/main/_autodocs/api-reference/message.md Retrieves the offset of the message within its partition. Returns an integer or None if not set. ```python message = Message(offset=42) offset = message.offset() # Returns: 42 ``` -------------------------------- ### Test with Clean State Source: https://github.com/alm0ra/mockafka-py/blob/main/_autodocs/index.md Illustrates how to ensure a fresh state for testing by initializing FakeAdminClientImpl with clean=True. This is useful for isolating test cases. ```python admin = FakeAdminClientImpl(clean=True) # Fresh state ``` -------------------------------- ### Get Message Partition Source: https://github.com/alm0ra/mockafka-py/blob/main/_autodocs/api-reference/message.md Retrieves the partition number where the message resides. Returns an integer or None if not set. ```python message = Message(partition=2) partition = message.partition() # Returns: 2 ``` -------------------------------- ### Get Message Topic Source: https://github.com/alm0ra/mockafka-py/blob/main/_autodocs/api-reference/message.md Retrieves the topic name associated with the message. Returns a string or None if not set. ```python message = Message(topic="orders") topic = message.topic() # Returns: "orders" ``` -------------------------------- ### Instantiate and Produce Message with FakeProducer Source: https://github.com/alm0ra/mockafka-py/blob/main/docs/fake-produce.md Demonstrates how to create an instance of FakeProducer and produce a message to a specified topic and partition. This is useful for testing Kafka producer logic without a live Kafka cluster. ```python from mockafka import FakeProducer # Create an instance of FakeProducer fake_producer = FakeProducer() # Produce a message fake_producer.produce(topic='sample_topic', value='Hello, Kafka!', partition=0) ``` -------------------------------- ### Create Partitions for a Topic Source: https://github.com/alm0ra/mockafka-py/blob/main/docs/fake-admin-client.md This snippet shows how to add new partitions to an existing topic. Specify the topic name and the desired new total count of partitions. ```python from mockafka.admin_client import FakeAdminClient from mockafka.topic import NewPartitions fake_admin_client = FakeAdminClient() fake_admin_client.create_partitions(partitions=[NewPartitions(topic='sample_topic', new_total_count=5)]) ``` -------------------------------- ### Get Message Value Source: https://github.com/alm0ra/mockafka-py/blob/main/_autodocs/api-reference/message.md Retrieves the value (body) of the message. The value is returned as bytes or None if not set. ```python message = Message(value=b"order confirmed") value = message.value() # Returns: b"order confirmed" ``` -------------------------------- ### FakeAdminClientImpl Initialization Source: https://github.com/alm0ra/mockafka-py/blob/main/docs/fake-admin-client.md Initializes the FakeAdminClientImpl with an optional clean slate flag. ```APIDOC ## __init__ (self, clean: bool = True, *args, **kwargs) ### Description Initializes the `FakeAdminClientImpl`. ### Parameters - `clean` (bool): Flag indicating whether to start with a clean slate. - `args`: Additional arguments (unused). - `kwargs`: Additional keyword arguments (unused). ``` -------------------------------- ### FakeConsumer Constructor Source: https://github.com/alm0ra/mockafka-py/blob/main/_autodocs/api-reference/fake-consumer.md Initializes a FakeConsumer instance with an empty internal state. This is the starting point for using the mock consumer. ```APIDOC ## FakeConsumer Constructor ### Description Initializes a consumer with empty state. ### Signature ```python FakeConsumer(*args, **kwargs) -> FakeConsumer ``` ### Example ```python from mockafka import FakeConsumer consumer = FakeConsumer() ``` ``` -------------------------------- ### Get Number of Messages in Topic Source: https://github.com/alm0ra/mockafka-py/blob/main/docs/kafka-store.md Calculates and returns the total count of messages across all partitions within a topic. ```APIDOC ## number_of_message_in_topic(self, topic: str) -> int ### Description Gets the total number of messages in a topic. ### Parameters - `topic` (str): The name of the topic. ### Returns (int) The total number of messages in the topic. ``` -------------------------------- ### Decorator Imports Source: https://github.com/alm0ra/mockafka-py/blob/main/_autodocs/README.md Import synchronous decorators such as produce, bulk_produce, setup_kafka, and consume from the mockafka package. ```python from mockafka import produce, bulk_produce, setup_kafka, consume ``` -------------------------------- ### FakeAIOKafkaProducer Constructor Source: https://github.com/alm0ra/mockafka-py/blob/main/_autodocs/api-reference/fake-aiokafka-producer.md Initializes an async producer with in-memory message storage. This is the entry point for creating a producer instance. ```APIDOC ## FakeAIOKafkaProducer Constructor ### Description Initializes an async producer with in-memory message storage. ### Signature ```python FakeAIOKafkaProducer(*args, **kwargs) -> FakeAIOKafkaProducer ``` ### Example ```python from mockafka.aiokafka import FakeAIOKafkaProducer producer = FakeAIOKafkaProducer() ``` ``` -------------------------------- ### Create a Kafka Topic Source: https://github.com/alm0ra/mockafka-py/blob/main/docs/fake-admin-client.md Use this snippet to create a new topic with a specified number of partitions. Ensure the NewTopic object is correctly instantiated. ```python from mockafka.admin_client import FakeAdminClient from mockafka.topic import NewTopic fake_admin_client = FakeAdminClient() fake_admin_client.create_topic(topic=NewTopic(topic='sample_topic', num_partitions=3)) ``` -------------------------------- ### Get Message Headers Source: https://github.com/alm0ra/mockafka-py/blob/main/_autodocs/api-reference/message.md Retrieves the headers associated with the message. Headers are returned as a list of (key, value) tuples. ```python message = Message(headers=[("correlation-id", b"123"), ("user", b"alice")]) headers = message.headers() # Returns: [("correlation-id", b"123"), ("user", b"alice")] ``` -------------------------------- ### Sync Producer Error Handling Source: https://github.com/alm0ra/mockafka-py/blob/main/_autodocs/errors.md Provides a robust error handling example for sync producers, catching both KafkaException and ValueError. ```python from confluent_kafka import KafkaException from mockafka import FakeProducer producer = FakeProducer() try: producer.produce( topic='orders', value='{"id": 123}', key='customer-1', partition=0 ) except (KafkaException, ValueError) as e: print(f"Production failed: {e}") # Handle error ``` -------------------------------- ### Initialize FakeProducer Source: https://github.com/alm0ra/mockafka-py/blob/main/_autodocs/configuration.md Instantiate FakeProducer with or without a configuration dictionary. The config parameter is for API compatibility and is not used internally. ```python from mockafka import FakeProducer # Without config producer = FakeProducer() # With config (ignored) producer = FakeProducer(config={'bootstrap.servers': 'localhost:9092'}) ``` -------------------------------- ### create_topics() Source: https://github.com/alm0ra/mockafka-py/blob/main/_autodocs/api-reference/fake-admin-client.md Creates multiple topics in the mock Kafka cluster. Each topic can be defined with its name and number of partitions. ```APIDOC ## create_topics() ### Description Creates multiple topics in the mock Kafka cluster. Each topic can be defined with its name and number of partitions. ### Method `create_topics(self, topics: list[NewTopic]) -> None` ### Parameters #### Path Parameters - **topics** (list[NewTopic]) - Required - List of NewTopic objects to create ### Raises - `KafkaException`: If topic already exists ### Example ```python from mockafka import FakeAdminClientImpl from mockafka.admin_client import NewTopic admin = FakeAdminClientImpl() admin.create_topics([ NewTopic(topic='orders', num_partitions=5), NewTopic(topic='events', num_partitions=3), ]) ``` ``` -------------------------------- ### Get Message Error Source: https://github.com/alm0ra/mockafka-py/blob/main/_autodocs/api-reference/message.md Retrieves any associated KafkaError object for the message. Returns the error object or None if no error is present. ```python message = Message(error=None) error = message.error() # Returns: None ```