📚

【boto3】Scheduler のステータス（有効/無効）を切り替える方法

2025/01/10に公開

AWS

Python

Boto3

tech

概要

boto3 を利用して EventBrdige Scheduler のステータス（有効/無効）を切り替える方法

前提

Python version 3.12.4 で動作確認済み
独学なので変なことしてたら教えてください。

スクリプト

switch_state_scheduler.py

import boto3
from botocore.exceptions import ClientError
import time
import inspect
from typing import Literal

class SchedulerClient:
  def __init__(self, profile: str = 'hoge', region: str = 'ap-northeast-1'):
    session = boto3.Session(profile_name=profile)
    self.client = session.client('scheduler', region_name=region)

  def _check_response(self, response: dict) -> None:
    """ API呼び出しのHTTPステータスコードを確認する """
    status_code = response['ResponseMetadata']['HTTPStatusCode']
    caller_function = inspect.stack()[1].function
    if status_code != 200:
      raise ValueError(f'ステータスコード: {status_code}, 関数名: {caller_function}')

  def sc_get_schedule(self, name: str) -> dict:
    """スケジューラの情報を取得"""
    try:
      response = self.client.get_schedule(Name=name)
    except ClientError as e:
      raise # エラー処理
    self._check_response(response)
    return response
  
  def sc_update_schedule(
    self, 
    name: str, 
    state: Literal['ENABLED', 'DISABLED'], 
    flexible_time_window: dict, 
    schedule_expression: str, 
    target: dict
  ) -> dict:
    """ スケジューラを更新"""
    try:
      response = self.client.update_schedule(
        Name = name,
        State = state,
        FlexibleTimeWindow = flexible_time_window,
        ScheduleExpression = schedule_expression,
        Target = target
      )
    except ClientError as e:
      raise # エラー処理
    self._check_response(response)
    return response

def _change_state_scheduler(client: SchedulerClient, scheduler_name: str, new_state: Literal['ENABLED', 'DISABLED']):
  """ 
  スケジューラのステータスを変更します。
  
  Args:
    client : SchedulerClient インスタンス
    scheduler_name : ステータスを変更するスケジューラ名
    new_state : 'ENABLED' もしくは 'DISABLED'
  """
  if not isinstance(client, SchedulerClient):
    raise TypeError('引数clientはSchedulerClient型である必要があります。')
  if not isinstance(scheduler_name, str):
    raise TypeError(f'引数scheduler_nameはstr型である必要があります。現在の型: {type(scheduler_name)}')
  if new_state not in ['ENABLED', 'DISABLED']:
    raise ValueError(f'引数new_stateの値が不正です。指定できる値は"ENABLED"もしくは"DISABLED"です。')
  
  print(f'スケジューラ {scheduler_name} のステータスを {new_state} に変更します。')
  
  response = client.sc_get_schedule(scheduler_name)
  if new_state == response['State']:
    print(f'ステータスは既に {response["State"]} です。')
    return
  
  params = {
    'name': scheduler_name,
    'state': new_state,
    'flexible_time_window': response['FlexibleTimeWindow'],
    'schedule_expression': response['ScheduleExpression'],
    'target': response['Target']
  }
  client.sc_update_schedule(**params)
  
  retry_count = 3  # 再試行回数
  retry_interval = 10  # 再試行間隔
  for _ in range(retry_count):
    time.sleep(retry_interval)
    response = client.sc_get_schedule(scheduler_name)
    if new_state == response['State']:
      break
  else:
    raise RuntimeError('所定の時間内にスケジューラのステータスが変更されませんでした。')
  
  print(f'スケジューラ {scheduler_name} のステータスを {new_state} に変更しました。')

def main():
  client = SchedulerClient()
  params = [
    {
    'client': client,
    'scheduler_name': 'hoge-scheduler',
    'new_state': 'ENABLED'
    }
  ]
  for param in params:
    _change_state_scheduler(**param)

if __name__ == '__main__':
  main()

実行

python switch_state_scheduler.py
---
スケジューラ hoge-scheduler のステータスを ENABLED に変更します。
スケジューラ hoge-scheduler のステータスを ENABLED に変更しました。
---

エラー出力は割愛してます。すみません

概要

前提

スクリプト

実行

Discussion