<SYSTEM_TASK:>
Given the following text description, write Python code to implement the functionality described below step by step
<END_TASK>
<USER_TASK:>
Description:
 Работа с директориями
Step1:  Работа с файлами
Step2:  "r" – открытие на чтение (является значением по умолчанию).
"w" – открытие на запись, содержимое файла удаляется, если файла не существует, создается новый.
"x" – открытие на запись, если файла не существует, иначе исключение.
"a" – открытие на дозапись, информация добавляется в конец файла.
"b" – открытие в двоичном режиме.
"t" – открытие в текстовом режиме (является значением по умолчанию).
"+" – открытие на чтение и запись
Step3:  stdin, stdout, stderr
По умолчанию Unix-оболочки связывают файловый дескриптор 0 с потоком стандартного ввода процесса (stdin), файловый дескриптор 1 — с потоком стандартного вывода (stdout), и файловый дескриптор 2 — с потоком диагностики (stderr, куда обычно выводятся сообщения об ошибках).
Step4:  Так как дескрипторы stdout и stderr переопределены в Jupyter notebook. Давайте посмотрим куда они ведут
Step5:  А ведут они как раз в этот ноутбук
Step6:  Что может пойти не так? Да все что угодно
Step7:  <table border="1" class="docutils" align="left">
<colgroup>
<col width="48%">
<col width="34%">
<col width="18%">
</colgroup>
<thead valign="bottom">
<tr class="row-odd"><th class="head">Method</th>
<th class="head">Checks that</th>
</tr>
</thead>
<tbody valign="top">
<tr class="row-even"><td>assertEqual(a, b)</td>
<td><code class="docutils literal"><span class="pre">a</span> <span class="pre">==</span> <span class="pre">b</span></code></td>
</tr>
<tr class="row-odd"><td>assertNotEqual(a, b)</td>
<td><code class="docutils literal"><span class="pre">a</span> <span class="pre">!=</span> <span class="pre">b</span></code></td>
</tr>
<tr class="row-even"><td>assertTrue(x)</td>
<td><code class="docutils literal"><span class="pre">bool(x)</span> <span class="pre">is</span> <span class="pre">True</span></code></td>
</tr>
<tr class="row-odd"><td>assertFalse(x)</td>
<td><code class="docutils literal"><span class="pre">bool(x)</span> <span class="pre">is</span> <span class="pre">False</span></code></td>
</tr>
<tr class="row-even"><td>assertIsNone(x)</td>
<td><code class="docutils literal"><span class="pre">x</span> <span class="pre">is</span> <span class="pre">None</span></code></td>
</tr>
<tr class="row-odd"><td>assertIsNotNone(x)</td>
<td><code class="docutils literal"><span class="pre">x</span> <span class="pre">is</span> <span class="pre">not</span> <span class="pre">None</span></code></td>
</tr>
<tr class="row-even"><td>assertIsInstance(a, b)</td>
<td><code class="docutils literal"><span class="pre">isinstance(a,</span> <span class="pre">b)</span></code></td>
</tr>
<tr class="row-even"><td colspan="2">И другие... https
Step8:  unittest.mock
Как тестировать код, выполняющий внешние вызовы
Step9:  Для начала посмотрим что такое monkey patching.
Step10:  https
Step11:  <div style="float
Step12:  <div style="float
Step13:  Зачем тогда нужны потоки?
Все потому что не все задачи CPU-bound. Есть IO-bound задачи, которые прекрасно параллелятся на несколько CPU. Кто приведет пример?
В качестве примера поднимем HTTP-сервер на порту 8000 (server.go). По адресу http
Step14:  <div style="float
Step15:  <div style="float
Step16:  Как мы видим - треды позволили получить лучший результат (Macbook Pro 2016 - 64 треда).
Чтобы упростить работу с тредами в Python есть модуль concurrent.futures
Step17:  Аналогично можно использовать ProcessPoolExecutor, чтобы вынести работу в пул процессов. 
Сложность многопоточных приложений
Step18:  Вернемся к примеру с загрузкой URL
Step19:  На запрос тратится 1 секунда, и мы ждем 5 секунд на 5 запросов - а ведь могли бы отправить их друг за другом и через секунду получить результаты для всех и обработать.
Подход с callback-ами
Step20:  Генераторы
Это функция, которая генерирует последовательность значений используя ключевое слово yield
Самый простой пример
Step21:  Первый плюс
Step22:  Второй плюс
Step23:  Можно работать с бесконечным потоком данных. Можно обмениваться результатами между отдельными генераторами по мере готовности - то есть иметь дело с несколькими параллельными задачами. При этом не обязательно эти задачи зависят друг от друга.
Для более глубокого понимания и изучения других особенностей - http
Step24:  В контексте лекции важно понять, что выполнение функции-генератора в Python можно приостановить, дождаться нужных данных, а затем продолжить выполнение с места прерывания. При этом сохраняется локальный контекст выполнения и пока мы ждем данных интерпретатор может заниматься другой полезной работой
Asyncio
Асинхронное программирование с использованием библиотеки asyncio строится вокруг понятия Event Loop - "цикл событий". Event loop является основным координатором в асинхронных программах на Python. Он отвечает за
Step25:  <div style="float
Step26:  В Python 3.4 вызов результата корутины выполнялся с помощью конструкции yield from (https
Step27:  В версии 3.5 появились специальные ключевые слова, позволяющие программировать в асинхронном стиле
Step28:  Чтобы программа стала работать асинхронно нужно использовать примитивы, которые есть в библиотеке asyncio
Step29:  Исключения при работе с корутинами работают точно так же как и в синхронном коде
Step30:  Примеры других реализаций Event Loop'ов
Step31:  
Step32:  Завершающий пример (asyncio + multiprocessing)
Step33:  Subprocess. Если останется время...
<ASSISTANT_TASK:>
Python Code:
import os
os.mkdir("/tmp/park-python")
try:
    os.rmdir("/tmp/park-python")
except IOError as err:
    print(err)
path = "/tmp/park-python/lectures/04"
if not os.path.exists(path):
    os.makedirs(path)
os.rmdir("/tmp/park-python")
import shutil
shutil.rmtree("/tmp/park-python")
import pprint
pprint.pprint(list(os.walk(os.curdir)))
Explanation: Работа с директориями
End of explanation
# открываем дескриптор файла для записи
f = open("/tmp/example.txt", "w")
# записываем содержимое
f.write("Технопарк\n")
# обязательно закрываем
f.close()
# открываем дескриптор файла для чтения
f = open("/tmp/example.txt", "r")
# читаем содержимое полностью.
data = f.read()
# обязательно закрываем!
f.close()
print(data)
Explanation: Работа с файлами
End of explanation
# используя context-manager
with open("/tmp/example.txt", "a") as f:
    f.write("МГТУ\n")
with open("/tmp/example.txt", "r") as f:
    print(f.readlines())
# читаем файл по строке, не загружая его полность в память
with open("/tmp/example.txt", "r") as f:
    for line in f:
        print(repr(line))
# Чтобы проверить целостность сохраненного файла
import hashlib
def hash_file(filename):
    h = hashlib.sha1()
    # открываем файл в бинарном виде.
    with open(filename,'rb') as file:
        chunk = 0
        while chunk != b'':
            # читаем кусочками по 1024 байта
            chunk = file.read(1024)
            h.update(chunk)
    # hex-представление полученной суммы.
    return h.hexdigest()
print(hash_file("/tmp/example.txt"))
print(hash_file("/tmp/example.txt"))
with open("/tmp/example.txt", "a") as f:
    f.write("1")
print("После изменений:", hash_file("/tmp/example.txt"))
Explanation: "r" – открытие на чтение (является значением по умолчанию).
"w" – открытие на запись, содержимое файла удаляется, если файла не существует, создается новый.
"x" – открытие на запись, если файла не существует, иначе исключение.
"a" – открытие на дозапись, информация добавляется в конец файла.
"b" – открытие в двоичном режиме.
"t" – открытие в текстовом режиме (является значением по умолчанию).
"+" – открытие на чтение и запись
End of explanation
import sys
print(sys.stdin)
print(sys.stdout)
print(sys.stderr)
print(sys.stdin.fileno())
print(sys.stdout.fileno())
Explanation: stdin, stdout, stderr
По умолчанию Unix-оболочки связывают файловый дескриптор 0 с потоком стандартного ввода процесса (stdin), файловый дескриптор 1 — с потоком стандартного вывода (stdout), и файловый дескриптор 2 — с потоком диагностики (stderr, куда обычно выводятся сообщения об ошибках).
End of explanation
sys.stdout.write("where am I")
Explanation: Так как дескрипторы stdout и stderr переопределены в Jupyter notebook. Давайте посмотрим куда они ведут:
End of explanation
def get_max_length_word(sentence):
    longest_word = None
    words = sentence.split()
    for word in words:
        if not longest_word or len(word) > len(longest_word):
            longest_word = word
    return longest_word
Explanation: А ведут они как раз в этот ноутбук:)
Отладка
Показываем на примере файла debugging.py
Полезные библиотеки для отладки (debugging) - https://github.com/vinta/awesome-python#debugging-tools
Тестирование
Зачем?
End of explanation
import unittest
class LongestWordTestCase(unittest.TestCase):
    
    def test_sentences(self):
        sentences = [
            ["Beautiful is better than ugly.", "Beautiful"],
            ["Complex is better than complicated.", "complicated"]
        ]
        for sentence, correct_word in sentences:
            self.assertEqual(get_max_length_word(sentence), correct_word)
