refactor: code style

Author: qwejid

src/nats_queue/nats_job.py

@@ -1,7 +1,7 @@
 from datetime import datetime, timedelta
 import time
 import uuid
-from typing import Any, Optional, Dict
+from typing import Any, Dict
 
 
 class Job:

src/nats_queue/nats_queue.py

@@ -1,5 +1,4 @@
 import logging
-import nats
 from nats.aio.client import Client
 from nats.js.errors import BadRequestError
 import json

src/nats_queue/nats_worker.py

@@ -3,9 +3,9 @@
 import json
 import logging
 import time
-from typing import Any, Awaitable, Callable, Dict, List, Optional
+from typing import Awaitable, Callable, Dict, List, Optional
 import uuid
-from nats_queue.nats_limiter import FixedWindowLimiter, IntervalLimiter, Limiter
+from nats_queue.nats_limiter import FixedWindowLimiter, IntervalLimiter
 from nats.js.client import JetStreamContext
 from nats.aio.client import Client
 from nats.aio.msg import Msg
@@ -107,7 +107,7 @@ async def _process_task(self, job: Msg):
                 delay = int(planned_time.total_seconds())
                 await job.nak(delay=delay)
                 logger.debug(
-                    f"Job: {job_data["name"]} id={job_data["id"]} is scheduled for later. "
+                    f"Job: {job_data["name"]} id={job_data["id"]} is scheduled later"
                     f"Requeueing in {delay} seconds."
                 )
                 return
@@ -120,7 +120,8 @@ async def _process_task(self, job: Msg):
                 return
 
             logger.info(
-                f'Job: {job_data["name"]} id={job_data["id"]} is started with data={job_data["data"]}) in queue={job_data["queue_name"]}'
+                f"""Job: {job_data["name"]} id={job_data["id"]} is started
+                with data={job_data["data"]}) in queue={job_data["queue_name"]}"""
             )
 
             timeout = job_data["meta"]["timeout"]
@@ -160,17 +161,20 @@ async def fetch_messages(
         try:
             msgs = await sub.fetch(count, timeout=self.fetch_timeout)
             logger.debug(
-                f"Consumer: name={(await sub.consumer_info()).name} fetched {len(msgs)} messages"
+                f"""Consumer: name={(await sub.consumer_info()).name}
+                fetched {len(msgs)} messages"""
             )
             return msgs
         except TimeoutError:
             logger.debug(
-                f"Consumer: name={(await sub.consumer_info()).name} failed to fetch messages: TimeoutError"
+                f"""Consumer: name={(await sub.consumer_info()).name}
+                failed to fetch messages: TimeoutError"""
             )
             return []
         except Exception as e:
             logger.error(
-                f"Consumer: name={(await sub.consumer_info()).name} error while fetching messages: {e}"
+                f"""Consumer: name={(await sub.consumer_info()).name}
+                error while fetching messages: {e}"""
             )
             raise
 
@@ -183,12 +187,14 @@ async def get_subscriptions(self) -> List[JetStreamContext.PullSubscription]:
                     topic, durable=f"worker_group_{priority}"
                 )
                 logger.info(
-                    f"Consumer: name={self.name} successfully subscribed to topic {topic}."
+                    f"""Consumer: name={self.name}
+                    successfully subscribed to topic {topic}."""
                 )
                 subscriptions.append(sub)
             except Exception as e:
                 logger.error(
-                    f"Consumer: name={self.name} error while subscribing to topic {topic}: {e}"
+                    f"""Consumer: name={self.name} error
+                    while subscribing to topic {topic}: {e}"""
                 )
                 raise
 

tests/test_job.py

@@ -1,4 +1,3 @@
-import uuid
 import pytest
 from nats_queue.nats_job import Job
 

tests/test_queue.py

@@ -1,4 +1,3 @@
-import uuid
 import pytest
 import pytest_asyncio
 import json
@@ -55,7 +54,7 @@ async def test_queue_close_success():
     await queue.setup()
     await queue.manager.delete_stream(queue.name)
     await queue.close()
-    assert queue.client.is_closed == True
+    assert queue.client.is_closed is True
 
 
 @pytest.mark.asyncio

tests/test_rate_limit.py

@@ -1,6 +1,4 @@
-import asyncio
 import pytest
-import time
 from nats_queue.nats_limiter import FixedWindowLimiter, IntervalLimiter
 
 
@@ -33,7 +31,7 @@ async def test_rate_limiter_increment():
     count = limiter_interval.inc()
 
     assert limiter.count == 2
-    assert count == None
+    assert count is None
 
 
 @pytest.mark.asyncio

tests/test_workers.py

@@ -122,11 +122,11 @@ async def test_worker_fetch_messages_success(get_client):
     jobs = [
         Job(
             queue_name="my_queue",
-            name=f"task_1",
+            name="task_1",
         ),
         Job(
             queue_name="my_queue",
-            name=f"task_2",
+            name="task_2",
         ),
     ]
     await queue.addJobs(jobs)
@@ -329,7 +329,7 @@ async def test_worker_get_subscriptions_error(get_client):
     client = get_client
 
     worker = Worker(client, name="my_queue", processor=process_job)
-    with pytest.raises(Exception) as e:
+    with pytest.raises(Exception):
         await worker.get_subscriptions()