init

Author: hsz1273327

.github/workflows/python-package.yml

@@ -0,0 +1,37 @@
+# This workflow will install Python dependencies, run tests and lint with a variety of Python versions
+# For more information see: https://help.github.com/actions/language-and-framework-guides/using-python-with-github-actions
+
+name: Python package
+
+on:
+  push:
+    branches: [master, dev]
+  pull_request:
+    branches: [master]
+
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        python-version: [3.8, 3.9]
+
+    steps:
+      - uses: actions/checkout@v2
+      - name: Set up Python ${{ matrix.python-version }}
+        uses: actions/setup-python@v2
+        with:
+          python-version: ${{ matrix.python-version }}
+      - name: Install devDependence
+        run: |
+          python -m pip install --upgrade pip
+          pip install mypy pycodestyle coverage lxml
+      - name: Install dependencies
+        run: |
+          if [ -f requirements.txt ]; then pip install -r requirements.txt; fi
+      - name: Lint with pep8
+        run: |
+          pycodestyle --max-line-length=140 --ignore=E501 --first --statistics kafkaproxy
+      - name: Type Hint Check
+        run: |
+          mypy --ignore-missing-imports --show-column-numbers --follow-imports=silent --check-untyped-defs --disallow-untyped-defs --no-implicit-optional --warn-unused-ignores kafkaproxy

.github/workflows/python-publish.yml

