はじめに

こちらの書籍のまとめになります

Effective Python 第 2 版 ――Python プログラムを改良する 90 項目 (Brett Slatkin 　著、黒川利明　訳、石本敦夫　技術監修)

全てのパートをまとめているわけではありません
個人的に難しくて理解できていなかったり腑に落ちていない箇所は省いています
もしくは新たな気づきは特にないなと感じたところも省略しています
コードに関しては書籍のものを丸々掲載するでなく、改変しています（その過程も個人的に有意義な時間でした）
そのような理由からこのブログでは多くの部分を削ってしまっています。オリジナルの書籍はかなり勉強になるなと思いました。興味ある人は是非読んでください。
このページでは本書の６〜７章序盤をまとめています。他の章はこちらを参照ください
- １〜３章
- ４〜５章
- ６〜７章序盤
- ７章中盤〜８章序盤
  - 7 章は全体的に難易度が高くほとんどの部分を削っています
- ８章中盤〜１０章

第六章　メタクラスと属性

get メソッドや set メソッドは使わず属性をそのまま使う

part2 でも触れたプライベート属性の話ともリンクしてくるが、Java や C++と違い、Python ではなるべくgetter/setterは使わない
- この辺の概念はこの記事が非常にわかりやすくまとめてくれています
  - なぜ Python には setter/getter メソッドがないのか？ (Qiita)

再利用可能な`@property`メソッドにディスクリプタを使う

例えば以下のような Pokemon のステータス管理をするクラスがあったとする

class Pokemon:
    def __init__(self, name, hp=1, attack=1):
        self.name = name
        self._hp = hp
        self._attack = attack

    @property
    def hp(self):
        return self._hp

    @hp.setter
    def hp(self, value):
        if not (0 < value < 256):
            raise ValueError('Value must be between 1 and 255')
        self._hp = value

    @property
    def attack(self):
        return self._attack

    @attack.setter
    def attack(self, value):
        if not (0 < value < 256):
            raise ValueError('Value must be between 1 and 256')
        self._attack = value


lizardon = Pokemon('リザードン')
lizardon.hp = 78
lizardon.attack = 84

print(lizardon.hp)        # -> 78
print(lizardon.attack)    # -> 84

pikachu = Pokemon('ピカチュウ', hp=35, attack=55)

print(pikachu.hp)         # -> 35
print(pikachu.attack)     # -> 55

この場合、defense, speed, ...と増えていった場合にいちいち追加するのが面倒
なのでこう書く
- 本書ではこの結論に至るまでに色々と途中経過を掲載しているのですが少し難しく、自分は咀嚼できてないので載せてません。是非みてみてください

from weakref import WeakKeyDictionary


class Status:
    def __init__(self):
        self._values = WeakKeyDictionary()

    def __get__(self, instance, instance_type):
        if instance is None:
            return self
        return self._values.get(instance, 0)

    def __set__(self, instance, value):
        if not (0 < value < 256):
            raise ValueError('Value must be between 1 and 256')
        self._values[instance] = value


class Pokemon:
    def __init__(self, name):
        self.name = name
        hp = Status()
        attack = Status()
        defense = Status()
        special_attack = Status()
        special_defense = Status()
        speed = Status()


pikachu = Pokemon('ピカチュウ')
pikachu.hp = 35
pikachu.attack = 55
pikachu.defense = 40
pikachu.special_attack = 50
pikachu.special_defense = 50
pikachu.speed = 90

print(f'{pikachu.name}の攻撃力は{pikachu.attack}')

[out]
ピカチュウの攻撃力は55

サブクラスをinit_subclassで事前検証する

MetaClass -> valid 用の SuperClass -> SubClass で検証していくと色々大変
- どういう流れで大変かは本書を参照
SuperClass に__init_subclass__を定義することでサブクラスの妥当性をチェックする方が妥当

class Typ:
    typ = None  # サブクラスで指定する必要がある

    def __init_subclass__(cls):
        super().__init_subclass__()
        if cls.typ not in {'Normal', 'Fire', 'Water', 'Grass'}:
            raise ValueError('Typ need a valid typ')


class PokemonFire(Typ):
    typ = 'Fire'


class Pokemon(PokemonFire):
    def __init__(self, name):
        self.name = name


