Creating a new logger for each async function invocation, good idea or not?

Are there any down sides to [creating a new logger for each coroutine]?

Other than the possible price of creating a logger, another downside is that the logger you create stays associated with the unique name forever and is never destroyed, so you effectively have a memory leak. This is explicitly promised by the documentation:

Multiple calls to getLogger() with the same name will always return a reference to the same Logger object.

I'd recommend just biting the bullet and creating a helper with the desired functionality. Building on Brad Solomon's answer, the wrapper might look like this (untested):

import itertools, weakref, logging

logging.basicConfig(format='%(asctime)-15s %(task_name)s %(message)s')

class TaskLogger:
    _next_id = itertools.count().__next__
    _task_ids = weakref.WeakKeyDictionary()

    def __init__(self):
        self._logger = logging.getLogger('myscript.request')

    def _task_name(self):
        task = asyncio.current_task()
        if task not in self._task_ids:
            self._task_ids[task] = self._next_id()
        return f'task-{self._task_ids[task]}'

    def debug(self, *args, **kwargs):
        self._logger.debug(*args, task_name=self._task_name(), **kwargs)

    # the same for info, etc.

logger = TaskLogger()

Instead of creating a new logger, you may want to consider taking advantage of using custom attributes in the log message, via the extra parameter:

For example:

FORMAT = '%(asctime)-15s %(unique_name)s %(message)s'
# [Configure/format loggers & handlers]

Then within the coroutine call logging a debug level message would look something like:

logger.debug('making request with args %r', args, extra={'unique_name': unique_name())

One other thing to keep in mind: unique_name() could get expensive if you're making a lot of requests. A common pattern when creating concurrency via multiprocessing is to log the calling process ID via os.getpid(). With asyncio, perhaps a very rough cousin would be some identifier for the current Task, which you could get to via asyncio.current_task(). Each task has a _name attribute that should be unique because it calls an incrementing _task_name_counter():

class Task(futures._PyFuture): # Inherit Python Task implementation
    def __init__(self, coro, *, loop=None, name=None):
    # ...
        if name is None:
            self._name = f'Task-{_task_name_counter()}'
        else:
            self._name = str(name)

I was looking for similar solution to log from multiple async tasks. As mentioned above, identifying which line is from which task is impossible.

This is my solution for this problem. If this has downsides or there's a better way, please let me know.

import asyncio
import logging

log_format = '%(levelname).1s,%(thread)s,%(name)s,%(lineno)d: %(message)s'
logging.basicConfig(format=log_format, level=logging.NOTSET)
log = logging.getLogger()


class MyStreamHandler(logging.StreamHandler):
    def __init__(self):
        super().__init__()

    def emit(self, record: logging.LogRecord) -> None:
        try:
            task = asyncio.current_task(asyncio.get_running_loop())
            if task is not None:
                record.__setattr__("thread", f"{record.thread}[{task.get_name()}]")
        except RuntimeError:
            pass
        super().emit(record)


handler = MyStreamHandler()
handler.setFormatter(logging.Formatter(log_format))
log.handlers = []
log.addHandler(handler)


def synchronous():
    log.debug("I'm synchronous")


async def asynchronous():
    log.debug("I'm a debug")
    log.warning("I'm a warning")
    log.error("I'm an error")


async def main():
    loop = asyncio.get_event_loop()

    tasks = []
    for i in range(5):
        tasks.append(asynchronous())

    await asyncio.gather(*tasks)

synchronous()
asyncio.run(main())

sample output:

D,6052,root,30: I'm synchronous
D,6052,asyncio,623: Using proactor: IocpProactor
D,6052[Task-2],root,34: I'm a debug
W,6052[Task-2],root,35: I'm a warning
E,6052[Task-2],root,36: I'm an error
D,6052[Task-3],root,34: I'm a debug
W,6052[Task-3],root,35: I'm a warning
E,6052[Task-3],root,36: I'm an error
D,6052[Task-4],root,34: I'm a debug
W,6052[Task-4],root,35: I'm a warning
E,6052[Task-4],root,36: I'm an error
D,6052[Task-5],root,34: I'm a debug
W,6052[Task-5],root,35: I'm a warning
E,6052[Task-5],root,36: I'm an error
D,6052[Task-6],root,34: I'm a debug
W,6052[Task-6],root,35: I'm a warning
E,6052[Task-6],root,36: I'm an error

To provide some closure to this old question, after following Vinay Sajip's comment about LoggerAdapter, I actually found exactly what I want in the documentation. Quoting from the docs:

If you need a different method, e.g. if you want to prepend or append the contextual information to the message string, you just need to subclass LoggerAdapter and override process() to do what you need. Here is a simple example:

class CustomAdapter(logging.LoggerAdapter):
    """
    This example adapter expects the passed in dict-like object to have a
    'connid' key, whose value in brackets is prepended to the log message.
    """
    def process(self, msg, kwargs):
        return '[%s] %s' % (self.extra['connid'], msg), kwargs

which you can use like this:

logger = logging.getLogger(__name__)
adapter = CustomAdapter(logger, {'connid': some_conn_id})

Then any events that you log to the adapter will have the value of some_conn_id prepended to the log messages.