import subprocess

result = subprocess.run(
    ['echo', 'Hello from the child!'],
    capture_output=True,  # stdout/stderrをキャプチャーする
    encoding='utf-8')

result.check_returncode()
print(result.stdout)
>>>
Hello from the child!

import subprocess
import time

start = time.time()
sleep_procs = []
for _ in range(10):
    proc = subprocess.Popen(['sleep', '1'])
    sleep_procs.append(proc)

time.sleep(0.3)

for proc in sleep_procs:
    proc.communicate()

end = time.time()
delta = end - start
print(f'Finished in {delta:.3} seconds')
>>>
Finished in 1.02 seconds

import subprocess
import os
def run_encrypt(data):
    env = os.environ.copy()
    env['password'] = 'start123'
    proc = subprocess.Popen(
	['openssl', 'enc', '-des3', '-pass', 'env:password'],
	env=env,
	stdin=subprocess.PIPE,
	stdout=subprocess.PIPE)
    proc.stdin.write(data)
    proc.stdin.flush()
    return proc

procs = []
for _ in range(3):
    data = os.urandom(10)  # ランダムな10バイトを生成
    proc = run_encrypt(data)
    procs.append(proc)

for proc in procs:
    out, _ = proc.communicate()
    print(out[-10:])  # 後ろから10バイトをスライス
>>>
b'\x0f\xbc4\x94O\x93\xa5G\xbe\xe3'
b'm\xb3\x89\r\xc9pP7\xdc\xeb'
b"\xda\x16z N=\x850v'"

def run_hash(input_stdin):
    return subprocess.Popen(
	['openssl', 'dgst', '-whirlpool', '-binary'],
	stdin=input_stdin,  # stdinを指定
	stdout=subprocess.PIPE)

encrypt_procs = []
hash_procs = []
for _ in range(3):
    data = os.urandom(100)  # ランダムな100バイトを生成

    encrypt_proc = run_encrypt(data)
    encrypt_procs.append(encrypt_proc)
    hash_proc = run_hash(encrypt_proc.stdout)  # stdoutを指定
    hash_procs.append(hash_proc)

    encrypt_proc.stdout.close()  # 閉じてしまってよい???
    encrypt_proc.stdout = None

for proc in encrypt_procs:
    proc.communicate()
    assert proc.returncode == 0

for proc in hash_procs:
    out, _ = proc.communicate()
    print(out[-10:])
    assert proc.returncode == 0
>>>
'\x99\xd8*\x15~\x88\xd4\x89\x1c3'
b'\x00\x87\xd3\x93Ti\x12v\x01\xaa'
b'\x1b\x85\xdf\x94z\x96\xd3\xb0\x91\x9a'

import subprocess
proc = subprocess.Popen(['sleep', '10'])
try:
    proc.communicate(timeout=0.1)
except subprocess.TimeoutExpired:
    proc.terminate()
    proc.wait()

print('Exist status', proc.poll())
>>>
Exist status -15

import select
import socket
import time
from threading import Thread
def slow_systemcall():
    select.select([socket.socket()],[],[],0.1)

start = time.time()
threads = []
for _ in range(5):
    thread = Thread(target=slow_systemcall)
    thread.start()
    threads.append(thread)

for thread in threads:
    thread.join()

end = time.time()
delta = end - start
print(f'Took {delta:.3f} seconds')
>>>
Took 0.103 seconds

from threading import Lock
class LockingCounter:
    def __init__(self):
	self.lock = Lock()
	self.count = 0  # ロック対象
    def increment(self, offset):
	with self.lock:
	    self.count += offset

from threading import Thread
from queue import Queue

my_queue = Queue()  # キューのクラスが用意されています

def consumer():
    print('Consumer waiting')
    my_queue.get()  # ブロックされます
    print('Consumer done')

thread = Thread(target=consumer)
thread.start()

print('Producer putting')
my_queue.put(object())
print('Producer done')
thread.join()
>>>
Consumer waiting  # アイテム(object)が入ってくるまで待ちます
Producer putting
Producer done
Consumer done

