Merge pull request #20 from taskiq-python/feat-rewrite-broker-with-aioboto

feat: implement broker with aiobotocore

Author: danfimov

.github/workflows/code-check.yml

@@ -28,7 +28,7 @@ jobs:
     strategy:
       matrix:
         os: [ubuntu-latest] # eventually add `windows-latest` and `macos-latest`
-        python-version: ["3.10", "3.11", "3.12"]
+        python-version: ["3.10", "3.11", "3.12", "3.13", "3.14"]
     services:
       ministack:
         image: ministackorg/ministack:1.3.53

docker-compose.yml

@@ -22,19 +22,6 @@ services:
     networks:
       - taskiq-sqs-network
 
-  redis:
-    image: bitnamilegacy/redis:7.4.2
-    environment:
-      ALLOW_EMPTY_PASSWORD: "yes"  # pragma: allowlist secret
-    healthcheck:
-      test: ["CMD", "redis-cli", "ping"]
-      interval: 5s
-      timeout: 5s
-      retries: 3
-      start_period: 10s
-    ports:
-      - 6379:6379
-
 networks:
   taskiq-sqs-network:
     driver: bridge

examples/example_broker.py

@@ -8,33 +8,51 @@
 
 import asyncio
 
-import boto3
 import dotenv
+from aiobotocore.session import get_session
 
 from taskiq_sqs import S3Bucket, S3ResultBackend, SQSBroker
 
 
 dotenv.load_dotenv()
 
 QUEUE_NAME = "my-queue"
-QUEUE_URL = f"http://localhost:4566/000000000000/{QUEUE_NAME}"
-
-
-boto3.client("sqs").create_queue(QueueName=QUEUE_NAME)
-
-broker = SQSBroker(QUEUE_URL, sqs_region_override="us-east-1").with_result_backend(
-    S3ResultBackend(bucket=S3Bucket(name="response-bucket"))
+ENDPOINT_URL = "http://localhost:4566"
+AWS_REGION = "us-east-1"
+
+
+broker = SQSBroker(
+    queue_name=QUEUE_NAME,
+    endpoint_url=ENDPOINT_URL,
+    aws_region_name=AWS_REGION,
+).with_result_backend(
+    S3ResultBackend(
+        bucket=S3Bucket(name="response-bucket"),
+        endpoint_url=ENDPOINT_URL,
+        aws_region_name=AWS_REGION,
+    ),
 )
 
 
 @broker.task()
 async def i_love_aws() -> None:
     """I hope my cloud bill doesn't get too high!"""
-    await asyncio.sleep(5.5)
+    await asyncio.sleep(2)
     print("Hello there!")
 
 
+async def ensure_queue_exists() -> None:
+    session = get_session()
+    async with session.create_client(
+        "sqs",
+        region_name=AWS_REGION,
+        endpoint_url=ENDPOINT_URL,
+    ) as sqs:
+        await sqs.create_queue(QueueName=QUEUE_NAME)
+
+
 async def main() -> None:
+    await ensure_queue_exists()
     await broker.startup()
     task = await i_love_aws.kiq()
     print(await task.wait_result())

pyproject.toml

@@ -2,7 +2,6 @@
 name = "taskiq-sqs"
 version = "0.0.11"
 description = "SQS Broker for TaskIQ"
-urls = {source = "https://github.com/taskiq-python/taskiq-sqs"}
 readme = "README.md"
 licence = "MIT"
 license = {file = "LICENSE"}
@@ -32,14 +31,13 @@ keywords = ["taskiq", "broker", "aws", "sqs"]
 requires-python = ">=3.10"
 dependencies = [
     "taskiq>=0.12.1",
-    "asyncer~=0.0.5",
-    "boto3~=1.34.34",
     "aiobotocore>=2.13.3",
 ]
 
-# [project.urls]
-# "Bug Tracker" = "https://github.com/taskiq-python/taskiq-sqs/issues"
-# "Repository" = "https://github.com/taskiq-python/taskiq-sqs/"
+[project.urls]
+"Source" = "https://github.com/taskiq-python/taskiq-sqs"
+"Bug Tracker" = "https://github.com/taskiq-python/taskiq-sqs/issues"
+"Repository" = "https://github.com/taskiq-python/taskiq-sqs/"
 
 [dependency-groups]
 dev = [
@@ -59,8 +57,6 @@ lint = [
 ]
 types = [
     "mypy>=2.1.0",
-    "mypy-boto3-sqs>=1.34.101",
-    "boto3-stubs[essential]>=1.34.84",
     "types-aiobotocore[essential]>=3.7.0",
 ]
 examples = [