charmander = Pokemon('ヒトカゲ')

以下の場合、サブクラスを定義しようとした時点で ValueError が発生する

class PokemonEsper(Typ):
    typ = 'Esper'

ダイヤモンド継承のような複雑なケースでも__init_subclass__は使える
下記例の場合、BottomにLeftとRightの両方が継承されているので初期化の際に 2 回Top.__init_subclass__が呼び出されそうだが、実際は 1 回しか呼び出されない。上手くできてる

class Top:
    def __init_subclass__(cls):
        super().__init_subclass__()
        print(f'Top -> {cls}')

print()

class Left(Top):
    def __init_subclass__(cls):
        super().__init_subclass__()
        print(f'Left -> {cls}')

print()

class Right(Top):
    def __init_subclass__(cls):
        super().__init_subclass__()
        print(f'Right -> {cls}')

print()

class Bottom(Left, Right):
    def __init_subclass__(cls):
        super().__init_subclass__()
        print(f'Bottom -> {cls}')

print()

class Bottom2(Bottom):
    def __init_subclass__(cls):
        super().__init_subclass__()
        print(f'Bottom2 -> {cls}')

[out]
Top -> <class '__main__.Left'>

Top -> <class '__main__.Right'>

Top -> <class '__main__.Bottom'>
Right -> <class '__main__.Bottom'>
Left -> <class '__main__.Bottom'>

Top -> <class '__main__.Bottom2'>
Right -> <class '__main__.Bottom2'>
Left -> <class '__main__.Bottom2'>
Bottom -> <class '__main__.Bottom2'>

第七章　並行性と並列性

並行性
- 異なるタスクを見かけ上同じ時間に行う。例えば OS は単一プロセッサ上で実行するプログラムを忙しく切り替えている
- Python では比較的容易に並行性プログラムを書くことができる
並列性
- 異なるタスクを実際に同じ時間で行う。各 CPU が別々のプログラムを同時に実行する
- 並行 Python コードを本当に並列に実行することは非常に難しい場合もある。このような困難な状況において、Python をどのように使うのが最良かを理解することが重要

subprocess で外部プログラムを実行する子プロセスを管理する

色んなことを同時実行しようとした挙句、シェルスクリプトがとんでもないことになり始めたら Python のsubprocessの導入を考える
Python3.5 以降はsubprocess.runでコマンドを実行することが公式的に推奨されている

import subprocess


result = subprocess.run(['ls'], stdout=subprocess.PIPE)
print(result.stdout.decode())   # -> Untitled.ipynb

runでなくPopenを使うと、より高度な操作が可能
例えば子プロセスの状態を定期的にポーリングすることが可能

import time


proc = subprocess.Popen(['sleep', '10'])


while proc.poll() is None:
    print('実行中')
    time.sleep(1)
print(proc.poll())

[out]
実行中
実行中
実行中
実行中
実行中
実行中
実行中
実行中
実行中
実行中
0

多数の子プロセスを並列に実行可能

before = time.time()

p03 = subprocess.Popen(['sleep', '3'])
p10 = subprocess.Popen(['sleep', '10'])
proc_list = [p03, p10]

for proc in proc_list:
    proc.communicate()

after = time.time()
print(after - before)   # -> 10.019080877304077(13秒かかってないs)

二つの子プロセスをパイプで繋ぎ、順番に実行することも可能

p1 = subprocess.Popen(['ls'], stdout=subprocess.PIPE)
p2 = subprocess.Popen(['wc', '-l'], stdin=p1.stdout, stdout=subprocess.PIPE)
out, err = p2.communicate()
print(out.decode())   # -> 1

前プロセスの正常終了が確かでない場合など、後プロセスに TimeOut を加えることも可能

proc = subprocess.Popen(['sleep', '120'])
proc.communicate(timeout=1)   # -> TimeoutExpired

マルチスレッドでは高速化できない

Python には複数の実装が存在する C 言語で実装された CPython や Java 実装の Jython など
その中でも CPython が標準実装
CPython の一貫性を保つために Python（や Ruby）では Global Interpreter Lock（GIL）といった排他ロックの仕組みだ導入されている
C++や Java などでいうマルチスレッド（並列性）の仕組みが GIL のせいで機能しない
他言語では複数 CPU を同時に動かせるのに Python ではそれができない
シングルスレッドで計算

