Pythonで簡単に定周期処理を実装する方法と実践例10選

import time

while True:
    # 実行したいタスクをここに記述
    print("タスクを実行しています")

    # 指定した秒数だけ待機
    time.sleep(5)  # 5秒間待機

import time
from datetime import datetime

try:
    while True:
        # 現在時刻を取得
        current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")

        # 現在時刻を表示
        print(f"現在時刻: {current_time}")

        # 1秒間待機
        time.sleep(1)
except KeyboardInterrupt:
    print("プログラムを終了します")

現在時刻: 2024-08-03 12:34:56
現在時刻: 2024-08-03 12:34:57
現在時刻: 2024-08-03 12:34:58
現在時刻: 2024-08-03 12:34:59
...

import schedule
import time
from datetime import datetime

def generate_daily_report():
    current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    print(f"[{current_time}] デイリーレポートを生成しています...")
    # ここに実際のレポート生成処理を記述します

# 毎日午前9時にタスクを実行するようにスケジュール
schedule.every().day.at("09:00").do(generate_daily_report)

print("スケジューラを開始しました。Ctrl+Cで終了します。")

try:
    while True:
        schedule.run_pending()
        time.sleep(1)
except KeyboardInterrupt:
    print("スケジューラを終了します。")

スケジューラを開始しました。Ctrl+Cで終了します。
[2024-08-03 09:00:00] デイリーレポートを生成しています...
[2024-08-04 09:00:00] デイリーレポートを生成しています...
[2024-08-05 09:00:00] デイリーレポートを生成しています...
...

import schedule
import time
from datetime import datetime

def generate_weekly_report():
    current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    print(f"[{current_time}] 週次レポートを生成しています...")
    # ここに実際の週次レポート生成処理を記述します

# 毎週月曜日の午前10時にタスクを実行するようにスケジュール
schedule.every().monday.at("10:00").do(generate_weekly_report)

print("週次レポート生成スケジューラを開始しました。Ctrl+Cで終了します。")

try:
    while True:
        schedule.run_pending()
        time.sleep(1)
except KeyboardInterrupt:
    print("スケジューラを終了します。")

週次レポート生成スケジューラを開始しました。Ctrl+Cで終了します。
[2024-08-05 10:00:00] 週次レポートを生成しています...
[2024-08-12 10:00:00] 週次レポートを生成しています...
[2024-08-19 10:00:00] 週次レポートを生成しています...
...

from apscheduler.schedulers.blocking import BlockingScheduler
from datetime import datetime

def job_function():
    print(f"毎分実行: {datetime.now()}")

def another_job_function():
    print(f"30秒ごとに実行: {datetime.now()}")

# スケジューラーの作成
scheduler = BlockingScheduler()

# ジョブの追加
scheduler.add_job(job_function, 'interval', minutes=1)
scheduler.add_job(another_job_function, 'interval', seconds=30)

print("スケジューラーを開始します。Ctrl+Cで終了します。")

try:
    scheduler.start()
except KeyboardInterrupt:
    print("スケジューラーを終了します。")
    scheduler.shutdown()

スケジューラーを開始します。Ctrl+Cで終了します。
30秒ごとに実行: 2024-08-03 12:00:00
30秒ごとに実行: 2024-08-03 12:00:30
毎分実行: 2024-08-03 12:01:00
30秒ごとに実行: 2024-08-03 12:01:00
30秒ごとに実行: 2024-08-03 12:01:30
毎分実行: 2024-08-03 12:02:00
30秒ごとに実行: 2024-08-03 12:02:00
...

from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.triggers.cron import CronTrigger
from datetime import datetime

def business_hour_task():
    current_time = datetime.now()
    print(f"営業時間内のタスクを実行: {current_time}")

# スケジューラーの作成
scheduler = BlockingScheduler()

# 平日の午前9時から午後5時まで、10分おきにタスクを実行
scheduler.add_job(
    business_hour_task,
    CronTrigger(day_of_week='mon-fri', hour='9-17', minute='*/10')
)

print("営業時間スケジューラーを開始します。Ctrl+Cで終了します。")

try:
    scheduler.start()
except KeyboardInterrupt:
    print("スケジューラーを終了します。")
    scheduler.shutdown()

