shafin.hasnat/RMQ-cleanup
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

added type in clean checker

Browse Source
This commit is contained in:
syed.shafin 2025-09-17 23:14:37 +06:00
parent cdf381192e
commit 3e4ddb22d9
2 changed files with 22 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
.dockerignore Normal file
View File

@ -0,0 +1,3 @@
venv/
*.json
*.dev.*

29
main.py
View File

@ -4,7 +4,7 @@ import sys
import argparse import argparse
from typing import Optional, Tuple, Dict, Any from typing import Optional, Tuple, Dict, Any
class RmqPurger: class RMQCleaner:
def __init__( def __init__(
self, self,
host: str = "localhost", host: str = "localhost",
@ -38,7 +38,7 @@ class RmqPurger:
self.connection.close() self.connection.close()
print("✅ Disconnected from RabbitMQ") print("✅ Disconnected from RabbitMQ")
def read_unacked_messages(self, job_id: str) -> None: def read_unacked_messages(self, job_id: str, type: str) -> None:
""" """
Read all unacknowledged messages from the queue, excluding messages with specified job_id, Read all unacknowledged messages from the queue, excluding messages with specified job_id,
and save them to unack.json and save them to unack.json
@ -48,6 +48,8 @@ class RmqPurger:
try: try:
job_id = job_id.split(',') job_id = job_id.split(',')
print(f"Job IDs: {job_id}")
print(f"Type: {type}")
# Get queue information # Get queue information
queue_info = self.channel.queue_declare(queue=self.queue_name, passive=True) queue_info = self.channel.queue_declare(queue=self.queue_name, passive=True)
message_count = queue_info.method.message_count message_count = queue_info.method.message_count
@ -75,9 +77,12 @@ class RmqPurger:
try: try:
message = json.loads(body.decode()) message = json.loads(body.decode())
# Save message if job_id doesn't match # Save message if job_id doesn't match
if str(message.get("job_id")) not in job_id: if message.get("type") != type:
messages_to_save.append(message) messages_to_save.append(message)
messages_filtered += 1 else:
if str(message.get("job_id")) not in job_id:
messages_to_save.append(message)
messages_filtered += 1
except json.JSONDecodeError: except json.JSONDecodeError:
print(f"⚠️ Error decoding message: {body}") print(f"⚠️ Error decoding message: {body}")
except Exception as e: except Exception as e:
@ -153,13 +158,17 @@ def parse_args():
required=True, required=True,
help='Comma-separated list of job IDs to exclude') help='Comma-separated list of job IDs to exclude')
parser.add_argument('--type',
required=True,
help='Type of operation (example: createClusters)')
return parser.parse_args() return parser.parse_args()
def main(): def main():
args = parse_args() args = parse_args()
# Initialize RmqPurger with command line arguments # Initialize RMQCleaner with command line arguments
purger = RmqPurger( cleaner = RMQCleaner(
host=args.host, host=args.host,
user=args.user, user=args.user,
password=args.password, password=args.password,
@ -167,15 +176,15 @@ def main():
) )
try: try:
purger.connect() cleaner.connect()
# First read and save unacked messages (excluding specified job_ids) # First read and save unacked messages (excluding specified job_ids)
purger.read_unacked_messages(args.job_id) cleaner.read_unacked_messages(args.job_id, args.type)
# Then publish saved messages back to the queue # Then publish saved messages back to the queue
purger.publish_messages() cleaner.publish_messages()
finally: finally:
purger.disconnect() cleaner.disconnect()
if __name__ == "__main__": if __name__ == "__main__":
main() main()