cloudwatch logs から特定のログストリーム全取得。 https://qiita.com/ozzy3/items/fd79d07f42215298e38d を改変

get_cw_logs.py

"""
USAGE:
     poetry init -n
     poetry add boto3
     poetry run python get_cw_logs.py -p <profile> <log_group> <log_stream>
"""
from __future__ import annotations

import argparse
import re
from datetime import datetime
from typing import Any, TYPE_CHECKING

import boto3

if TYPE_CHECKING:
    from collections.abc import Iterator

def parse_args() -> argparse.Namespace:
    arg_parser = argparse.ArgumentParser()
    arg_parser.add_argument(
        '-p',
        '--profile',
        metavar='PROFILE NAME',
        help='プロファイル名',
        required=True,
    )
    arg_parser.add_argument(
       'log_group',
       metavar='LOG_GROUP',
       help='ロググループ名を指定する',
    )
    arg_parser.add_argument(
        'log_stream',
        metavar='LOG_STREAM',
        help='ログストリーム名を指定する',
    )
    return arg_parser.parse_args()

def get_log_events(
    client: Any, # botocore.client.CloudWatchLogs
    log_group_name: str,
    log_stream_name: str
) -> Iterator[dict]:
    response = client.get_log_events(
        logGroupName=log_group_name,
        logStreamName=log_stream_name,
        startFromHead=True
    )
    print(f'Count of events: {len(response['events'])}')
    print(f'nextForwardToken: {response['nextForwardToken']}')

    yield response['events']

    prev_token = None
    while response['nextForwardToken'] != prev_token:
        # nextForwardTokenを取得
        prev_token = response['nextForwardToken']

        # 2回目以降のリクエスト
        response = client.get_log_events(
            logGroupName=log_group_name,
            logStreamName=log_stream_name,
            nextToken=prev_token
        )

        len_events = len(response['events'])

        print(f'Count of events: {len_events}')
        print(f'nextForwardToken: {response['nextForwardToken']}')
        if len_events == 0:
            break

        # 取得したイベントを返す
        yield response['events']

def main() -> None:
    args = parse_args()

    sanitized_log_stream = re.sub('[\\/:*?"<>|]', '_', args.log_stream)
    output_filename = f'aws_logs_{sanitized_log_stream}.txt'

    session = boto3.session.Session(profile_name=args.profile)
    client = session.client('logs')

    with open(output_filename, 'w', encoding='UTF-8') as f:
        for events in get_log_events(client, args.log_group, args.log_stream):
            for event in events:
                timestamp = datetime.fromtimestamp(
                    event.get('timestamp') / 1000
                ).isoformat()
                message = event.get('message')
                f.write(f'{timestamp}\t{message}\n')

if __name__ == '__main__':
    main()