# Обычно в реальных проектах использует механизм автоматического нахождения тестов (discover).
suite = unittest.defaultTestLoader.loadTestsFromTestCase(LongestWordTestCase)
unittest.TextTestRunner().run(suite)
Explanation: Что может пойти не так? Да все что угодно:
SyntaxError
Ошибка в логике
Обратная несовместимость новой версии используемой библиотеки
...
Через полгода после запуска приложения без тестов изменения в код большого приложения вносить очень страшно!
Некоторые даже используют TDD - Test-Driven Development.
unittest
End of explanation
class BoomException(Exception):
    pass
class Material:
    def __init__(self, name, reacts_with=None):
        self.name = name
        self.reacts_with = reacts_with or []
    def __repr__(self):
        return self.name
class Alchemy:
    def __init__(self):
        self.materials = []
    def add(self, material):
        for existing_material in self.materials:
            if material.name not in existing_material.reacts_with:
                continue
            self.materials = []
            raise BoomException("{0} + {1}".format(existing_material.name, material.name))
        self.materials.append(material)
    def remove(self, material):
        self.materials.remove(material)
# 2Na + 2H2O = 2NaOH + H2 (Не повторять дома!!! Щелочь чрезвычайно опасна!)
alchemy = Alchemy()
material_ca = Material("Ca", reacts_with=[])
material_h20 = Material("H2O", reacts_with=["Na"])
material_na = Item("Na", reacts_with=["H2O"])
alchemy.add(material_ca)
alchemy.add(material_h20)
try:
    alchemy.add(material_na)
