python関連の良さげな情報メモ: 2023年6月号
Tamtampythonのfinalのアノテーション
You can use the final decorator to declare that a:
- Method should not be overridden
- Class should not be subclassed
# example 3
favorite_color: Final[str] = "red"
favorite_color = "blue"
# mypy foo.py
# error: Cannot assign to final name "favorite_color"
The most straightforward way to create a clone is to leverage slicing:
b = a[:]
In Python 3.9 , it is possible to merge dictionaries using | (bitewise OR)
a = {"a": 1, "b": 2}
b = {"c": 3, "d": 4}
a_and_b = a | b
print(a_and_b)
"""
{"a": 1, "b": 2, "c": 3, "d": 4}
"""
"it is generally better to append a single trailing underscore rather than use an abbreviation or spelling corruption. Thus list_ is better than lst."
list_ = [0, 1, 2, 3, 4]
global_ = "Hi there"
we could use the underscore is as a visual separator for digit grouping in integral, floating-point, and complex number literals – this was introduced in Python 3.6.
The idea was to aid the readability of long literals, or literals whose value should clearly be separated into parts – you can read more about it in PEP 515.
number = 1_500_000
print(number)
"""
15000000
"""
A more concise way to do this is to use the setdefault() method on your dictionary object.
import pprint
text = "It's the first of April. It's still cold in the UK. But I'm going to the museum so it should be a wonderful day"
counts = {}
for word in text.split():
counts.setdefault(word, 0)
counts[word] += 1
pprint.pprint(counts)
The Ellipsis is a Python object that can be called by providing a sequence of three dots (...) or calling the object itself (Ellipsis).
print(...)
"""
Ellipsis
"""
def some_function():
...
# Alternative solution
def another_function():
pass
you’ve defined several type aliases with meaningful names and used them as type hints in your function’s parameters to help clarify their purpose and intended use.
from math import sin
from typing import Callable, TypeAlias
Amplitude: TypeAlias = float
AngularFrequency: TypeAlias = float
PhaseShift: TypeAlias = float
Time: TypeAlias = float
SineWave: TypeAlias = Callable[[Time], float]
def sinusoid(A: Amplitude, ω: AngularFrequency, ϕ: PhaseShift) -> SineWave:
"""Return a function that computes the sine wave at a given time."""
return lambda t: A * sin(ω * t + ϕ)
typer でcli系実装してみたい
Tamtam
イベント駆動プログラミングの3つのパターン
1.callback(基本系？)
トリガーで呼ばれる関数を定義。
2.subject(いわゆるobserverパターン)
subjectを定義し、observerがsubjectを監視。
3.topic(kafkaがこれ)
イベントを定義。イベントがpub/sub側に依存しない。
イベント駆動アーキテクチャ
上をネットワーク越しに行うようにしたもの。
メリット:
処理を分けることにより、保守性拡張性が上がる。ボトルネックとなる処理を非同期にできる。
デメリット:
アーキテクチャが複雑になる。
celeryのtips
- taskの処理は短く
- プリミティブな値を返す
- testではalways_eagerをtrueに
例外処理(リトライ周り)、ログ(キューのステータスやが詰まってる状況を確認)
Tamtam
We published a pre-built python.wasm binary, which can be used to run Python scripts with WebAssembly to get better security and portability.
The sample application can be found at WLR/python/examples/bindings/se2-bindings. It is easy to run and can guide you on how to embed Python in a Wasm application and implement bindings for bi-directional communication.
Tamtam
GILとは
PythonやRuby等の言語に見られる排他ロックの仕組み。複数スレッド下でもロックを持つ単一スレッドでしかバイトコードが実行できず、その他のスレッドは待機状態になる。
GILの特徴
CPUバウンドな処理は一スレッドでしか実行できず、処理の並列化が制限されると認識しておけば良いでしょう。
GILの必要性
そもそもなぜ並列処理に制約をもたらすGILが存在するのでしょうか。
メモリ管理やC連携といった低レベルの仕組みをシンプルにするため。
CPythonはC言語で実装されたライブラリと連携が容易であるが、それらは通常スレッドセーフではないため。
CPUバウンドな処理を効率的に行う
マルチプロセスにすることでCPUバウンドな処理を並列実行することができます。これは各プロセスがインタプリタを保持し、GILはインタプリタ単位で存在するためです。
マルチスレッド下では約1.1秒かかった処理が、マルチプロセス下では約0.65秒となりました。CPUバウンドでも並列実行できていることがわかります。
スレッドと比べてオーバーヘッドが大きいものの、プロセス間でも値を共有できますし、CPUバウンドな処理を並列実行するときには有用です。
IOバウンドな処理を効率的に行う
GILの説明という本筋から外れ、方法論的な話になりますが、現行のPythonではI/O待ちが複数発生するケースでは、asyncioによるイベントループを活用するほうが実用的かもしれません。
asyncioはI/O多重化によって、複数のI/O処理をシングルスレッドで効率的に処理できるため、マルチスレッドによって得られる恩恵と近いものがあります。
Tamtam
FastAPIの仕様で、defで作成されたpath関数は、自動的にマルチスレッド化されて平行処理されます。
一方async defで作成されて場合は、単一スレッドとして処理されます。
PythonではGIL（グローバルインタプリタロック）を使っており、これはスレッドセーフでないコード（C言語ライブラリなど）を他のスレッドと共有してしまうことを防ぐための排他ロックであるため、同時に実行できるスレッドは１つに制限されます。
なのでCPUバウンドな処理では外部スレッドプールであろうと、メインasync loopのスレッドであろうと違いはありません。
→この場合は、defの中でマルチプロセスを組む形なのかな???
Oバウンドな同期処理をasync def内で実行すると前述のようにブロックされてパフォーマンスが低いことが分かります。
一方以下のベンチマークではCPUバウンドな処理はasync defの方がパフォーマンスが良かったとあります。
スレッドの切り替えはコンテキストスイッチが発生するので、その分パフォーマンスに影響したのではと思われます。
かといってどの関数がCPUバウンドか、を常に意識するのは実装やレビューでコストになるため、基本的には同程度のパフォーマンスと考えてdef優先にするのが良いです。
The easiest and most native way to execute a function in a separate process and immediately wait for the results is to use the loop.run_in_executor with ProcessPoolExecutor.
-> そもそもcpu boundな処理でレスポンス遅延が懸念する場合に対して、backgoundtaskをfastapiでは提供し非同期で処理を行えるようにしている。
FastAPIのBackgroundTasksは、HTTPレスポンスがクライアントに送信された後にバックグラウンドでタスクを実行するため、エンドポイント関数がdefでもasync defでも、バックグラウンドタスクの実行タイミング自体は変わりません。しかし、エンドポイント関数やバックグラウンドタスク内で行われる処理が同期的であるべきか非同期的であるべきかは、その処理の内容（特にI/O操作の有無やその量）によります。時間がかかるI/O操作を含む処理は、非同期に行うことで全体のパフォーマンスとスケーラビリティを向上させることが可能です。
-->--> 上のfastapiのBackgroundTasks周りでまとめるのはアリな気がする。パフォーマンス面など
Tamtampydanticのto json
pyserdeの紹介 この方参考になるな
Tamtampytestまとめたい
TamtamPythonの文法としての型エイリアスとライブラリとしてのTypealias
Pythonでは文法として型エイリアスを提供している。
from typing import List
Vector = List[float]
def scale(scalar: float, vector: Vector) -> Vector:
return [scalar * num for num in vector]
# type checks; a list of floats qualifies as a Vector.
new_vector = scale(2.0, [1.0, -4.2, 5.4])
現在の仕様では、MyType1 = F が正しくない型エイリアスの定義とは認識できず、たんなる代入文でしかないため、この行ではエラーを出すのが困難です。
そこで、PEP 613 では typingモジュールに新しく TypeAlias を追加し、型エイリアスの 型 として明示的に定義できるようになりました。
from typing import TypeAlias
def F()->None:pass
MyType1:TypeAlias = F
a: MyType1
次のように TypeAlias を使って明示的に型エイリアスを定義すると
MyType2: TypeAlias = "SampleType"
型チェッカはMyType2が型エイリアスであることを理解して、型名として利用できるようになります。
Put simply, the point of using type aliases is twofold:
- to let the user know, in a relatively simple way, what type an argument should have (should, as we’re still talking about type hints)
- to make static checkers happy.
Tamtam今後のpydantic ORM、非構造化データに対するバリデーション、pydanticカタログなどを提供予定
TamtamPydantic V2にしたい&コントリビュートしたい
Tamtampickleの注意点
pickleはプログラムで管理されるメモリと自己管理するバイト(jsonとか)間をエンコーディングやデコーディングする際に使われるPython用のもの。
Pythonで利用されるオブジェクトの構造を保ったままファイルに書き込めるのが特長
しかしそれはpythonという言語に依存していたりとか、セキュリティ的な問題がある。
以下はとくにセキュリティの問題に触れた内容。
class Foo:
def __init__(self, name, age):
self.name = name
self.age = age
foo = Foo('かわさき', 999)
with open('pickled.pkl', 'wb') as f:
pickle.dump(foo, f)
del foo → インスタンスを削除
with open('pickled.pkl', 'rb') as f:
foo = pickle.load(f) # 復元
print(f'name: {foo.name}, age: {foo.age}') # name: かわさき, age: 999
foo = Foo('かわさき', 999)
with open('pickled.pkl', 'wb') as f:
pickle.dump(foo, f)
del Foo, foo # Fooクラスとそのインスタンスであるfooを削除
with open('pickled.pkl', 'rb') as f:
foo = pickle.load(f) # FooクラスがないのでAttributeError例外
しかし、
class Foo: # 上とは別のFooクラスを定義してみる
def __init__(self, a, b):
self.a = a
self.b = b
with open('pickled.pkl', 'rb') as f:
foo = pickle.load(f) # 復元できてしまう
print(foo.a) # AttributeError例外（復元したfooにはa属性はない）
復元できてしまう（main.Fooというクラスが存在していれば、復元が可能のため）。
つまり、pickle化されたデータを扱う場合、全体的な整合性を取るのはプログラマーに任されるということだ。
これ以外にも、pickle化されたデータを改ざんして、任意のコードを実行させるようにすることも可能だ。
こうしたことから、pickleモジュールは安全ではないことには注意するしないといけない。自分が知らないところでpickle化されたファイルを安易に非pickle化しないようにして、非pickle化するときにはpickle化したときと同じ環境を整えるようにすべし。
copyregを用いて後方互換性を保つ
Tamtamfastapiのurlパラメータの処理がどうしているかをまとめたいな
Tamtam# endpoint
@router.post("/create_user", response_model=UserCreateResponse)
def create_user(
dto: UserCreate, background_tasks: BackgroundTasks
) -> dict[str, str]:
usecase =CreateUserUsecase()
return usecase.handle(dto, background_tasks)
# usecase
def handle(
self,dto: UserCreate, background_tasks: BackgroundTasks
) -> UserCreateResponse:
user = func(UserCreate)
new_user = self.__user_repository.save(user)
background_tasks.add_task(
self.__task.save,
new_user,
)
return UserCreateResponse(user_id=new_user.id)
backgroundtaskはpath_operator関数と同ファイルでbackgroundtaskを呼ばなくても当たり前だが実行される。
しかし、実行環境がAWS Lambdaの場合はそもそもBackgroundTasksが使えない。
- FastAPIのBackgroundTaskが使えない
- BackgroundTaskの使用はuvicorn（WSGI）を前提としている
- 今回はLambdaであるため非同期タスクができない
- 非同期タスクをやるにしてもLambda内で完結しない
- Lambdaはレスポンスを返すとasyncで実行中のスレッドも止める（はず）
→別で非同期タスクを制御しないといけない
- Lambdaはレスポンスを返すとasyncで実行中のスレッドも止める（はず）
Tamtamimport networkx as nx
# 類似度のサンプルデータ
data = [
('A', 'B', 0.8),
('A', 'C', 0.6),
('B', 'D', 0.7)
]
# 空のグラフを作成
G = nx.Graph()
# データを元にグラフにエッジを追加
for node1, node2, score in data:
G.add_edge(node1, node2, weight=1.0/score) # 競合度合いが高いほど距離が短いとする
def predict_conflict(node1, node2):
# 最短経路の長さを取得
distance = nx.shortest_path_length(G, source=node1, target=node2, weight='weight')
# 予測される類似度合いを計算（ここでは単純に距離の逆数を取ることで計算）
predicted_conflict = 1.0 / distance
return predicted_conflict
# 例: ノードAとノードDの間の類似度合いを予測
print(predict_conflict('A', 'D'))