import time
import random
from math import gcd


def lcm(m, n):
    """最小公倍数アルゴリズム
    """
    return (m * n) // gcd(m, n)


def lcm_list(m_list, n_list):
    return [lcm(m, n) for m, n in zip(m_list, n_list)]


numbers1 = [random.randint(1, 10**2) for _ in range(10**6)]
numbers2 = [random.randint(1, 10**2) for _ in range(10**6)]

start = time.time()
ans = lcm_list(numbers1, numbers2)
end = time.time()

print(f'{end - start:.3f} sec.')   # -> 0.142 sec.

マルチスレッドで計算、シングルスレッドと同じような時間で終わると思いきや倍以上かかってる

from threading import Thread

start = time.time()

# 処理開始
t1 = Thread(target=lcm_list, args=(numbers1, numbers2))
t2 = Thread(target=lcm_list, args=(numbers1, numbers2))
t1.start()
t2.start()
t1.join()
t2.join()

end = time.time()

print(f'{end - start:.3f} sec.')   # -> 0.305 sec.

なお、下記のような場合は特に sleep させている間にロックが解除され、別スレッドが動き始めるので確かに同時に実行しているといえる

def sleep(n):
    time.sleep(n)


start = time.time()
t1 = Thread(target=sleep, args=(1,))
t2 = Thread(target=sleep, args=(2,))
t3 = Thread(target=sleep, args=(3,))

for t in [t1, t2, t3]:
    t.start()

for t in [t1, t2, t3]:
    t.join()

end = time.time()
print(f'{end - start:.3f} sec.')   # -> 3.002 sec.

スレッド間の線形パイプラインには Queue を使う

例えば下記のような実装をした場合いくつかの問題点がある
1. 前処理が終わってないのにポーリングしていて作業の無駄が発生してしまってる
2. スレッド 3 つが常時 run 状態
3. これらの影響でどこかのqueueに渋滞が発生した場合、メモリクラッシュもありうる

import time
from collections import deque
from threading import Thread, Lock


def plus(item):
    return item + 2

def times(item):
    return item * 2

def minus(item):
    return item - 2


class MyQueue:
    def __init__(self):
        self.items = deque()
        self.lock = Lock()

    def put(self, item):
        with self.lock:
            self.items.append(item)

    def get(self):
        with self.lock:
            return self.items.popleft()


class MyThread(Thread):
    def __init__(self, func, in_queue, out_queue):
        super().__init__()
        self.func = func
        self.in_queue = in_queue
        self.out_queue = out_queue
        self.polled_count = 0  # ポーリングの総回数
        self.work_done = 0     # エンキュー/デキューが行われた回数

    def run(self):
        while True:
            self.polled_count += 1
            try:
                item = self.in_queue.get()
            except IndexError:
                time.sleep(0.01)   # 待ち行列がない時に発生
            except AttributeError:
                return
            else:
                result = self.func(item)
                self.out_queue.put(result)
                self.work_done += 1


queue1 = MyQueue()
queue2 = MyQueue()
queue3 = MyQueue()
queue4 = MyQueue()
threads = [
    MyThread(plus, queue1, queue2),
    MyThread(times, queue2, queue3),
    MyThread(minus, queue3, queue4),
]

# スレッドを起動
# まだ値は追加されてないのでこの時点では実際の処理は始まらない
for thread in threads:
    thread.start()

# 1番目のqueueにエンキュー
for i in range(10**5):
    queue1.put(i)

while len(queue4.items) < 10**5:
    print('処理中...')
    time.sleep(0.1)

print('処理終了')
for thread in threads:
    thread.in_queue = None
    thread.join()

[out]
処理中...
処理中...
処理終了

print(f'スレッド1の無駄: {threads[0].polled_count - threads[0].work_done}')
print(f'スレッド2の無駄: {threads[1].polled_count - threads[1].work_done}')
print(f'スレッド3の無駄: {threads[2].polled_count - threads[2].work_done}')

[out]
スレッド1の無駄: 10
スレッド2の無駄: 12
スレッド3の無駄: 11

なので組み込みモジュールのQueueクラスを使う
- コードは本書の丸コピになってしまいそうなので割愛