43 lines
1.4 KiB
Python
43 lines
1.4 KiB
Python
import os
|
|
import logging
|
|
from typing import Optional
|
|
import redis.asyncio as redis
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
REDIS_URL = os.getenv("REDIS_URL", "redis://copykar_redis:6379/0" if os.path.exists("/app") else "redis://localhost:6379/0")
|
|
|
|
class RedisQueue:
|
|
def __init__(self, redis_url: Optional[str] = None, queue_key: str = "copykar:queue:incoming"):
|
|
self.redis_url = redis_url or REDIS_URL
|
|
self.queue_key = queue_key
|
|
self.client: Optional[redis.Redis] = None
|
|
|
|
async def connect(self):
|
|
if not self.client:
|
|
self.client = redis.from_url(self.redis_url, decode_responses=True)
|
|
await self.client.ping()
|
|
logger.info(f"Connected to Redis at {self.redis_url}")
|
|
|
|
async def push(self, post_id: int):
|
|
if not self.client:
|
|
await self.connect()
|
|
await self.client.rpush(self.queue_key, str(post_id))
|
|
logger.info(f"Enqueued post ID {post_id} to Redis queue [{self.queue_key}]")
|
|
|
|
async def pop(self) -> Optional[int]:
|
|
if not self.client:
|
|
await self.connect()
|
|
val = await self.client.lpop(self.queue_key)
|
|
return int(val) if val else None
|
|
|
|
async def qsize(self) -> int:
|
|
if not self.client:
|
|
await self.connect()
|
|
return await self.client.llen(self.queue_key)
|
|
|
|
async def close(self):
|
|
if self.client:
|
|
await self.client.aclose()
|
|
logger.info("Redis connection closed.")
|