a = 0
def x():
    global a
    for i in range(100000):
        a += 1

import dis; dis.dis(x)
>>>
# Цикл убран для наглядности
14 LOAD_GLOBAL              1 (a)   # Загружаем в стек глобальную переменную a
16 LOAD_CONST               2 (1)   # Загружаем в стек 1
18 INPLACE_ADD                      # Сложение верхних элементов в стеке

import threading
threads = []
for j in range(5):
    thread = threading.Thread(target=x)
    threads.append(thread)
    thread.start()
for thread in threads:
    thread.join()
assert a == 500000
>>> AssertionError

import asyncio
import random
from contextvars import ContextVar
from aiohttp import web
request_id: ContextVar[int] = ContextVar('request_id')
async def perform_external_request():
    # Cозданная задача всегда будет иметь контекст родительской
    await asyncio.sleep(5)
    print('request_id =', request_id.get())
    # Здесь выполняем запрос к стороннему сервису
async def test_handler(request):
    r = random.randint(1, 100)
    request_id.set(r)
    asyncio.ensure_future(perform_external_request())
    return web.Response(text='ok')
app = web.Application()
app.router.add_route('GET', '/test', test_handler)
web.run_app(app, port=8000)

import uuid
import logging
from contextvars import ContextVar
from aiohttp import web
request_id: ContextVar[str] = ContextVar('request_id')
class RequestIdFilter(logging.Filter):
    def filter(self, record):
        # Добавление нужного поля в запись
        record.request_id = request_id.get()
        return True
logger = logging.getLogger(__name__)
ch = logging.StreamHandler()
# Все сообщения от этого логгера будут иметь текущий X-Request-Id,
# вне зависимости от места вызова!
ch.setFormatter(logging.Formatter('%(request_id)s: %(message)s'))
logger.addFilter(RequestIdFilter())
logger.addHandler(ch)
async def test_handler(request):
    logger.warning('Calling test handler')
    return web.Response(text='OK')
@web.middleware
async def request_id_middleware(request, handler):
    # Установка / чтение request_id
    request_id.set(request.headers.get('X-Request-Id', str(uuid.uuid4())))
    response = await handler(request)
    return response
app = web.Application(middlewares=[request_id_middleware])
app.router.add_route('GET', '/test', test_handler)
web.run_app(app, port=8000)

var1 = contextvars.ContextVar('var1')
var2 = contextvars.ContextVar('var2')
def gen():
    var1.set('gen')
    assert var1.get() == 'gen'
    assert var2.get() == 'main'
    yield 1
    # Это изменение не будет применено, так как между итерациями модификации запрещены
    var1.set('genXXXX')
    # var1 модифицируется снаружи, но внутри генератора изменение не видно,
    # так как в нем эта переменная была изменена
    assert var1.get() == 'gen'
    # var2 меняется "снаружи" без изменения "внутри", поэтому оно доступно
    assert var2.get() == 'main modified'
    yield 2
def main():
    g = gen()
    var1.set('main')
    var2.set('main')
    next(g)
    # Модификация "изнутри" не доступна "снаружи"
    assert var1.get() == 'main'
    var1.set('main modified')
    var2.set('main modified')
    next(g)

import asyncio
import aiotask_context as context
async def test():
    print(context.get('some_data', default='not set'))
loop = asyncio.get_event_loop()
loop.set_task_factory(context.task_factory)
loop.run_until_complete(test())