BOOKPYTHON Telegram 3482
Если вы хотите передать данные по цепочке вызовов, самый простой способ — использовать аргументы функций.

Однако в некоторых случаях модифицировать все функции в цепочке, чтобы передать новые данные, крайне неудобно. Вместо этого вы можете настроить некоторый контекст, который будет доступен для всех функций по цепочке. Как это можно реализовать технически?

Самое простое решение — глобальная переменная. В Python также можно использовать модули и классы как хранилища контекста, так как они, строго говоря, тоже являются глобальными переменными. Вы, вероятно, делаете это ежедневно, например, для работы с логгерами.

Если ваше приложение многопоточное, обычная глобальная переменная не подойдет, так как она не является потокобезопасной. Одновременно может выполняться несколько цепочек вызовов, и каждая из них нуждается в собственном контексте. Модуль threading решает эту проблему с помощью объекта threading.local(), который является потокобезопасным. Вы можете хранить данные, просто устанавливая атрибуты, например: threading.local().symbol = '@'.

Но оба этих подхода не подходят для асинхронных приложений, где функции не только вызываются, но и могут быть приостановлены с помощью await. Если корутина выполняет await, цикл событий может переключиться на другую корутину из совершенно другой цепочки вызовов. Это приведет к некорректной работе, как в следующем примере:


import asyncio
import sys

global_symbol = '.'

async def indication(timeout):
while True:
print(global_symbol, end='')
sys.stdout.flush()
await asyncio.sleep(timeout)

async def sleep(t, indication_t, symbol='.'):
loop = asyncio.get_event_loop()

global global_symbol
global_symbol = symbol
loop.create_task(indication(indication_t))
await asyncio.sleep(t)

loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.gather(
sleep(1, 0.1, '0'),
sleep(1, 0.1, 'a'),
sleep(1, 0.1, 'b'),
sleep(1, 0.1, 'c'),
))


Решить эту проблему можно, если цикл событий будет устанавливать и восстанавливать контекст каждый раз, когда он возобновляет выполнение корутины. Модуль aiotask_context реализует это, изменяя способ создания задач с помощью loop.set_task_factory. Пример рабочей версии:


import asyncio
import sys
import aiotask_context as context

async def indication(timeout):
while True:
print(context.get('symbol'), end='')
sys.stdout.flush()
await asyncio.sleep(timeout)

async def sleep(t, indication_t, symbol='.'):
loop = asyncio.get_event_loop()

context.set(key='symbol', value=symbol)
loop.create_task(indication(indication_t))
await asyncio.sleep(t)

loop = asyncio.get_event_loop()
loop.set_task_factory(context.task_factory)
loop.run_until_complete(asyncio.gather(
sleep(1, 0.1, '0'),
sleep(1, 0.1, 'a'),
sleep(1, 0.1, 'b'),
sleep(1, 0.1, 'c'),
))


👉@BookPython
👍4



tgoop.com/BookPython/3482
Create:
Last Update:

Если вы хотите передать данные по цепочке вызовов, самый простой способ — использовать аргументы функций.

Однако в некоторых случаях модифицировать все функции в цепочке, чтобы передать новые данные, крайне неудобно. Вместо этого вы можете настроить некоторый контекст, который будет доступен для всех функций по цепочке. Как это можно реализовать технически?

Самое простое решение — глобальная переменная. В Python также можно использовать модули и классы как хранилища контекста, так как они, строго говоря, тоже являются глобальными переменными. Вы, вероятно, делаете это ежедневно, например, для работы с логгерами.

Если ваше приложение многопоточное, обычная глобальная переменная не подойдет, так как она не является потокобезопасной. Одновременно может выполняться несколько цепочек вызовов, и каждая из них нуждается в собственном контексте. Модуль threading решает эту проблему с помощью объекта threading.local(), который является потокобезопасным. Вы можете хранить данные, просто устанавливая атрибуты, например: threading.local().symbol = '@'.

Но оба этих подхода не подходят для асинхронных приложений, где функции не только вызываются, но и могут быть приостановлены с помощью await. Если корутина выполняет await, цикл событий может переключиться на другую корутину из совершенно другой цепочки вызовов. Это приведет к некорректной работе, как в следующем примере:


import asyncio
import sys

global_symbol = '.'

async def indication(timeout):
while True:
print(global_symbol, end='')
sys.stdout.flush()
await asyncio.sleep(timeout)

async def sleep(t, indication_t, symbol='.'):
loop = asyncio.get_event_loop()

global global_symbol
global_symbol = symbol
loop.create_task(indication(indication_t))
await asyncio.sleep(t)

loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.gather(
sleep(1, 0.1, '0'),
sleep(1, 0.1, 'a'),
sleep(1, 0.1, 'b'),
sleep(1, 0.1, 'c'),
))


Решить эту проблему можно, если цикл событий будет устанавливать и восстанавливать контекст каждый раз, когда он возобновляет выполнение корутины. Модуль aiotask_context реализует это, изменяя способ создания задач с помощью loop.set_task_factory. Пример рабочей версии:


import asyncio
import sys
import aiotask_context as context

async def indication(timeout):
while True:
print(context.get('symbol'), end='')
sys.stdout.flush()
await asyncio.sleep(timeout)

async def sleep(t, indication_t, symbol='.'):
loop = asyncio.get_event_loop()

context.set(key='symbol', value=symbol)
loop.create_task(indication(indication_t))
await asyncio.sleep(t)

loop = asyncio.get_event_loop()
loop.set_task_factory(context.task_factory)
loop.run_until_complete(asyncio.gather(
sleep(1, 0.1, '0'),
sleep(1, 0.1, 'a'),
sleep(1, 0.1, 'b'),
sleep(1, 0.1, 'c'),
))


👉@BookPython

BY Библиотека Python разработчика | Книги по питону


Share with your friend now:
tgoop.com/BookPython/3482

View MORE
Open in Telegram


Telegram News

Date: |

Done! Now you’re the proud owner of a Telegram channel. The next step is to set up and customize your channel. Joined by Telegram's representative in Brazil, Alan Campos, Perekopsky noted the platform was unable to cater to some of the TSE requests due to the company's operational setup. But Perekopsky added that these requests could be studied for future implementation. Telegram Android app: Open the chats list, click the menu icon and select “New Channel.” best-secure-messaging-apps-shutterstock-1892950018.jpg How to build a private or public channel on Telegram?
from us


Telegram Библиотека Python разработчика | Книги по питону
FROM American