営業時間スケジューラーを開始します。Ctrl+Cで終了します。
営業時間内のタスクを実行: 2024-08-05 09:00:00
営業時間内のタスクを実行: 2024-08-05 09:10:00
営業時間内のタスクを実行: 2024-08-05 09:20:00
...
営業時間内のタスクを実行: 2024-08-05 16:50:00
営業時間内のタスクを実行: 2024-08-05 17:00:00

import asyncio
import time

async def task_a():
    while True:
        print(f"タスクA実行中: {time.strftime('%H:%M:%S')}")
        await asyncio.sleep(2)  # 2秒間隔

async def task_b():
    while True:
        print(f"タスクB実行中: {time.strftime('%H:%M:%S')}")
        await asyncio.sleep(5)  # 5秒間隔

async def task_c():
    while True:
        print(f"タスクC実行中: {time.strftime('%H:%M:%S')}")
        await asyncio.sleep(7)  # 7秒間隔

async def main():
    # 3つのタスクを同時に実行
    await asyncio.gather(task_a(), task_b(), task_c())

if __name__ == "__main__":
    print("非同期タスクを開始します。Ctrl+Cで終了します。")
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("プログラムを終了します。")

非同期タスクを開始します。Ctrl+Cで終了します。
タスクA実行中: 12:00:00
タスクB実行中: 12:00:00
タスクC実行中: 12:00:00
タスクA実行中: 12:00:02
タスクA実行中: 12:00:04
タスクB実行中: 12:00:05
タスクA実行中: 12:00:06
タスクC実行中: 12:00:07
タスクA実行中: 12:00:08
タスクA実行中: 12:00:10
タスクB実行中: 12:00:10
...

import asyncio
import aiohttp
import csv
from bs4 import BeautifulSoup
from datetime import datetime

async def fetch_data(session, url):
    async with session.get(url) as response:
        return await response.text()

async def scrape_website():
    url = "https://example.com/data"
    async with aiohttp.ClientSession() as session:
        html = await fetch_data(session, url)
        soup = BeautifulSoup(html, 'html.parser')
        # ここで必要なデータを抽出
        data = soup.find('div', class_='target-data').text.strip()
        return data

async def save_to_csv(data):
    timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    with open('scraped_data.csv', 'a', newline='') as file:
        writer = csv.writer(file)
        writer.writerow([timestamp, data])

async def main():
    while True:
        try:
            data = await scrape_website()
            await save_to_csv(data)
            print(f"データを保存しました: {data}")
        except Exception as e:
            print(f"エラーが発生しました: {e}")
        await asyncio.sleep(3600)  # 1時間ごとに実行

if __name__ == "__main__":
    print("ウェブスクレイピングを開始します。Ctrl+Cで終了します。")
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("プログラムを終了します。")

import asyncio
import psutil
import smtplib
from email.message import EmailMessage

async def check_system_resources():
    cpu_percent = psutil.cpu_percent()
    memory_percent = psutil.virtual_memory().percent
    return cpu_percent, memory_percent

async def send_alert(subject, body):
    msg = EmailMessage()
    msg.set_content(body)
    msg['Subject'] = subject
    msg['From'] = "alert@example.com"
    msg['To'] = "admin@example.com"

    # SMTPサーバーの設定は環境に応じて変更してください
    server = smtplib.SMTP('localhost')
    server.send_message(msg)
    server.quit()

async def monitor_resources():
    while True:
        cpu, memory = await check_system_resources()
        if cpu > 90 or memory > 90:
            subject = "システムリソース警告"
            body = f"CPU使用率: {cpu}%, メモリ使用率: {memory}%"
            await send_alert(subject, body)
            print(f"警告を送信しました: {body}")
        else:
            print(f"リソース正常: CPU {cpu}%, メモリ {memory}%")
        await asyncio.sleep(300)  # 5分ごとに確認

if __name__ == "__main__":
    print("システムリソース監視を開始します。Ctrl+Cで終了します。")
    try:
        asyncio.run(monitor_resources())
    except KeyboardInterrupt:
        print("プログラムを終了します。")

import asyncio
import aiomysql
import aioshutil
from datetime import datetime