except BoomException:
    print("We are alive! But all items lost!")
import unittest
class AlchemyTest(unittest.TestCase):
    def setUp(self):
        self.alchemy = Alchemy()
    
    def test_add(self):
        self.alchemy.add(Material("C"))
        self.alchemy.add(Material("F"))
        self.assertEqual(len(self.alchemy.materials), 2)
    def test_remove(self):
        material_c = Material("C")
        self.alchemy.add(material_c)
        self.assertEqual(len(self.alchemy.materials), 1)
        self.alchemy.remove(material_c)
        self.assertEqual(len(self.alchemy.materials), 0)
    def test_boom(self):
        material_na = Material("Na", reacts_with=["H2O"])
        material_h20 = Material("H2O", reacts_with=["Na"])
        self.alchemy.add(material_na)
        self.assertRaises(BoomException, self.alchemy.add, material_h20)
        self.assertEqual(len(self.alchemy.materials), 0)
# Обычно в реальных проектах использует механизм автоматического нахождения тестов (discover).
suite = unittest.defaultTestLoader.loadTestsFromTestCase(AlchemyTest)
unittest.TextTestRunner().run(suite)
Explanation: <table border="1" class="docutils" align="left">
<colgroup>
<col width="48%">
<col width="34%">
<col width="18%">
</colgroup>
<thead valign="bottom">
<tr class="row-odd"><th class="head">Method</th>
<th class="head">Checks that</th>
</tr>
</thead>
<tbody valign="top">
<tr class="row-even"><td>assertEqual(a, b)</td>
<td><code class="docutils literal"><span class="pre">a</span> <span class="pre">==</span> <span class="pre">b</span></code></td>
</tr>
<tr class="row-odd"><td>assertNotEqual(a, b)</td>
<td><code class="docutils literal"><span class="pre">a</span> <span class="pre">!=</span> <span class="pre">b</span></code></td>
</tr>
<tr class="row-even"><td>assertTrue(x)</td>
<td><code class="docutils literal"><span class="pre">bool(x)</span> <span class="pre">is</span> <span class="pre">True</span></code></td>
</tr>
<tr class="row-odd"><td>assertFalse(x)</td>
<td><code class="docutils literal"><span class="pre">bool(x)</span> <span class="pre">is</span> <span class="pre">False</span></code></td>
</tr>
<tr class="row-even"><td>assertIsNone(x)</td>
<td><code class="docutils literal"><span class="pre">x</span> <span class="pre">is</span> <span class="pre">None</span></code></td>
</tr>
<tr class="row-odd"><td>assertIsNotNone(x)</td>
<td><code class="docutils literal"><span class="pre">x</span> <span class="pre">is</span> <span class="pre">not</span> <span class="pre">None</span></code></td>
</tr>
<tr class="row-even"><td>assertIsInstance(a, b)</td>
<td><code class="docutils literal"><span class="pre">isinstance(a,</span> <span class="pre">b)</span></code></td>
</tr>
<tr class="row-even"><td colspan="2">И другие... https://docs.python.org/3.5/library/unittest.html</td>
</tr>
</tbody>
</table>
Протестируем class.
End of explanation
import requests
def get_location_city(ip):
    data = requests.get("https://freegeoip.net/json/{ip}".format(ip=ip)).json()
    return data["city"]
