LoginSignup
6

More than 5 years have passed since last update.

Python たまに使うやつメモ(boto3とその他)

Last updated at Posted at 2018-12-05

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

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

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

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

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

S3 - jsonをdictに変換

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

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

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

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

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)

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 - ジョブ送信

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

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)

実行サンプル

$ 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

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
6