From 6b6f20a032eb31d747366c8bc270ca81731eaf6a Mon Sep 17 00:00:00 2001 From: Tom van der Lee Date: Fri, 4 Sep 2026 17:09:24 +0200 Subject: Fixed pre-commit issues --- proxy/queue.py | 63 +++++++++++++++++++++++++--------------------------------- 1 file changed, 27 insertions(+), 36 deletions(-) (limited to 'proxy/queue.py') diff --git a/proxy/queue.py b/proxy/queue.py index cfa0f3c..70fe3d0 100644 --- a/proxy/queue.py +++ b/proxy/queue.py @@ -2,8 +2,7 @@ import asyncio import json import logging import os -import traceback -from typing import Type +from typing import ClassVar, Self from ttun_server.redis import RedisConnectionPool from ttun_server.types import Message @@ -16,37 +15,37 @@ class BaseProxyQueue: self.identifier = identifier @classmethod - async def create_for_identifier(cls, identifier: str) -> 'BaseProxyQueue': - raise NotImplementedError(f'Please implement create_for_identifier') + async def create_for_identifier(cls, identifier: str) -> BaseProxyQueue: + raise NotImplementedError("Please implement create_for_identifier") @classmethod - async def get_for_identifier(cls, identifier: str) -> Type['self']: + async def get_for_identifier(cls, identifier: str) -> Self: assert await cls.has_connection(identifier) return cls(identifier) @classmethod async def has_connection(cls, identifier) -> bool: - raise NotImplementedError(f'Please implement has_connection') + raise NotImplementedError("Please implement has_connection") async def enqueue(self, message: Message): - raise NotImplementedError(f'Please implement send_request') + raise NotImplementedError("Please implement send_request") async def dequeue(self) -> Message: - raise NotImplementedError(f'Please implement handle_requests') + raise NotImplementedError("Please implement handle_requests") async def delete(self): - raise NotImplementedError(f'Please implement delete') + raise NotImplementedError("Please implement delete") class MemoryProxyQueue(BaseProxyQueue): - connections: dict[str, asyncio.Queue] = {} + connections: ClassVar[dict[str, asyncio.Queue]] = {} @classmethod async def has_connection(cls, identifier) -> bool: return identifier in cls.connections @classmethod - async def create_for_identifier(cls, identifier: str) -> 'MemoryProxyQueue': + async def create_for_identifier(cls, identifier: str) -> MemoryProxyQueue: instance = cls(identifier) cls.connections[identifier] = asyncio.Queue() @@ -66,67 +65,59 @@ class RedisProxyQueue(BaseProxyQueue): def __init__(self, identifier): super().__init__(identifier) - self.pubsub = RedisConnectionPool()\ - .get_connection()\ - .pubsub() + self.pubsub = RedisConnectionPool().get_connection().pubsub() self.subscription_queue = asyncio.Queue() @classmethod - async def create_for_identifier(cls, identifier: str) -> 'BaseProxyQueue': + async def create_for_identifier(cls, identifier: str) -> BaseProxyQueue: instance = cls(identifier) - await instance.pubsub.subscribe(f'request_{identifier}') + await instance.pubsub.subscribe(f"request_{identifier}") return instance @classmethod - async def get_for_identifier(cls, identifier: str) -> 'RedisProxyQueue': - instance: 'RedisProxyQueue' = await super().get_for_identifier(identifier) + async def get_for_identifier(cls, identifier: str) -> RedisProxyQueue: + instance: RedisProxyQueue = await super().get_for_identifier(identifier) - await instance.pubsub.subscribe(f'response_{identifier}') + await instance.pubsub.subscribe(f"response_{identifier}") return instance @classmethod async def has_connection(cls, identifier) -> bool: logger.debug(await RedisConnectionPool.get_connection().pubsub_channels()) - return f'request_{identifier}' in { + return f"request_{identifier}" in { channel.decode() - for channel - in await RedisConnectionPool \ - .get_connection() \ - .pubsub_channels() + for channel in await RedisConnectionPool.get_connection().pubsub_channels() } async def wait_for_message(self): async for message in self.pubsub.listen(): - match message['type']: - case 'subscribe': + match message["type"]: + case "subscribe": continue case _: - return message['data'] + return message["data"] async def enqueue(self, message: Message): - await RedisConnectionPool \ - .get_connection() \ - .publish(f'request_{self.identifier}', json.dumps(message)) + await RedisConnectionPool.get_connection().publish( + f"request_{self.identifier}", json.dumps(message) + ) async def dequeue(self) -> Message: message = await self.wait_for_message() return json.loads(message) async def delete(self): - await self.pubsub.unsubscribe(f'request_{self.identifier}') + await self.pubsub.unsubscribe(f"request_{self.identifier}") - await RedisConnectionPool.get_connection()\ - .srem('connections', self.identifier) + await RedisConnectionPool.get_connection().srem("connections", self.identifier) class ProxyQueueMeta(type): def __new__(cls, name, superclasses, attributes): - return RedisProxyQueue \ - if 'REDIS_URL' in os.environ \ - else MemoryProxyQueue + return RedisProxyQueue if "REDIS_URL" in os.environ else MemoryProxyQueue class ProxyQueue(BaseProxyQueue, metaclass=ProxyQueueMeta): -- cgit v1.3