def get_ip():
    data = requests.get("https://httpbin.org/ip").json()
    return data["origin"]
get_location_city(get_ip())
Explanation: unittest.mock
Как тестировать код, выполняющий внешние вызовы: чтение файла, запрос содержимого URL?
End of explanation
import math
def fake_sqrt(num):
    return 42
original_sqrt = math.sqrt
math.sqrt = fake_sqrt
# вызываем ф-ю, которую мы запатчили.
print(math.sqrt(16))
math.sqrt = original_sqrt
math.sqrt(16)
import unittest
from unittest.mock import patch, Mock
class FakeIPResponse:
    def json(self):
        return {"origin": "127.0.0.1"}
class LongestWordTestCase(unittest.TestCase):
    @patch('requests.get', Mock(return_value=FakeIPResponse()))
    def test_get_ip(self):
        self.assertEqual(get_ip(), "127.0.0.1")
suite = unittest.defaultTestLoader.loadTestsFromTestCase(LongestWordTestCase)
unittest.TextTestRunner().run(suite)
from unittest.mock import Mock
mock = Mock()
mock.method(1, 2, 3, test='wow')
mock.method.assert_called_with(1, 2, 3, test='wow')
mock.non_existing_method.assert_not_called()
Explanation: Для начала посмотрим что такое monkey patching.
End of explanation
STEPS = 50000000
# Простая программа, складывающая числа.
def worker(steps):
    count = 0
    for i in range(steps):
        count += 1
    return count
%timeit -n1 -r1 worker(STEPS)
print("Напомните преподавателю показать actvity monitor")
Explanation: https://docs.python.org/3/library/unittest.mock.html
Библиотека coverage позволяет оценить степень покрытия кода тестами.
Помимо unit-тестирования существует масса других типов тестов:
* Интеграционные (как разные компоненты взаимодействуют друг с другом)
* Функциональные (напр. Selenium)
* Тестирование производительности (бенчмарки)
* ...
Полезные библиотеки для тестирования - https://github.com/vinta/awesome-python#testing
Многопоточность
Что такое процесс?
UNIX является многозадачной операционной системой. Это означает, что одновременно может быть запущена более чем одна программа. Каждая программа, работающая в некоторый момент времени, называется процессом.
http://www.kharchuk.ru/%D0%A1%D1%82%D0%B0%D1%82%D1%8C%D0%B8/15-unix-foundations/80-unix-processes
End of explanation
import threading
import queue
result_queue = queue.Queue()
STEPS = 50000000
NUM_THREADS = 2
def worker(steps):
    count = 0
    for i in range(steps):
        count += 1
    result_queue.put(count)