@@ -0,0 +1,38 @@
+# This workflows will upload a Python Package using Twine when a release is created
+# For more information see: https://help.github.com/en/actions/language-and-framework-guides/using-python-with-github-actions#publishing-to-package-registries
+
+name: Upload Python Package
+
+on:
+  release:
+    types: [created]
+
+jobs:
+  deploy:
+
+    runs-on: ubuntu-latest
+
+    steps:
+    - uses: actions/checkout@v2
+    - name: Set up Python
+      uses: actions/setup-python@v2
+      with:
+        python-version: '3.x'
+    - name: Install dependencies
+      run: |
+        python -m pip install --upgrade pip
+        pip install setuptools wheel twine
+    - name: Build and publish
+      run: |
+        python setup.py sdist bdist_wheel bdist_egg
+    - name: 'Upload dist'
+      uses: 'actions/upload-artifact@v2'
+      with:
+        name: packages
+        path: dist/*
+    - name: Publish
+      env:
+        TWINE_USERNAME: ${{ secrets.PYPI_USERNAME }}
+        TWINE_PASSWORD: ${{ secrets.PYPI_PASSWORD }}
+      run:
+        twine upload dist/*

.gitignore

@@ -127,3 +127,5 @@ dmypy.json
 
 # Pyre type checker
 .pyre/
+zk
+kafka

CHANGELOG.md

@@ -0,0 +1,3 @@
+# v0.0.1
+
+项目创建

MANIFEST.in

@@ -0,0 +1,5 @@
+include LICENSE
+include README.md
+include CHANGELOG.md
+include requirements.txt
+recursive-include kafkaproxy *.pyx *.pxd *.pxi *.py *.c *.h *.temp *.jinja

README.md

@@ -1,2 +1,141 @@
 # kafkaproxy
-kafka的代理对象
+
+kafka生产者和消费者的代理工具.
+
+代理对象用于推迟初始化.我们可以在需要的地方用代理对象的全局变量直接编写逻辑,避免被代理的对象来回在函数间传递.
+
+## 特性
+
++ 支持代理`kafka-python`,`aiokafka`和`confluent-kafka`的生产者消费者对象.
++ 提供统一通用的生产者消费者接口包装
+
+## 安装
+
++ 只安装本项目不安装被代理对象的依赖: `pip install kafkaproxy`
++ 安装本项目同时确定要代理的对象包为`kafka-python`: `pip install kafkaproxy[kafka]`
++ 安装本项目同时确定要代理的对象包为`aiokafka`: `pip install kafkaproxy[aio]`
++ 安装本项目同时确定要代理的对象包为`confluent-kafka`: `pip install kafkaproxy[confluent]`
+
+## 使用
+
+本项目支持代理3种kafka模块中的对应模块,使用枚举`KafkaType`中的取值在调用方法`initialize_from_addresses`初始化时指定.
+代理对象除了原样代理对象外还提供了生产者和消费者的统一通用接口包装.
+由于对应的方法是动态绑定的,因此如果需要他们的typehints可以用`typing.cast`将代理对象转化为对应的协议对象
+
++ 同步接口生产者使用`ProducerProtocol`
++ 异步接口生产者使用`AioProducerProtocol`
++ 同步接消费产者使用`ConsumerProtocol`
++ 异步接消费产者使用`AioConsumerProtocol`
+
+> 代理`kafka-python`或`confluent-kafka`生产者
+
+```python
+from kafkaproxy import ProducerProxy, KafkaType, ProducerProtocol
+from typing import cast
+import time
+kafkap = ProducerProxy()
+
+
+def run() -> None:
+    p = cast(ProducerProtocol, kafkap)
+    with p.mount() as cli:
+        for i in range(10):
+            cli.publish("topic1", f"send {i}")
+            time.sleep(0.1)
+
+
+# kafkap.initialize_from_addresses("localhost:9094", kafka_type=KafkaType.ConfluentKafka, acks="all")
+kafkap.initialize_from_addresses("localhost:9094", kafka_type=KafkaType.Kafka)
+try:
+    print("start publishing")
+    run()
+finally:
+    print("stoped")
+```
+
+> 代理`kafka-python`或`confluent-kafka`消费者
+
+```python
+from kafkaproxy import ConsumerProxy, KafkaType, ConsumerProtocol
+from typing import cast
+
+kafkac = ConsumerProxy()
+
+
+def run() -> None:
+    c = cast(ConsumerProtocol, kafkac)
+    with c.watch() as g:
+        for record in g:
+            print(record.value)
+
+
+# kafkac.initialize_from_addresses("localhost:9094", "topic1", group_id="test2", kafka_type=KafkaType.Kafka)
+kafkac.initialize_from_addresses("localhost:9094", "topic1", group_id="test2", kafka_type=KafkaType.ConfluentKafka)
+try:
+    print("start watching")
+    run()
+finally:
+    print("stoped")
+
+```
+
+> 代理`aiokafka`生产者
+
+```python
+import asyncio
+from kafkaproxy import ProducerProxy, KafkaType, AioProducerProtocol
+from typing import cast
+
+kafkap = ProducerProxy()
+
+
+async def run() -> None:
+    p = cast(AioProducerProtocol, kafkap)
+    async with p.mount() as cli:
+        for i in range(10):
+            await cli.publish("topic1", f"send {i}")
+            await asyncio.sleep(0.1)
+
+
+async def main() -> None:
+    kafkap.initialize_from_addresses("localhost:9094", kafka_type=KafkaType.AioKafka, acks="all")
+    await run()
+
+
+try:
+    print("start watching")
+    asyncio.run(main())
+finally:
+    print("stoped")
+
+```
+
+> 代理`aiokafka`消费者
+
+```python
+import asyncio
+from kafkaproxy import ConsumerProxy, KafkaAutoOffsetReset, KafkaType, AioConsumerProtocol
+from typing import cast
+
+kafkac = ConsumerProxy()
+
+
+async def run() -> None:
+    c = cast(AioConsumerProtocol, kafkac)
+    async with c.watch() as g:
+        async for record in g:
+            print(record.value)
+
+
+async def main() -> None:
+    kafkac.initialize_from_addresses("localhost:9094", "topic1", group_id="test2", kafka_type=KafkaType.AioKafka, auto_offset_reset=KafkaAutoOffsetReset.earliest)
+    await run()
+
+
+try:
+    print("start watching")
+    asyncio.run(main())
+finally:
+    print("stoped")
+
+```

docker-compose.yml

@@ -0,0 +1,45 @@
+version: "2.4"
+services:
+  # kafka相关工具
+#   zookeeper:
+#     image: docker.io/bitnami/zookeeper:3.8
+#     ports:
+#       - "2181:2181"
+#     volumes:
+#       - "zookeeper_data:/bitnami"
+#     environment:
+#       - ALLOW_ANONYMOUS_LOGIN=yes
+#   kafka:
+#     image: docker.io/bitnami/kafka:3.3
+#     ports:
+#       - "9092:9092"
+#     volumes:
+#       - "kafka_data:/bitnami"
+#     environment:
+#       - KAFKA_CFG_ZOOKEEPER_CONNECT=zookeeper:2181
+#       - ALLOW_PLAINTEXT_LISTENER=yes
+#     depends_on:
+#       - zookeeperk
+    kafka:
+        image: 'bitnami/kafka:latest'
+        ports:
+            - '9094:9094'
+        environment:
+            - KAFKA_ENABLE_KRAFT=yes
+            - KAFKA_CFG_PROCESS_ROLES=broker,controller
+            - KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER
+            - KAFKA_CFG_LISTENERS=CONTROLLER://:9093,CLIENT://:9092,EXTERNAL://:9094
+            - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,CLIENT:PLAINTEXT,EXTERNAL:PLAINTEXT
+            - KAFKA_CFG_ADVERTISED_LISTENERS=CLIENT://kafka:9092,EXTERNAL://localhost:9094
+            - KAFKA_BROKER_ID=1
+            - KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=1@127.0.0.1:9093
+            - KAFKA_CFG_INTER_BROKER_LISTENER_NAME=CLIENT
+            - ALLOW_PLAINTEXT_LISTENER=yes
+            - KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE=true
+            - KAFKA_CFG_LOG_RETENTION_HOURS=12
+            # - KAFKA_AUTO_CREATE_TOPICS_ENABLE=yes
+            # - KAFKA_LOG_RETENTION_HOURS=12
+            # - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CLIENT:PLAINTEXT,EXTERNAL:PLAINTEXT
+            # - KAFKA_CFG_LISTENERS=CLIENT://:9092,EXTERNAL://:9093
+            # - KAFKA_CFG_ADVERTISED_LISTENERS=CLIENT://kafka:9092,EXTERNAL://localhost:9093
+            # - KAFKA_CFG_INTER_BROKER_LISTENER_NAME=CLIENT

kafkaproxy/__init__.py

@@ -0,0 +1,27 @@
+"""kafka生产者和消费者的代理工具.
+
+代理对象用于推迟初始化.我们可以在需要的地方用代理对象的全局变量直接编写逻辑,避免被代理的对象来回在函数间传递.
+本项目支持代理3种kafka模块中的对应模块,使用枚举`KafkaType`中的取值在调用方法`initialize_from_addresses`初始化时指定.
+代理对象除了原样代理对象外还提供了生产者和消费者的统一通用接口包装.
+由于对应的方法是动态绑定的,因此如果需要他们的typehints可以用`typing.cast`将代理对象转化为对应的协议对象
+
++ 同步接口生产者使用`ProducerProtocol`
++ 异步接口生产者使用`AioProducerProtocol`
++ 同步接消费产者使用`ConsumerProtocol`
++ 异步接消费产者使用`AioConsumerProtocol`
+"""
+
+from .consumer import ConsumerProxy
+from .producer import ProducerProxy
+from .models import (
+    ConsumerRecord,
+    KafkaType,
+    KafkaAutoOffsetReset
+)
+
+from .protocols import (
+    ProducerProtocol,
+    AioProducerProtocol,
+    ConsumerProtocol,
+    AioConsumerProtocol
+)