たまにやろうとして、あれ、何だっけ？ってなるやつ
色々散らばってるの少しずつここに集めていく予定
割と適当にコピペしてきたので動かなかったらごめんなさい

S3 - バケット直下のディレクトリ一覧

.py

import boto3
r = boto3.resource('s3').meta.client.list_objects(Bucket='バケット名', Delimiter='/')
[t['Prefix'].replace('/', '') for t in r['CommonPrefixes']]

S3 - 指定ディレクトリ以下のファイルを再帰検索

.py

import boto3
r = boto3.resource('s3').Bucket('バケット名').objects.filter(Prefix='パスprefix')
[r.key for r in r]

S3 - jsonをdictに変換

.py

import json
import boto3
json.loads(boto3.resource('s3').Object('バケット名', 'ファイルパス').get()['Body'].read())

S3 - バイナリとして取得(zipinfo抜くときとか使ってた)

.py

from io import BytesIO
import boto3
bj = boto3.resource('s3').Object(bucket_name='バケット名', key='ファイルパス')
BytesIO(bj.get()['Body'].read())

Dynamodb - 条件付き書き込み(put_item)

.py

import boto3
from botocore.exceptions import ClientError
try:
    boto3.resource('dynamodb').put_item(
        Item={
            'key1': 'val1',
            'key2': 'val2',
            'key3': 'val3',
        },
        ConditionExpression='attribute_not_exists(key1)',
    )
except ClientError as e:
    if e.response['Error']['Code'] == 'ConditionalCheckFailedException':
        return False
    else:
        raise

Dynamodb - 条件付き書き込み(update_item)

.py

import boto3
from botocore.exceptions import ClientError
try:
    boto3.resource('dynamodb').update_item(
        Key={'key1': 'val1'},
        UpdateExpression='set key3 = :u',
        ConditionExpression='key2 = :c',
        ExpressionAttributeValues={
            ':u': 'val3',
            ':c': 'val2',
        },
    )
    return True
except ClientError as e:
    if e.response['Error']['Code'] == 'ConditionalCheckFailedException':
        return False
    else:
        raise

Batch - ジョブ送信

.py

import boto3
boto3.client('batch').submit_job(
    jobName='job definition 名前',
    jobQueue='job queue ARN',
    jobDefinition='job definition ARN',
    containerOverrides={
        'command': [],
        'environment': [
            {'name': '環境変数名', 'value': '環境変数値'},
        ],
    },
)

番外 loggingでSlack飛ばすHandler

.py

import json
import logging
import requests
class SlackHandler(logging.Handler):
    def __init__(self, name, webhook_url, channel, icon_emoji=':slack:'):
        super().__init__()
        self.name = name
        self.webhook_url = webhook_url
        self.channel = channel

    def emit(self, record, color='#FF0000'):
        try:
            if self.formatter:
                v = self.formatter.format(record)
            else:
                v = record.getMessage()

            requests.post(
                self.webhook_url,
                headers={'content-type': 'application/json'},
                data=json.dumps({
                    'username': self.name,
                    'icon_emoji': self.icon_emoji,
                    'channel': self.channel,
                    'attachments': [{
                        'color': color,
                        'fields': [{
                            'value': v,
                            'short': False,
                        }],
                    }],
                })
            )
        except Exception:
            self.handleError(record)

番外コマンドライン引数いろいろ

sample.py

import argparse
parser = argparse.ArgumentParser()

# 位置引数
parser.add_argument('position1', help='position1 required')
parser.add_argument('position2', help='position2 required')

# optional引数
parser.add_argument('--optional1', help="optional1")
parser.add_argument('--optional2', help="optional2", action='store_true')

args = parser.parse_args()
print(args.position1)
print(args.position2)
print(args.optional1)
print(args.optional2)

実行サンプル

.sh

$ python sample.py
usage: sample.py [-h] [--optional1 OPTIONAL1] [--optional2]
                 position1 position2
sample.py: error: the following arguments are required: position1, position2
$
$ python sample.py -h
usage: sample.py [-h] [--optional1 OPTIONAL1] [--optional2]
                 position1 position2

positional arguments:
  position1             position1 required
  position2             position2 required

optional arguments:
  -h, --help            show this help message and exit
  --optional1 OPTIONAL1
                        optional1
  --optional2           optional2
$
$ python sample.py AAA BBB
AAA
BBB
None
False
$
$ python sample.py AAA BBB --optional1 CCC --optional2 
AAA
BBB
CCC
True