如何在Celery任务中用循环生成数据库中未存在的随机字符串

Hey there! Let's get this sorted out. The main issue with your current code is that you're only generating new_string once, outside the while loop—so even if it exists in the database, you never create a new one. Plus, we can clean up the subprocess call and make the loop logic clearer.

First, let's break down what needs to happen:

• Generate a new string using your command
• Check if it already exists in Used_String (with atom=0)
• If it exists, repeat the generation step
• If it doesn't exist, save it to the database and exit the loop

Here's how to rewrite your task properly:

from celery import app
from django.db import IntegrityError, DatabaseError
from .models import User, Used_String
import subprocess
import logging

logger = logging.getLogger(__name__)

def generate_new_string():
    """Helper function to generate a new string via your command"""
    try:
        # Use check_output for simpler command execution (raises error if command fails)
        result = subprocess.check_output(
            "$get_new_string_cmd",  # Replace with your actual command path if needed
            shell=True,
            text=True  # Directly get string output instead of bytes
        )
        return result.strip()
    except subprocess.CalledProcessError as e:
        logger.error(f"Failed to run string generation command: {e}")
        raise  # Re-raise to let Celery handle the task retry

@app.task(bind=True, max_retries=5)  # Add max_retries to avoid infinite task loops
def allocate_new_string(self, user_pk):
    try:
        user = User.objects.get(pk=user_pk)
        max_attempts = 100  # Prevent infinite loop if command keeps generating duplicates
        attempts = 0

        while attempts < max_attempts:
            attempts += 1
            new_string = generate_new_string()
            
            # Check if this string is already used
            if not Used_String.objects.filter(string=new_string, atom=0).exists():
                # Create and save the new used string (create() handles save automatically)
                Used_String.objects.create(user=user, string=new_string, atom=0)
                logger.info(f"New unique string allocated for user {user_pk}: {new_string}")
                return new_string  # Exit the task successfully
            
            logger.debug(f"String {new_string} already exists, retrying... (Attempt {attempts}/{max_attempts})")
        
        # If we hit max attempts without finding a unique string
        logger.error(f"Failed to generate unique string after {max_attempts} attempts for user {user_pk}")
        self.retry(exc=RuntimeError("Max attempts reached for unique string generation"), countdown=60)

    except User.DoesNotExist:
        logger.error(f"User with pk {user_pk} does not exist")
        raise
    except (DatabaseError, IntegrityError) as e:
        logger.error(f"Database error while allocating string: {e}")
        self.retry(exc=e, countdown=30)
    except Exception as e:
        logger.error(f"Unexpected error: {e}")
        raise

Key Improvements Explained: #

• Helper Function: generate_new_string() wraps your command execution logic, so we can easily call it repeatedly in the loop without duplicating code. This makes the task cleaner and easier to maintain.
• Proper While Loop Flow: We generate a string inside each loop iteration, check if it's unique, and only exit the loop when we find one that's not in the database. This fixes your original issue of reusing the same string over and over.
• Infinite Loop Protection: Added max_attempts to stop the loop if your command keeps generating duplicates. If we hit the limit, the Celery task retries after a delay instead of running forever.
• Targeted Error Handling: Replaced the bare except with specific exceptions (like User.DoesNotExist, subprocess.CalledProcessError) so you can debug issues more easily. We also use Celery's self.retry() to handle transient errors (like temporary database outages).
• Cleaner Subprocess Call: subprocess.check_output is simpler than Popen.communicate() and automatically raises an error if the command fails, which helps catch issues with your string generation tool early.

Quick Notes: #

• Make sure $get_new_string_cmd is your actual command—if it's an environment variable, fetch it with os.environ.get("get_new_string_cmd") instead.
• Adjust max_attempts and countdown values based on how often your command generates duplicates and how quickly you want retries to happen.

内容的提问来源于stack exchange，提问作者user11684966