from threading import Thread
from queue import Queue
import time

my_queue = Queue(1)  # バッファサイズが1

def consumer():
    time.sleep(0.1)  # まず0.1秒スリープする
    my_queue.get()
    print('Consumer got 1')
    my_queue.get()
    print('Consumer got 2')
    print('Consumer done')

thread = Thread(target=consumer)
thread.start()

my_queue.put(object())  # producerは立て続けに二つputしようとする
print('Producer put 1')
my_queue.put(object())  # ここでブロックする
print('Producer put 2')
print('Producer done')
thread.join()
>>>
Producer put 1  # 最初にproducerがputするのは前回と同じ
Consumer got 1  # 0.1秒待ってからgetする
Producer put 2  # consumerがgetして、やっとputから戻る
Producer done
Consumer got 2
Consumer done

from threading import Thread
from queue import Queue
import time

in_queue = Queue()
def consumer():
    print('Consumer waiting')
    work = in_queue.get()
    print('Consumer working')
    time.sleep(1)  # この例でのタスクはスリープすること
    print('Consumer done')
    in_queue.task_done()  # タスク完了を宣言する

thread = Thread(target=consumer)
thread.start()

print('Producer putting')
in_queue.put(object())
print('Producer waiting')
in_queue.join()  # in_queueの完了(=task_doneが呼ばれる)までブロックされる
print('Producer done')
thread.join()
>>>
Consumer waiting
Producer putting
Producer waiting
Consumer working
# ここで1秒スリープする
Consumer done
Producer done

from threading import Thread
from queue import Queue
import time

def download(item):
    pass
def resize(item):
    pass
def upload(item):
    pass

class ClosableQueue(Queue):
    SENTINEL = object()
    def close(self):
	self.put(self.SENTINEL)
    def __iter__(self):
	while(True):
	    item = self.get()  # 写真を一つ取り出す。無かったらブロックする
	    try:
		if item is self.SENTINEL:
		    return
		yield item  # まだfinallyは実行しない
	    finally:
		self.task_done()  # この写真の処理が完了

class StoppableWorker(Thread):
    def __init__(self, func, in_queue, out_queue):
	super().__init__()
	self.func = func  # やる作業
	self.in_queue = in_queue
	self.out_queue = out_queue

    def run(self):
	for item in self.in_queue:  # queueをiterateする
	    result = self.func(item)
	    self.out_queue.put(result)

download_queue = ClosableQueue()
resize_queue = ClosableQueue()
upload_queue = ClosableQueue()
done_queue = ClosableQueue()
threads = [
    StoppableWorker(download, download_queue, resize_queue),
    StoppableWorker(resize, resize_queue, upload_queue),
    StoppableWorker(upload, upload_queue, done_queue),
    ]

for thread in threads:
    thread.start()

for _ in range(1000):
    download_queue.put(object())  # object()=写真を入れる

download_queue.close()  # SENTINEL投入
download_queue.join()  # task_done()が呼ばれるまでここでブロックされる
resize_queue.close()  # SENTINEL投入
resize_queue.join()  # task_done()が呼ばれるまでここでブロックされる
upload_queue.close()  # SENTINEL投入
upload_queue.join()  # task_done()が呼ばれるまでここでブロックされる
print(done_queue.qsize(), 'items finished')
for thread in threads:
    thread.join()
>>>
1000 items finished

def start_threads(count, *args):
    threads = [StoppableWorker(*args) for _ in range(count)]
    for thread in threads:
	thread.start()  # スレッドをスタートさせる
    return threads

def stop_threads(closable_queue, threads):
    for _ in threads:
	closable_queue.close()  # SENTINEL投入
    closable_queue.join()  # 全てのtask_done()を待ち、キューをクローズする
    for thread in threads:
	thread.join()  # 全てのスレッドの完了を待つ

download_queue = ClosableQueue()
resize_queue = ClosableQueue()
upload_queue = ClosableQueue()
done_queue = ClosableQueue()

download_threads = start_threads(
    3, download, download_queue, resize_queue)  # ダウンロードは3多重
