"""
Основные определения
"""
import asyncio
import types
# платформенная сопрограмма
async def coro():
await asyncio.sleep(1)
def gen():
yield 11
# Классическая сопрограмма
def classic_coro():
for _ in range(10):
yield from gen()
x = yield "YIELD"
print(x)
c = classic_coro()
print(next(c))
print(next(c))
c.send(1)
print(next(c))
c.send(2)
print(next(c))
c.send(3)
# Генераторные сопрограммы
@types.coroutine
def gen_coro():
for _ in range(10):
yield from asyncio.sleep(1)
async def test():
await gen_coro()
# asyncio.run(test())
"""
В asyncio есть генератор, отдающий переданные ему сопрограммы, в порядке готовности, а не в порядке подачи.
Это выглядит как параллельное исполнение сопрограмм
"""
import random
async def test_coro(id_: int):
await asyncio.sleep(random.randint(1, 5))
return id_
async def task(id_: int):
await asyncio.sleep(5)
return id_
async def main_test():
coros = []
tasks = []
for i in range(5):
tasks.append(
asyncio.create_task(task(i))
)
for i in range(5):
coros.append(test_coro(i))
for c in asyncio.as_completed(coros):
print("ID: ", await c)
# Таски выполняются в фоне, хоть и спят. 5 сек. Они спят асинхронно с test_coro корутинами.
# Нам необязательно дожидаться, пока таска будет выполнена. Но если вы этого хотим, достаточно просто
# вызвать таску с await если мы хотим передать ей управление в конкретной корутине
for t in asyncio.as_completed(tasks):
print("TASK ID: ", await t)
# asyncio.run(main_test())
"""
Если вы не можете переписать блокирующую функцию, как асинхронную. То её надо запустить в отдельном потоке или процессе.
"""
"""
В asyncio есть возможность использовать семафоры и потоки.
Семафор работает так: асинхронный контекстный менеджер семафора в asyncio не блокирует программу целиком.
Он блокирует только конкретную сопрограмму, если счетчик семаформа обращается в ноль.
Также, можно вынести отдельную функцию в поток, если нет возможности сделать её асинхронной.
"""
import time
def dont_coro(id_: int):
time.sleep(2)
return id_
# Выполнение в отдельном потоке и асинхронно
async def test_thread():
thread_tasks = []
for i in range(5):
thread_tasks.append(asyncio.create_task(asyncio.to_thread(dont_coro, i)))
print("test")
for _, dc in zip(range(10), asyncio.as_completed(thread_tasks)):
print("T ID: ", await dc)
await asyncio.sleep(1)
print("HELLO FROM test_thread")
# asyncio.run(test_thread())
"""
Рекомендуется помещать перед ASGI приложением прокси-сервер, который обрабатывает все статические файлы, а также по
возможности, использовать CDN(система доставки контента)
В состав FastAPI входят скрипты для генерации проектов
"""
async def value():
return random.randint(1, 10)
# Асинхронные генераторы
async def agen():
for _ in range(5):
yield await value()
async def main():
async for v in agen():
print(v)
# asyncio.run(main())
"""
Чтобы сделать так, чтобы Асинхронные генераторы стали поддерживать контекстные менеджеры, надо применить к ним
декоратор asynccontextmanager
"""
# Пример асинхронного включения:
async def acomp():
print([x async for x in agen()])
# asyncio.run(acomp())
"""
Также можно определить асинхронное генераторное выражение (x async for x in agen()) где угодно в программе,
но использовать его можно только в контексте асинхронного кода.
"""
"""
Чтобы контролировать исключения в функции
asyncio.gather()
Надо задать аргумент return_exceptions при вызове как return_exceptions=True
По умолчанию return_exceptions=False
"""
"""
Если вручную создать таски в через create_task, то надо дождаться, что они закончатся явно.
Например через await или as_complited
"""
"""
У каждой задачи можно задать имя, это можно использовать как id задачи)))))))))))))))))
Пример:
"""
async def some_task(wait: int) -> int:
print("some_task with wait: ", wait)
await asyncio.sleep(wait)
return 42
async def some_main():
tasks = [
asyncio.create_task(some_task(num), name=f"number_{num}") for num in range(5)
]
for task in tasks:
print("...")
await task
if task.get_name() == "number_4":
print("NUM 4 IS RUNNING!")
task.cancel()
print("NUM 4 IS CANCELED!")
# asyncio.run(some_main())
"""
Задачи можно отменить с помощью метода .cancel() с обычными корутинами так не сделать.
"""
"""
Документация по Python 3.11+ рекомендует использовать TaskGroup() вместо gather()
TaskGroup - это асинхронный контекстный менеджер. Все таски в нем создаются, но await'тя после уже выхода из контекста
Пример:
"""
async def task(id_: int):
await asyncio.sleep(1)
print(f"Hello from task with id: {id_}")
if id_ % 2 == 0:
raise Exception
return id_ * 100
# Можно выполнить хоть 500 задач, все равно они будут выполнены за ~1 секунду
async def task_group():
results = []
try:
async with asyncio.TaskGroup() as tasks:
for i in range(500):
results.append(
tasks.create_task(
task(i)
)
)
print("inner context manager")
except Exception as e:
print(f"ERROR: {e}")
"""
ПО КАКОЙ-ТО ПРИЧИНЕ, ИСКЛЮЧЕНИЕ ПРОБРАСЫВАЕТСЯ НЕ ТОЛЬКО В КОНТЕКСТНЫЙ МЕНЕДЖЕР, НО И В МЕТОД
.result()
"""
try:
# Получение результатов от задач
print("outer context manager", [t.result() for t in results])
except:
...
# asyncio.run(task_group())
"""
Есть также возможость запустить задачи через wait.
Интересный способ. По завершению wait вернет список выполненых задач и тех, которые были не выполнены.
Например, из-за таймаута
wait имеет параметр return_when у которого есть 3 состояния:
* FIRST_COMPLETED - Вернуть done, pending когда выполнена, хотя бы одна задача
* FIRST_EXCEPTION - вернуть done, pending когда было хотя бы одно исключение
* ALL_COMPLETED(По умолчанию) - done, pending, когда выполнены все задачи
Пример:
"""
async def wait_task(wait_time: int):
await asyncio.sleep(wait_time)
async def wait_test():
tasks = [
asyncio.create_task(wait_task(1), name="d1"),
asyncio.create_task(wait_task(2), name="d2"),
asyncio.create_task(wait_task(3), name="d3"),
]
done, pending = await asyncio.wait(tasks, timeout=2.1)
print(
f"{[d.get_name() for d in done]}, "
f"{[p.get_name() for p in pending]}"
)
# asyncio.run(wait_test())
"""
Для работы с файлами асинхронно можно юзать aiofiles это внешная библиотека
"""