async def create_db_backup():
    # データベース接続情報
    db_config = {
        'host': 'localhost',
        'user': 'your_username',
        'password': 'your_password',
        'db': 'your_database'
    }

    # バックアップファイル名
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    backup_file = f"backup_{timestamp}.sql"

    try:
        # データベースに接続
        connection = await aiomysql.connect(**db_config)

        # バックアップコマンドを実行
        cmd = f"mysqldump -u {db_config['user']} -p{db_config['password']} {db_config['db']} > {backup_file}"
        proc = await asyncio.create_subprocess_shell(cmd)
        await proc.communicate()

        print(f"バックアップを作成しました: {backup_file}")

        # バックアップファイルを別の場所にコピー（オプション）
        await aioshutil.copy(backup_file, "/path/to/backup/storage/")

    except Exception as e:
        print(f"バックアップ作成中にエラーが発生しました: {e}")
    finally:
        if 'connection' in locals():
            connection.close()

async def schedule_backup():
    while True:
        now = datetime.now()
        # 毎日午前2時にバックアップを実行
        if now.hour == 2 and now.minute == 0:
            await create_db_backup()
        await asyncio.sleep(60)  # 1分ごとに時間をチェック

if __name__ == "__main__":
    print("データベースバックアップスケジューラを開始します。Ctrl+Cで終了します。")
    try:
        asyncio.run(schedule_backup())
    except KeyboardInterrupt:
        print("プログラムを終了します。")

import asyncio
import aiohttp
import json
from datetime import datetime

# IoTデバイスのシミュレーション
async def simulate_iot_device(device_id):
    while True:
        # ランダムなセンサーデータを生成
        temperature = 20 + (device_id * 0.1) + (asyncio.get_event_loop().time() % 5)
        humidity = 50 + (device_id * 0.2) + (asyncio.get_event_loop().time() % 10)
        yield {
            'device_id': device_id,
            'temperature': round(temperature, 2),
            'humidity': round(humidity, 2),
            'timestamp': datetime.now().isoformat()
        }
        await asyncio.sleep(5)  # 5秒ごとにデータを生成

async def collect_data(device_id):
    async for data in simulate_iot_device(device_id):
        await process_data(data)

async def process_data(data):
    # データの簡単な分析
    if data['temperature'] > 25 or data['humidity'] > 60:
        print(f"警告: デバイス {data['device_id']} の異常値を検出")
    print(f"デバイス {data['device_id']} のデータ: 温度 {data['temperature']}°C, 湿度 {data['humidity']}%")

async def main():
    # 複数のIoTデバイスからデータを収集
    tasks = [collect_data(i) for i in range(5)]  # 5つのデバイスをシミュレート
    await asyncio.gather(*tasks)

if __name__ == "__main__":
    print("IoTデータ収集を開始します。Ctrl+Cで終了します。")
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("プログラムを終了します。")

import logging

logging.basicConfig(filename='error.log', level=logging.ERROR)

def periodic_task():
    try:
        # タスクの処理
        pass
    except Exception as e:
        logging.error(f"タスク実行中にエラーが発生しました: {str(e)}")
        # 必要に応じて再試行やエラー通知を行う

import gc

def memory_intensive_task():
    # メモリを大量に使用する処理
    pass

def periodic_task():
    memory_intensive_task()
    gc.collect()  # 明示的にガベージコレクションを呼び出す

import time

def cpu_intensive_task():
    # CPUを大量に使用する処理
    pass

def periodic_task():
    start_time = time.time()
    cpu_intensive_task()
    elapsed_time = time.time() - start_time
    if elapsed_time < 60:  # タスクが1分以内に完了した場合
        time.sleep(60 - elapsed_time)  # 残りの時間をスリープ

import re

def validate_input(data):
    if not re.match(r'^[a-zA-Z0-9_]+$', data):
        raise ValueError("無効な入力データです")

def periodic_task(input_data):
    try:
        validate_input(input_data)
        # 処理を続行
    except ValueError as e:
        logging.error(f"入力バリデーションエラー: {str(e)}")

def read_sensitive_file():
    with open('sensitive_data.txt', 'r') as f:
        data = f.read()
    return data

import os

def get_api_key():
    return os.environ.get('API_KEY')

def api_request():
    api_key = get_api_key()
    # APIリクエストの処理

