Ex-Machina

python で bot 書くためのフレームワーク

インストール

using pip

pip install -U exmachina

using poetry

poetry add exmachina@latest

使い方

import asyncio

from exmachina import Event, Machina, Retry, RetryFixed

bot = Machina()

bot.create_concurrent_group(
    name='limit',
    entire_calls_limit=4,
    time_calls_limit=3,
    time_limit=1,
)

@bot.emit(interval='1s')
async def emit(event: Event):
    res = await event.execute('execute')
    assert res == 42
    # or
    res = await execute()
    assert res == 42

# 特定の例外でリトライしたい場合
retry = Retry([
  RetryFixed(HTTPError, wait_time=1, retries=5),
])

@bot.execute(concurrent_groups=['limit'], retry=retry)
async def execute():
    return 42

if __name__ == "__main__":
    try:
        wasyncio.run(bot.run())
    except KeyboardInterrupt:
        print("終了")

Emit

定期的に実行したい関数を登録するためのデコレータ

• name
  • Emitでユニークな名前をつける
  • 省略した場合はデコレートした関数名になる
• count
  • 実行回数. これで指定した回数実行した後、aliveがFalseになる
  • Noneを指定した場合は無限回実行する
  • デフォルトはNone
• interval
  • ループのインターバル 1sや1d4hなどと指定できる
  • デフォルトは0s. つまり待機しない
• alive
  • Trueの場合、botの実行時に自動で実行される
  • 手動で起動する場合はFalseを指定する
  • デフォルトはTrue

Concurrent Group

時間あたりや同時実行数を制限するグループ
作成したグループは後述のExecuteに対して設定できる

• name
  • 必須プロパティ. Concurrent Groupでユニークな名前をつける
• entire_calls_limit
  • 全体の実行数制限
  • このグループに所属するExecuteの並列実行数
  • Noneを指定した場合、無制限
  • デフォルトはNone
• time_calls_limit
  • このグループに所属するtime_limit秒あたりに実行"開始"できるExecuteの数
  • デフォルトは1
• time_limit
  • time_calls_limitの制限時間(秒)
  • デフォルト0. つまり、制限なし

Execute

Emitから呼び出される、一回きりのタスク
Emitは主にbotの制御を行い、Executeは計算処理や外部との通信を行う処理を書く想定

• name
  • Executeでユニークな名前をつける
  • 省略した場合はデコレートした関数名になる
• concurrent_groups
  • executeの所属するconcurrent_groupを配列で指定する
  • デフォルトは[]
• retry
  • Execute実行中に発生した例外をトリガーにリトライを行う設定を指定する
  • from exmachina import Retry
  • デフォルトはNone

Event

Emitする関数に渡されるオブジェクト

• 停止中の別のEmit起動や停止
• Executeの実行
• ループの状態を確認できる属性を持つ

Emitの起動と停止

event.start('emit_name')
event.stop('emit_name')

Executeの実行

event.execute('execute_name', *args, **kwargs)

または、直接呼び出し

await execute_name(*args, **kwargs)

event.executeはexecuteのTaskを返す

属性

event.epoch # emitのループ回数
event.previous_execution_time # 直前のループの処理時間
event.count # emitの残りの実行回数(未指定の場合はNone)

Retry

Executeのリトライの設定を書くためのもの
特定の秒数を待機した後、そのまま再実行を行う

• rules
  • 指定するルールの配列
• logger
  • 自前のloggerを渡したい場合に使う

RetryRule

• RetryFixed: 常にwait_timeの間隔でリトライ
• RetryFibonacci: 1,1,2,3,5,...とフィボナッチ数列秒の間隔でリトライ
• RetryRange: minとmaxを指定し、その間の乱数秒の間隔でリトライ
• RetryExponentialAndJitter: 指数関数倍的に最大の待機時間を伸ばしつつリトライ のいずれかを使用する

共通引数

• exception
  • 指定する例外のクラス
• retries
  • リトライ回数. 省略するとRecursionErrorが出るまで再実行する
  • デフォルトはNone
• filter
  • 例外インスタンスを引数にboolを返す関数を指定
  • Trueを返した場合にこのリトライ条件にマッチする
  • HTTPのステータスコードなどで引っ掛けたいリトライ設定が異なる場合などを想定
  • デフォルトはNone. つまり、常にTrueを返す

開発

init

poetry install
poetry shell

fmt

poe fmt

lint

poe lint

test

poe test