def get_count_threaded():    
    count = 0
    threads = []
    for i in range(NUM_THREADS):
        t = threading.Thread(target=worker, args=(STEPS//NUM_THREADS,))
        threads.append(t)
        t.start()
    for i in range(NUM_THREADS):
        count += result_queue.get()
    return count
%timeit -n1 -r1 get_count_threaded()
Explanation: <div style="float:left;margin:0 10px 10px 0" markdown="1">![title](img/g3.png)</div>
Что такое поток?
Многопоточность является естественным продолжением многозадачности. Каждый из процессов может выполнятся в несколько потоков. Программа выше исполнялась в одном процессе в главном потоке.
<div style="float:left;margin:0 10px 10px 0" markdown="1">![title](img/threads.jpg)</div>
http://www.cs.miami.edu/home/visser/Courses/CSC322-09S/Content/UNIXProgramming/UNIXThreads.shtml
Логичный шаг предположить, что 2 потока выполнят программу выше быстрее. Проверим?
End of explanation
import multiprocessing
NUM_PROCESSES = 2
STEPS = 50000000
result_queue = multiprocessing.Queue()
def worker(steps):
    count = 0
    for i in range(steps):
        count += 1
    result_queue.put(count)
def get_count_in_processes():    
    count = 0
    processes = []
    for i in range(NUM_PROCESSES):
        p = multiprocessing.Process(target=worker, args=(STEPS//NUM_PROCESSES,))
        processes.append(p)
        p.start()
    for i in range(NUM_PROCESSES):
        count += result_queue.get()
    return count
%timeit -n1 -r1 get_count_in_processes()
Explanation: <div style="float:left;margin:0 10px 10px 0" markdown="1">![title](img/g4.png)</div>
GIL
https://jeffknupp.com/blog/2012/03/31/pythons-hardest-problem/
Ок. Неужели выхода нет? Есть - multiprocessing
Мультипроцессинг
End of explanation
import requests
STEPS = 100
def download():
    requests.get("http://127.0.0.1:8000").text
# Простая программа, загружающая контент URL-странички. Типичная IO-bound задача.
def worker(steps):
    for i in range(steps):
        download()
%timeit -n1 -r1 worker(STEPS)
Explanation: Зачем тогда нужны потоки?
Все потому что не все задачи CPU-bound. Есть IO-bound задачи, которые прекрасно параллелятся на несколько CPU. Кто приведет пример?
В качестве примера поднимем HTTP-сервер на порту 8000 (server.go). По адресу http://localhost:8000 будет отдаваться небольшой кусочек текста. Наша задача - скачивать контент по этому адресу.
End of explanation
import threading
STEPS = 100
NUM_THREADS = 2
def worker(steps):
    count = 0
    for i in range(steps):
        download()
def run_worker_threaded():    
    threads = []
    for i in range(NUM_THREADS):
        t = threading.Thread(target=worker, args=(STEPS//NUM_THREADS,))
        threads.append(t)
        t.start()
    for t in threads:
        t.join()
%timeit -n1 -r1 run_worker_threaded()
Explanation: <div style="float:left;margin:0 10px 10px 0" markdown="1">![title](img/g1.png)</div>
End of explanation
import multiprocessing
NUM_PROCESSES = 2
def worker(steps):
    count = 0
    for i in range(steps):
        download()
def run_worker_in_processes():    
    processes = []
    for i in range(NUM_PROCESSES):
        p = multiprocessing.Process(target=worker, args=(STEPS//NUM_PROCESSES,))
        processes.append(p)
        p.start()
    for p in processes:
        p.join()
%timeit -n1 -r1 run_worker_in_processes()
Explanation: <div style="float:left;margin:0 10px 10px 0" markdown="1">![title](img/g2.png)</div>
Ради интереса попробуем мультипроцессинг для этой задачи:
End of explanation
import concurrent.futures
import requests
STEPS = 100
def download():
    return requests.get("http://127.0.0.1:8000").text
def run_in_executor():
    executor = concurrent.futures.ThreadPoolExecutor(max_workers=64)
    future_to_url = {executor.submit(download): i for i in range(STEPS)}
    for future in concurrent.futures.as_completed(future_to_url):
        i = future_to_url[future]
        try:
            data = future.result()
        except Exception as exc:
            print('%d generated an exception: %s' % (i, exc))
        else:
            pass
            #print('%d page is %d bytes' % (i, len(data)))
    executor.shutdown()
%timeit -n1 -r1 run_in_executor()
Explanation: Как мы видим - треды позволили получить лучший результат (Macbook Pro 2016 - 64 треда).
Чтобы упростить работу с тредами в Python есть модуль concurrent.futures: 
он предоставляет доступ к 2-м высокоуровневым объектам: ThreadPoolExecutor и ProcessPoolExecutor
End of explanation
counter = 0
def worker(num):
    global counter
    for i in range(num):
        counter += 1
worker(1000000)
print(counter)
import threading
counter = 0
def worker(num):
    global counter
    for i in range(num):
        counter += 1
threads = []
for i in range(10):
    t = threading.Thread(target=worker, args=(100000,))
    threads.append(t)
    t.start()
for t in threads:
    t.join()
#print(counter)
import threading
counter = 0
lock = threading.Lock()
def worker(num):
    global counter
    for i in range(num):
        lock.acquire()
        counter += 1
        lock.release()
threads = []
for i in range(10):
    t = threading.Thread(target=worker, args=(100000,))
    threads.append(t)
    t.start()
for t in threads:
    t.join()
print(counter)
# deadlock example
import threading
counter = 0
lock = threading.Lock()
def print_counter():
    lock.acquire()
    print(counter)
    lock.release()
def worker():
    global counter
    lock.acquire()
    print_counter()
    counter += 1
    lock.release()
    
worker()
Explanation: Аналогично можно использовать ProcessPoolExecutor, чтобы вынести работу в пул процессов. 
Сложность многопоточных приложений
End of explanation
import time
def request(i):
    print(f"Sending request {i+1}")
    time.sleep(1)
    print(f"Got response from request {i+1}")
    print()
for i in range(5):
    request(i)
Explanation: Вернемся к примеру с загрузкой URL: можем ли мы сделать еще лучше?
загружать странички еще быстрее
потреблять меньше памяти, не расходуя ее на создание потоков
не задумываться о синхронизации потоков
Асинхронное программирование
Мотивация - IO-операции очень медленные, нужно заставить программу выполнять полезную работу во время ожидания ввода-вывода.
Сравнение Latency некоторых операций (https://gist.github.com/jboner/2841832)
<table align="left">
    <tr>
        <td>L1 CPU cache reference</td>
        <td>0.5 ns</td>
        <td></td>
    </tr>
    <tr>
        <td>Main memory reference</td>
        <td>100 ns</td>
        <td>200x L1 cache</td>
    </tr>
    <tr>
        <td>Read 1 MB sequentially from memory</td>
        <td>250,000 ns (250 us)</td>
        <td></td>
    </tr>
    <tr>
        <td>Round trip within same datacenter</td>
        <td>500,000 ns (500 us)</td>
        <td></td>
    </tr>
    <tr>
        <td>Read 1 MB sequentially from SSD*</td>
        <td>1,000,000 ns (1,000 us, 1ms)</td>
        <td></td>
    </tr>
    <tr>
        <td>Read 1 MB sequentially from disk</td>
        <td>20,000,000 ns (20,000 us, 20 ms)</td>
        <td>80x memory, 20X SSD</td>
    </tr>
    <tr>
        <td>Send packet CA->Netherlands->CA</td>
        <td>150,000,000 ns (150,000 us, 150 ms)</td>
        <td></td>
    </tr>    
</table>
Вернемся к упрощенному варианту нашей программы, загружавшей URL в одном потоке синхронно.
End of explanation
import time
def request(i):
    print("Sending request %d" % i)
    def on_data(data):
        print("Got response from request %d" % i)
    return on_data
callbacks = []
for i in range(5):
    cb = request(i)
    callbacks.append(cb)
time.sleep(1)
for cb in callbacks:
    cb("data")
Explanation: На запрос тратится 1 секунда, и мы ждем 5 секунд на 5 запросов - а ведь могли бы отправить их друг за другом и через секунду получить результаты для всех и обработать.
Подход с callback-ами:
End of explanation
def simple_gen():
    yield 1
    yield 2
gen = simple_gen()
print(next(gen))
print(next(gen))
print(next(gen))
gen = simple_gen()
for i in gen:
    print(i)
Explanation: Генераторы
Это функция, которая генерирует последовательность значений используя ключевое слово yield
Самый простой пример:
End of explanation
def fib():
    a, b = 0, 1
    while True:
        yield a
        a, b = b, a + b
gen = fib()
for i in range(6):
    print(next(gen))
Explanation: Первый плюс: получить значения, не загружая все элементы в память. Яркий пример - range.
Чуть посложнее (с состоянием):
End of explanation
def coro():
    next_value = yield "Hello"
    yield next_value
c = coro()
print(next(c))
print(c.send("World"))
Explanation: Второй плюс: получить значения сразу после того как они были вычислены
Корутины на основе генераторов:
End of explanation
import time
def request(i):
    print("Sending request %d" % i)
    data = yield
    print("Got response from request %d" % i)
generators = []
for i in range(5):
    gen = request(i)
    generators.append(gen)
    next(gen)
time.sleep(1)
for gen in generators:
    try:
        gen.send("data")
    except StopIteration:
        pass
Explanation: Можно работать с бесконечным потоком данных. Можно обмениваться результатами между отдельными генераторами по мере готовности - то есть иметь дело с несколькими параллельными задачами. При этом не обязательно эти задачи зависят друг от друга.
Для более глубокого понимания и изучения других особенностей - http://www.dabeaz.com/finalgenerator/
End of explanation
import asyncio
loop = asyncio.get_event_loop()
loop.run_forever()
Explanation: В контексте лекции важно понять, что выполнение функции-генератора в Python можно приостановить, дождаться нужных данных, а затем продолжить выполнение с места прерывания. При этом сохраняется локальный контекст выполнения и пока мы ждем данных интерпретатор может заниматься другой полезной работой
Asyncio
Асинхронное программирование с использованием библиотеки asyncio строится вокруг понятия Event Loop - "цикл событий". Event loop является основным координатором в асинхронных программах на Python. Он отвечает за:
шедулинг корутин и коллбеков
регистрацию отложенных вызовов
регистрацию таймеров
Это позволяет писать программы так, что в момент блокирующих IO операций контекст выполнения будет переключаться на другие задачи, ждущие выполнения.
End of explanation
import asyncio
def cb():
    print("callback called")
    loop.stop()
loop = asyncio.get_event_loop()
loop.call_later(delay=3, callback=cb)
print("start event loop")
loop.run_forever()
Explanation: <div style="float:left;margin:0 10px 10px 0" markdown="1">![title](img/loop.jpg)</div>
End of explanation
import asyncio
@asyncio.coroutine
def return_after_delay():
    yield from asyncio.sleep(3)
    print("return called")
loop = asyncio.get_event_loop()
print("start event loop")
loop.run_until_complete(return_after_delay())
Explanation: В Python 3.4 вызов результата корутины выполнялся с помощью конструкции yield from (https://www.python.org/dev/peps/pep-0380/), а функции-корутины помечались декоратором @asyncio.coroutine
End of explanation
import asyncio
async def return_after_delay():
    await asyncio.sleep(3)
    print("return called")
loop = asyncio.get_event_loop()
print("start event loop")
loop.run_until_complete(return_after_delay())
Explanation: В версии 3.5 появились специальные ключевые слова, позволяющие программировать в асинхронном стиле: async и await
async - ключевое слово, позволяющее обозначить функцию как асинхронную (корутина, coroutine). Такая функция может прервать свое выполнение в определенной точке (на блокирующей операции), а затем, дождавшись результата этой операции, продолжить свое выполнение.
await позволяет запустить такую функцию и дождаться результата.
End of explanation
import asyncio
async def get_data():
    await asyncio.sleep(1)
    return "boom"
async def request(i):
    print(f"Sending request {i+1}")
    data = await get_data()
    print(f"Got response from request {i+1}: {data}")
loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.gather(*[request(i) for i in range(5)]))
Explanation: Чтобы программа стала работать асинхронно нужно использовать примитивы, которые есть в библиотеке asyncio:
End of explanation
import asyncio
async def get_data():
    await asyncio.sleep(1)
    raise ValueError
async def request(i):
    print("Sending request %d" % i)
    try:
        data = await get_data()
    except ValueError:
        print("Error in request %d" % i)
    else:
        print("Got response from request %d" % i)
loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.gather(*[request(i) for i in range(5)]))
Explanation: Исключения при работе с корутинами работают точно так же как и в синхронном коде:
End of explanation
import aiohttp
import asyncio
STEPS = 100
async def download(loop):
    async with aiohttp.ClientSession(loop=loop) as session:
        async with session.get("http://127.0.0.1:8000") as response:
            return await response.text()
async def worker(steps, loop):
    await asyncio.gather(*[download(loop) for x in range(steps)])
loop = asyncio.get_event_loop()
%timeit -n1 -r1 loop.run_until_complete(worker(STEPS, loop))
Explanation: Примеры других реализаций Event Loop'ов:
Tornado IOLoop (https://github.com/tornadoweb/tornado)
Twisted reactor (https://twistedmatrix.com/trac/)
pyuv (https://github.com/saghul/pyuv)
PyGame (http://pygame.org/hifi.html)
...
Мы еще вернемся к asyncio в лекции про интернет и клиент-серверные приложения. В том числе подробно посмотрим на сетевые операции - неблокирующее чтение и запись в сокеты.
Список библиотек, написанных поверх asyncio - https://github.com/python/asyncio/wiki/ThirdParty
А пока вернемся к нашему примеру и перепишем его, используя асинхронный подход и asyncio в частности.
End of explanation
# future example.
import asyncio
async def slow_operation(future):
    try:
        await asyncio.wait_for(asyncio.sleep(1), 2) 
    except asyncio.TimeoutError:
        future.set_exception(ValueError("Error sleeping"))
    else:
        future.set_result('Future is done!')
def got_result(future):
    if future.exception():
        print("Exception:", type(future.exception()))
    else:
        print(future.result())
    loop.stop()
loop = asyncio.get_event_loop()
future = asyncio.Future()
future.add_done_callback(got_result)
asyncio.ensure_future(slow_operation(future))
loop.run_forever()
# выносим блокирующие вызовы в пул тредов
import asyncio
import requests
async def main():
    loop = asyncio.get_event_loop()
    future1 = loop.run_in_executor(None, requests.get, 'http://127.0.0.1:8000')
    future2 = loop.run_in_executor(None, requests.get, 'http://127.0.0.1:8000')
    response1 = await future1
    response2 = await future2
    print(response1.text)
    print(response2.text)
loop = asyncio.get_event_loop()
loop.run_until_complete(main())
Explanation: 
End of explanation
# asyncio + multiprocessing
import aiohttp
import asyncio
import multiprocessing
NUM_PROCESSES = 2
STEPS = 100
async def download(loop):
    async with aiohttp.ClientSession(loop=loop) as session:
        async with session.get("http://127.0.0.1:8000") as response:
            return await response.text()
async def worker(steps, loop):
    await asyncio.gather(*[download(loop) for x in range(steps)], loop=loop)
def run(steps):
    loop = asyncio.new_event_loop()
    loop.run_until_complete(worker(steps, loop))
def run_in_processes():    
    processes = []
    for i in range(NUM_PROCESSES):
        p = multiprocessing.Process(target=run, args=(STEPS//NUM_PROCESSES,))
        processes.append(p)
        p.start()
    for p in processes:
        p.join()
%timeit -n1 -r1 run_in_processes()
Explanation: Завершающий пример (asyncio + multiprocessing)
End of explanation
import subprocess
import os
result = subprocess.run(["ls", "-l", os.getcwd()], stdout=subprocess.PIPE)
print(result.stdout)
# используя shell
result = subprocess.run(
    "ls -l " + os.getcwd() + "|grep debug",
    stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True
)
print(result.stdout)
Explanation: Subprocess. Если останется время...
End of explanation
<END_TASK>