from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.cron import CronTrigger

def task():
    print("定期タスクを実行中")

scheduler = BackgroundScheduler()
scheduler.add_job(task, CronTrigger(hour='*/1'))  # 毎時0分に実行
scheduler.start()

import weakref

class Parent:
    def __init__(self):
        self.child = Child(self)

class Child:
    def __init__(self, parent):
        self.parent = weakref.ref(parent)  # 弱参照を使用

# 使用例
parent = Parent()

def process_file():
    with open('data.txt', 'r') as f:
        # ファイル処理
        pass
    # ファイルは自動的にクローズされる

import json

def process_large_data(data):
    # 大量のデータを処理
    result = some_heavy_processing(data)

    # 結果をファイルに書き出し
    with open('result.json', 'w') as f:
        json.dump(result, f)

    # メモリから解放
    del result

import threading

class SharedResource:
    def __init__(self):
        self.value = 0
        self.lock = threading.Lock()

    def increment(self):
        with self.lock:
            self.value += 1

# 使用例
shared = SharedResource()
def worker():
    for _ in range(1000000):
        shared.increment()

threads = [threading.Thread(target=worker) for _ in range(10)]
for t in threads:
    t.start()
for t in threads:
    t.join()

print(shared.value)  # 10000000

from concurrent.futures import ThreadPoolExecutor

def task(n):
    return n * n

with ThreadPoolExecutor(max_workers=5) as executor:
    results = executor.map(task, range(10))
    for result in results:
        print(result)

import schedule
import time
from sklearn.model_selection import train_test_split
from sklearn.ensemble import RandomForestClassifier
import pandas as pd

def load_data():
    # データの読み込み（実際のコードではデータベースから最新データを取得）
    return pd.read_csv('customer_data.csv')

def train_model(X, y):
    model = RandomForestClassifier()
    model.fit(X, y)
    return model

def evaluate_model(model, X_test, y_test):
    score = model.score(X_test, y_test)
    print(f"モデルの精度: {score}")

def retrain_model():
    print("モデルの再学習を開始します")
    data = load_data()
    X = data.drop('target', axis=1)
    y = data['target']
    X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2)

    model = train_model(X_train, y_train)
    evaluate_model(model, X_test, y_test)

    # モデルの保存（実際のコードではモデルをファイルやデータベースに保存）
    print("モデルを保存しました")

# 毎週月曜日の午前2時にモデルを再学習
schedule.every().monday.at("02:00").do(retrain_model)

while True:
    schedule.run_pending()
    time.sleep(1)

from celery import Celery
from celery.schedules import crontab

app = Celery('tasks', broker='redis://localhost:6379')

@app.task
def process_data():
    print("データ処理タスクを実行中")
    # 実際のデータ処理ロジックをここに記述

@app.task
def generate_report():
    print("レポート生成タスクを実行中")
    # 実際のレポート生成ロジックをここに記述

app.conf.beat_schedule = {
    'process-data-every-hour': {
        'task': 'tasks.process_data',
        'schedule': crontab(minute=0, hour='*'),
    },
    'generate-report-daily': {
        'task': 'tasks.generate_report',
        'schedule': crontab(minute=0, hour=0),
    },
}

import boto3
import json

def lambda_handler(event, context):
    s3 = boto3.client('s3')
    bucket_name = 'your-bucket-name'

    # S3バケット内のオブジェクトリストを取得
    response = s3.list_objects_v2(Bucket=bucket_name)

    for obj in response.get('Contents', []):
        # 各オブジェクトに対して処理を実行
        process_object(s3, bucket_name, obj['Key'])

    return {
        'statusCode': 200,
        'body': json.dumps('処理が完了しました')
    }

def process_object(s3, bucket, key):
    # オブジェクトの内容を取得
    response = s3.get_object(Bucket=bucket, Key=key)
    content = response['Body'].read().decode('utf-8')

    # ここでオブジェクトの処理を行う
    processed_content = content.upper()  # 例：全て大文字に変換

    # 処理結果を新しいオブジェクトとして保存
    s3.put_object(Bucket=bucket, Key=f"processed/{key}", Body=processed_content)

    print(f"オブジェクト {key} の処理が完了しました")