resize_threads = start_threads(
    4, resize, resize_queue, upload_queue)  # リサイズは4多重
upload_threads = start_threads(
    5, upload, upload_queue, done_queue)  # アップロードは5多重

for _ in range(1000):  # 1000個(枚)の写真を投入
    download_queue.put(object())

stop_threads(download_queue, download_threads)  # 完了待ちする
stop_threads(resize_queue, resize_threads)
stop_threads(upload_queue, upload_threads)

print(done_queue.qsize(), 'items finished')
>>>
1000 items finished

import concurrent.futures
import urllib.request

URLS = ['http://www.foxnews.com/',
	'http://www.cnn.com/',
	'http://europe.wsj.com/',
	'http://www.bbc.co.uk/',
	'http://some-made-up-domain.com/']

# Retrieve a single page and report the URL and contents
def load_url(url, timeout):
    with urllib.request.urlopen(url, timeout=timeout) as conn:
	return conn.read()

# We can use a with statement to ensure threads are cleaned up promptly
with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
    # {future: url, ...}の辞書がfuture_to_urlに入ります。
    # futureはそのcallableの実行を表すオブジェクトです。
    # ...この場合はワーカースレッドですね。
    future_to_url = {executor.submit(load_url, url, 60): url for url in URLS}
    # as_completedはfuture_to_urlのfuturesの完了(またはキャンセル)した
    # インスタンスのiteratorを返します。それをiterateしてfuture
    # = スレッドを得ます
    for future in concurrent.futures.as_completed(future_to_url):
	url = future_to_url[future]
	try:
	    data = future.result()
	except Exception as exc:
	    print('%r generated an exception: %s' % (url, exc))
	else:
	    print('%r page is %d bytes' % (url, len(data)))
>>>
'http://www.foxnews.com/' page is 323006 bytes
'http://www.cnn.com/' page is 1131345 bytes
'http://europe.wsj.com/' generated an exception: HTTP Error 404: Not Found
'http://some-made-up-domain.com/' page is 64668 bytes
'http://www.bbc.co.uk/' page is 300118 bytes

import asyncio
async def some_coroutine():
    ...
    await slow_io_disk_read()
    ...
    await slow_io_network_transfer()
    ...

async def count():
    print("One")
    await asyncio.sleep(1)
    print("Two")

async def main():
    # コルーチンcount()を3つ実行する
    await asyncio.gather(count(), count(), count())

if __name__ == "__main__":
    import time
    s = time.perf_counter()
    # コルーチンメイン関数実行(完了を待つ)
    asyncio.run(main())
    elapsed = time.perf_counter() - s
    print(f"{__file__} executed in {elapsed:0.2f} seconds.")

loop = asyncio.get_event_loop()
try:
    loop.run_until_complete(main())
finally:
    loop.close()

async def run_tasks(handles, interval, output_path):
    with open(output_path, 'wb') as output:
	async def write_async(data):
	    output.write(data)  # ブロックされるI/O
	tasks = []
	for handle in handles:
	    coro = tail_async(handle, interval, write_async)
	    task = asyncio.create_task(coro)
	    tasks.append(task)
	await asyncio.gather(*tasks)

async def run_fully_async(handles, interval, output_path):
    async with WriteThread(output_path) as output:
	tasks = []
	for handle in handles:
	    coro = tail_async(handle, interval, output.write)
	    task = asyncio.create_task(coro)
	    tasks.append(task)
	await asyncio.gather(*tasks)

async with aiofiles.open('filename', mode='r') as f:
    contents = await f.read()
print(contents)
'My file contents'

def some_func():
    # 例えばファイルをオープンする処理
    # ここでの例外はすぐに呼び元に上がる
    try:
	# 例外が上がる可能性のあるオペレーション
    except ZeroDivisionError as e:
	# 想定した例外が上がった場合
    else:
	# tryで例外が上がらなかった場合
	# ここでの例外は呼び元に伝搬する
    finally:
	# (tryに来ていたら)関数がリターンする前に必ず実行される
	# 例えばファイルのクローズ処理

