Как влияет порядок вызова create_task и await на контекст логирования и исполнение кода?

Порядок create_task и await: влияние на контекст логирования и исполнение

Это один из самых коварных аспектов asyncio в Python. Порядок вызова create_task и await кардинально влияет на контекст выполнения, логирование и обработку ошибок. Разберу все нюансы.

Основное отличие

import asyncio

async def task():
    print("Task running")
    return "result"

async def main():
    # Вариант 1: create_task — задача запускается ТУТ ЖЕ
    t = asyncio.create_task(task())
    print("After create_task")  # Печатается ПОСЛЕ "Task running"
    result = await t
    
    # Вариант 2: await напрямую — ждём выполнения
    result = await task()
    print("After await")  # Печатается ПОСЛЕ выполнения task()

# create_task немедленно стартует задачу
# await просто ждёт её завершения (если уже запущена)

create_task запускает ТУТ ЖЕ

import asyncio
import contextvars

request_id = contextvars.ContextVar('request_id')

async def background_task():
    req_id = request_id.get(None)
    print(f"Background task: {req_id}")  # Какой request_id?

async def handle_request():
    request_id.set("req-123")
    
    # create_task запускается в текущем контексте
    t = asyncio.create_task(background_task())
    print(f"Main: {request_id.get()}")
    await t

asyncio.run(handle_request())
# Выведет:
# Background task: req-123
# Main: req-123

Задача наследует контекст родительской корутины в момент create_task, а не в момент await.

Контекст логирования: проблема

import asyncio
import logging
from contextvars import ContextVar

request_id_var = ContextVar('request_id')

# Настроим логирование с request_id
logging.basicConfig(level=logging.INFO)

logger = logging.getLogger(__name__)

class RequestIDFilter(logging.Filter):
    def filter(self, record):
        request_id = request_id_var.get(None)
        record.request_id = request_id or "N/A"
        return True

logger.addFilter(RequestIDFilter())
formatter = logging.Formatter('%(request_id)s - %(message)s')

async def worker():
    logger.info("Worker started")  # Какой request_id?
    await asyncio.sleep(0.1)
    logger.info("Worker done")

async def main():
    request_id_var.set("req-456")
    
    # Проблема: create_task запущена, контекст скопирован
    t = asyncio.create_task(worker())
    
    # Но если мы создали задачу ИЗ другого контекста...
    request_id_var.set("req-789")
    
    await t

# Выведет:
# req-456 - Worker started
# req-456 - Worker done

# Потому что контекст скопирован в момент create_task!

Проблема с исключениями

import asyncio

async def failing_task():
    await asyncio.sleep(0.01)
    raise ValueError("Oops!")

async def main():
    # Вариант 1: create_task
    t = asyncio.create_task(failing_task())
    # Ошибка НЕ вызовется, пока мы не await
    print("After create_task")
    try:
        await t  # ОК, ловим ошибку
    except ValueError as e:
        print(f"Caught: {e}")
    
    # Вариант 2: await напрямую
    try:
        await failing_task()  # Ошибка вызывается сразу
    except ValueError as e:
        print(f"Caught: {e}")

asyncio.run(main())
# Выведет:
# After create_task
# Caught: Oops!
# Caught: Oops!

Но есть подвох:

import asyncio

async def failing_task():
    await asyncio.sleep(0.01)
    raise ValueError("Oops!")

async def main():
    t = asyncio.create_task(failing_task())
    # Если мы НЕ await'им задачу до конца main()
    # Ошибка не будет обработана!
    return "Done"

try:
    result = asyncio.run(main())
except ValueError:
    pass

# В Python 3.8+ будет warning:
# Task exception was never retrieved

Порядок выполнения

import asyncio

async def task1():
    print("Task 1 start")
    await asyncio.sleep(0.01)
    print("Task 1 end")

async def task2():
    print("Task 2 start")
    await asyncio.sleep(0.01)
    print("Task 2 end")

async def main():
    print("Main start")
    
    # create_task запускает задачу параллельно
    t1 = asyncio.create_task(task1())
    t2 = asyncio.create_task(task2())
    
    print("After create_task")
    await t1
    print("After t1")
    await t2
    print("After t2")

asyncio.run(main())
# Выведет:
# Main start
# After create_task
# Task 1 start
# Task 2 start
# Task 1 end
# After t1
# Task 2 end
# After t2

# Обе задачи запустились ДО первого await!

Правильный способ: asyncio.gather()

import asyncio

async def task(name):
    print(f"{name} start")
    await asyncio.sleep(0.01)
    print(f"{name} end")
    return f"{name} result"

async def main():
    # Рекомендуемый способ
    results = await asyncio.gather(
        task("Task1"),
        task("Task2"),
        task("Task3"),
        return_exceptions=True  # Ловим ошибки
    )
    return results

results = asyncio.run(main())
print(results)

Контекст и задачи

import asyncio
import contextvars

user_var = contextvars.ContextVar('user')

async def task():
    print(f"User: {user_var.get(None)}")

async def main():
    user_var.set("Alice")
    
    # Контекст скопирован
    t = asyncio.create_task(task())
    
    # Меняем контекст
    user_var.set("Bob")
    
    await t  # Печатает "Alice", не "Bob"

asyncio.run(main())

Лучшая практика: явное управление контекстом

import asyncio
from contextvars import copy_context

async def task():
    print("Task executing")

async def main():
    # Если нужно явно контролировать контекст
    ctx = copy_context()
    t = asyncio.create_task(ctx.run(asyncio.create_task, task()))
    await t

НО лучше просто избегать сложных сценариев.

Real-world пример: обработка request'ов в FastAPI

from fastapi import FastAPI, Request
import asyncio
from contextvars import ContextVar

app = FastAPI()
request_id_var: ContextVar[str] = ContextVar('request_id')

@app.middleware("http")
async def add_request_id(request: Request, call_next):
    request_id = request.headers.get("X-Request-ID", "unknown")
    request_id_var.set(request_id)
    
    response = await call_next(request)
    return response

@app.post("/process")
async def process(data: dict):
    # Запускаем фоновую задачу
    # create_task сохранит request_id в контексте
    asyncio.create_task(background_work(data))
    return {"status": "processing"}

async def background_work(data):
    req_id = request_id_var.get()
    logger.info(f"Background work for {req_id}")
    await asyncio.sleep(1)
    logger.info(f"Done for {req_id}")

Итоги

• create_task запускает задачу ТУТ ЖЕ и наследует контекст
• await просто ждёт результата (если задача уже запущена)
• Контекст скопирован в момент create_task, не в момент await
• Исключения не вызовутся, пока не await'им
• Используй asyncio.gather() для нескольких задач
• Будь осторожен с контекстом в фоновых задачах