Python异步日志报错修复：coroutine未被await问题

Let’s break down the issues in your code and fix them step by step:

1. Unnecessary self Parameter in asy_log #

Your asy_log function is defined with a self parameter, but it’s not a method of a class. This causes parameter misalignment when you call it with arguments, and will lead to errors when trying to concatenate logsource + logtext (since logsource would end up being the first argument you pass, not the second).

Fix: Remove the self parameter from the function definition.

2. Calling Async Function Without Awaiting It #

The error RuntimeWarning: coroutine 'asy_log' was never awaited happens because you’re calling the async function directly like a regular synchronous function:

asy_log("Warning message", "Errorsource1", 'warning')

Async functions return coroutines, which need to be either:

• Awaited inside another async function, or
• Executed via the asyncio event loop (using asyncio.run() for Python 3.7+).

3. Unnecessary await on Synchronous Logging Methods #

The standard logging module’s methods (like logger.warning()) are synchronous. Using await on them does nothing useful—you can safely remove the await here.

4. Handling None Values to Avoid Concatenation Errors #

If logsource or logtext is None, concatenating them will throw an error. We should handle those cases gracefully to prevent crashes.

Corrected Full Code #

import asyncio
import logging
import sys

# Set up logger
logger = logging.getLogger("asyncio")
logger.setLevel(logging.INFO)
formatter = logging.Formatter('%(asctime)s %(levelname)s %(message)s')

# Stream handler (output to console)
streamhandler = logging.StreamHandler(stream=sys.stdout)
streamhandler.setLevel(logging.INFO)
streamhandler.setFormatter(formatter)
logger.addHandler(streamhandler)

# File handler (output to log file)
filehandler = logging.FileHandler('test3.log')
filehandler.setLevel(logging.INFO)
filehandler.setFormatter(formatter)
logger.addHandler(filehandler)

# Fixed async log function
async def asy_log(logtext=None, logsource=None, logcriticality=None):
    # Handle None values to prevent concatenation errors
    logsource = logsource or ""
    logtext = logtext or ""
    # Format log message cleanly
    loginfo = f"{logsource}: {logtext}" if logsource else logtext
    
    # Get the appropriate logging method (default to INFO if criticality is invalid)
    log_method = getattr(logger, logcriticality.lower(), logger.info)
    # Execute the synchronous logging method (no await needed)
    log_method(loginfo)

# Execute the async function using asyncio.run()
async def main():
    await asy_log("Warning message", "Errorsource1", 'warning')

asyncio.run(main())

Key Notes #

• Use asyncio.run() instead of the older get_event_loop() pattern—it’s simpler and handles loop setup/teardown automatically.
• Always await async functions or run them via the event loop to avoid the "never awaited" warning.
• If you need truly asynchronous logging (to avoid blocking the event loop with file I/O), consider using a library like aiologger. For most use cases, though, the standard logging module works fine when called from async code (since the blocking time is minimal).

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