from contextlib import contextmanager
import logging

@contextmanager
def debug_logging(level):
    logger = logging.getLogger()
    old_level = logger.getEffectiveLevel()
    logger.setLevel(level)  # 一時的に指定ログレベルを設定
    try:
	yield
    finally:
	logger.setLevel(old_level)  # ログレベルを戻す

def my_function():
    logging.debug('Some debug data')  # DEBUG
    logging.error('Error log here')  # ERROR 
    logging.debug('More debug data')  # DEBUG

with debug_logging(logging.DEBUG):  # DEBUGレベルのブロック
    print('* Inside:')
    my_function()

print('* After:')
my_function()
>>>
​* Inside:
DEBUG:root:Some debug data  # DEBUGレベルが表示されている
ERROR:root:Error log here
DEBUG:root:More debug data  # DEBUGレベルが表示されている
​* After:
ERROR:root:Error log here

from contextlib import contextmanager
import logging

@contextmanager
def log_level(level, name):
    logger = logging.getLogger(name)
    old_level = logger.getEffectiveLevel()
    logger.setLevel(level)
    try:
	yield logger  # コンテキストloggerを返す
    finally:
	logger.setLevel(old_level)

logging.basicConfig()
with log_level(logging.DEBUG, 'my-log') as logger:
    logger.debug(f'This is a message for {logger.name}!')
    logging.debug('This will not print')
>>>
DEBUG:my-log:This is a message for my-log!

from datetime import datetime, timezone
import pytz
time_format = '%Y-%m-%d %H:%M:%S'
arrival_bos = '2020-08-29 10:01:00'
bos_dt_native = datetime.strptime(arrival_bos, time_format)
edt = pytz.timezone('US/Eastern')
bos_dt = edt.localize(bos_dt_native)  # datetime形式のボストン時間
utc_dt = pytz.utc.normalize(bos_dt.astimezone(pytz.utc))  # UTCに変換
print(utc_dt)
>>>
2020-08-29 14:01:00+00:00

jst = pytz.timezone('Asia/Tokyo')
tokyo_dt = jst.normalize(utc_dt.astimezone(jst))  # JSTに変換
print(tokyo_dt)
>>>
2020-08-29 23:01:00+09:00

import pickle
import copyreg
class GameState:
    def __init__(self, level=0, lives=4, points=0):
	self.level = level
	self.lives = lives
	self.points = points

def pickle_game_state(game_state):  # pickle.dumpsすると呼ばれる
    print("pickling")
    kwargs = game_state.__dict__
    return unpickle_game_state, (kwargs,)  # unpickle用関数を返す

def unpickle_game_state(kwargs):  # pickle.loadsすると呼ばれる
    print("unpickling")
    return GameState(**kwargs)

copyreg.pickle(GameState, pickle_game_state)  # pickle用関数を登録する

state = GameState()
state.points += 1000
print("call pickling")
serialized = pickle.dumps(state)
print("call unpickling")
state_after = pickle.loads(serialized)
print(state_after.__dict__)
>>>
call pickling
pickling  # pickle_game_stateで表示している
b'\x80\x04\x95L\x00\x00\x00\x00\x00\x00\x00\x8c\x08__main__\x94\x8c\x13unpickle_game_state\x94\x93\x94}\x94(\x8c\x05level\x94K\x00\x8c\x05lives\x94K\x04\x8c\x06points\x94M\xe8\x03u\x85\x94R\x94.'
call unpickling
unpickling  # unpickle_game_stateで表示している
{'level': 0, 'lives': 4, 'points': 1000}

def pickle_game_state(game_state):
    kwargs = game_state.__dict__
    kwargs['version'] = 2
    return unpickle_game_state, (kwargs,)

def unpickle_game_state(kwargs):
    version = kwargs.pop('version', 1)
    if version == 1:  # 読んできたバージョンが1なら古いアトリビュートがある
	del kwargs['deleted_attribute']
    return GameState(**kwargs)

