Python たまに使うやつメモ(boto3とその他)
たまにやろうとして、あれ、何だっけ?ってなるやつ
色々散らばってるの少しずつここに集めていく予定
割と適当にコピペしてきたので動かなかったらごめんなさい
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
Author And Source
この問題について(Python たまに使うやつメモ(boto3とその他)), 我々は、より多くの情報をここで見つけました https://qiita.com/KosukeJin/items/0ca31b14707df9acb0f3著者帰属:元の著者の情報は、元のURLに含まれています。著作権は原作者に属する。
Content is automatically searched and collected through network algorithms . If there is a violation . Please contact us . We will adjust (correct author information ,or delete content ) as soon as possible .