>>> from decimal import Decimal
>>> print(Decimal('1.45'))  # 文字列渡し
1.45
>>> print(Decimal(1.45))  # float渡し
1.4499999999999999555910790149937383830547332763671875

from decimal import Decimal, ROUND_UP
rate = Decimal('0.05')
seconds = Decimal('5')
small_cost = rate * seconds / Decimal(60)
print("実の値 - ", small_cost)
print("四捨五入 - ", round(small_cost, 2))
rounded = small_cost.quantize(Decimal('0.01'), rounding=ROUND_UP)
print("切り上げ - ", rounded)
>>>
実の値 -  0.004166666666666666666666666667
四捨五入 -  0.00
切り上げ -  0.01

from cProfile import Profile
profiler = Profile()
profiler.runcall(test)

from pstats import Stats
stats = Stats(profiler)
stats.strip_dirs()
stats.sort_stats('cumulative')
stats.print_stats()
>>>
	 30003 function calls in 0.026 seconds

   Ordered by: cumulative time

   ncalls  tottime  percall  cumtime  percall filename:lineno(function)
	1    0.000    0.000    0.026    0.026 t.py:17(<lambda>)
	1    0.002    0.002    0.026    0.026 t.py:1(insertion_sort)
    10000    0.003    0.000    0.024    0.000 t.py:9(insert_value)
    10000    0.016    0.000    0.016    0.000 {method 'insert' of 'list' objects}
    10000    0.005    0.000    0.005    0.000 {built-in method _bisect.bisect_left}
	1    0.000    0.000    0.000    0.000 {method 'disable' of '_lsprof.Profiler' objects}

import collections
import timeit
def print_result(count, tests):
    avg_iteration = sum(tests) / len(tests)
    print(f'Count {count:>5,} takes {avg_iteration:.4f}s')
def deque_append_benchmark(count)
    def prepare():
	return collections.deque()
    def run(queue):
	for i in range(count):
	    queue.append(i)
    tests = timeit.repeat(
	setup='queue = prepare()',
	stmt='run(queue)',
	globals=locals(),
	repeat=1000
	number=1)
    return print_result(count, tests)

import functools
@functools.total_ordering
class Book:
    def __init__(self, title, due_date):
	self.title = title
	self.due_date = due_date
	self.returnd = False  # 返却フラグ
    def __lt__(self, other):
	return self.due_date < other.due_date  # 返却日で比較する

from heapq import heapify
queue = [
    Book('Pride and Prejudice', '2019-06-10'),
    Book('The Time Machine', '2019-05-30'),
    ...
    ]
queue.sort()
または
heapify(queue)

from heapq import heappop
class NoOverdueBooks(Exception):
    pass
def next_overdue_book(queue, now):
    while queue:
	book = queue[0]  # Most overdue first
	if book.returned:  # 返却済みなら、、
	    heappop(queue)
	    continue
	if book.due_date < now:  # 期限切れなら
	    heappop(queue)
	    return book
	break

    raise NoOverdueBooks

def return_book(queue, book):
    book.returned = True

data = b'shave and a haircut, two bits'
view = memoryview(data)
chunk = view[12:19]
print(type(chunk))
print('Size: ', chunk.nbytes)
print('Data in view', chunk.tobytes())
print('Underlying data:', chunk.obj)
>>>
<class 'memoryview'>
Size:  7
Data in view b'haircut'
Underlying data: b'shave and a haircut, two bits'

my_array = bytearray(b'row, row, row your boat')
my_view = memoryview(my_array)
write_view = my_view[3:13]
write_view[:] = b'1234567890'
print(my_array)
>>>
bytearray(b'row1234567890 your boat')

socket = ... # クライアントへのソケットコネクション
video_cache = ... # 入ってくるビデオストリーム用のキャッシュ
byte_offset = ... # バッファ上にデータが入ってくる位置
size = 1024 * 1024 # 入ってくるデータのチャンクサイズ

video_array = bytearray(video_cache)
write_view = memoryview(video_array)
chunk = write_view[byte_offset:byte_offset + size]
socket.recv_into(chunk)