可変長引数
引数の数が決まっていない引数。汎用的なメソッドを作成したい場合に登場する。
*args (*1個。タプルで受け取る。)
**kwargs (*2個。辞書で受け取る。)
String型をlistに
input = "[1,2,3]"を
input = [1,2,3]に変えたい
import ast
def string_to_list(string):
return ast.literal_eval(string)
冒頭のuやr
r"文字列" ... raw文字列 / raw string literal
プログラムの中に突然現れる数字
悪い例 : return $price * 0.08;
良い例:return $price * TAX_RATE;
PEP8
・ファイル名
ファイル名は英小文字の短い名前
読みやすさのためにアンダースコアで単語を区切ってもよい
(例)s3_read.py
・クラス名
最初大文字 + 大文字区切り
(例)MyFavoriteClass
・メソッド名
全小文字 + アンダースコア区切り
boto3の認証情報
優先順位順が高い順
>>> import boto3
>>> client = boto3.client('iam',
aws_
access_key_id='YOURACCESSKEY',
aws_secret_
access_key='YOURSECRETKEY')
>>> client.list_users()
>>> import boto3
>>> session = boto3.session.Session(
aws_
access_key_id='YOURACCESSKEY',
aws_secret_
access_key='YOURSECRETKEY')
>>> client = session.client('iam')
>>> client.list_users()
プロファイル指定
>>> import boto3
>>> session = boto3.session.Session(profile_name='YOURPROFILENAME')
>>> client = session.client('iam')
>>> client.list_users()
$ python3
Python 3.6.9 (default, Nov 7 2019, 10:44:02)
Type "help", "copyright", "credits" or "license" for more information.
>>> import boto3
>>> client = boto3.client('iam')
>>> client.list_users()
ここまでの認証情報がない場合、認証情報ファイル(~/.
aws/credentials)内に default プロファイルとして構成されている認証情報が使われる。
ここまでの認証情報がない場合、
AWS設定ファイル(~/.
aws/config)内に default プロファイルとして構成されている認証情報が使われる。
現在の日付取得
datetime.datetime.strftime(current_datetime(), '%Y%m%d%H%M%S')
抽象(基底)クラス
別名:abstract base class: ABC
抽象基底クラスを実装するための標準ライブラリとして、abcモジュールが提供されている。abstract=抽象。
抽象クラスでは、メソッドのインターフェースのみ規定し、その実装はサブクラス(子クラス)に任せる。
特徴
例:u = Users()
・専ら他のクラスに継承されることによって使用される。
(実装例)
from abc import ABCMeta
from abc import abstractmethod
#抽象クラスvehicleの定義(抽象基底クラス、親クラス、とも。継承元としてのみ存在するクラス。)
class vehicle(metaclass = ABCMeta):
#@abstractmethodデコレーターを使用(中身は空)
@abstractmethod
def start(self):
pass
@abstractmethod
def stop(self):
pass
#vehicleを継承したクラスcarの定義
class car(vehicle):
def start(self):
print("car start.")
def stop(self):
print("car stop")
#vehicleを継承したクラスmotorcycleの定義
class motorcycle(vehicle):
def start(self):
print("moto start.")
def stop(self):
print("moto stop")
#テスト部
if __name__ == "__main__":
mycar = car()
mycar.start()
mycar.stop()
dict = {"name": "太郎", "age": 23, "gender": "男"}
enc =
json.dumps(dict, indent=2, ensure_ascii=False)
(ensure_ascii=False 指定しないと、日本語が文字化けする。)
基底クラス
テストメソッド
setUpメソッドは、各テストメソッドが呼び出される前に実行。
逆に、tearDownメソッドでは、各テストメソッドの終了後に実行されるメソッド。
オブジェクトのメモリ使用量を確認(単位:バイト)
import sys
print(sys.getsizeof("aaaaa"))
辞書を使用していて、存在しないキーにアクセスしようとすると、エラー
回避方法:defaultdictを使用する
from collections import defaultdict
my_dictonary = defaultdict(str)
my_dictonary['name'] = "Name"
my_dictonary['surname'] = "Surname"
print(my_dictonary["age"])
os.environ['AAAAAAAAAAA'] = 'aaa'
os.environ.get('AAAAAAAAAAA')
os.chdir(f'{os.path.dirname(__file__)}/../')
HOMEDIR = os.getcwd()
os.environ['HOMEDIR'] = HOMEDIR
# テスト実行用にカレント
ディレクトリをsrc配下に移動する
os.chdir(f'{HOMEDIR}/src')
# テスト実行用にsys,pathを通す
sys.path.append(f"{HOMEDIR}/src") # nopep8 # テスト実行時にsrc側にパスを通す
sys.path.append(f"{HOMEDIR}/tests") # nopep8 # テスト実行時にtests側にパスを通す
S3操作
from boto3_type_annotations.s3 import Client as s3Client
from botocore.config import Config
import boto3
session = boto3.session.Session(profile_name='localstack')
s3client: s3Client = session.client('s3', endpoint_url=localstack_url, config=Config(signature_version='s3v4'))
def count_s3_keys(
bucket: str, prefix: str):
"""
キーが指定したプレフィクスであるオブジェクトの件数を取得する
"""
keys =
res = s3client.list_objects_v2(
Prefix=prefix
)
if res and res.get('Contents'):
keys = [content.get("Key") for content in res.get("Contents")]
return len(keys)
def get_s3_keys(
bucket: str, prefix: str):
"""
キーが指定したプレフィクスであるオブジェクトを取得する
"""
keys = []
res = s3client.list_objects_v2(
Prefix=prefix
)
if res and res.get('Contents'):
keys = [content.get("Key") for content in res.get("Contents")]
return keys
辞書のループ
test = {'
apple': 'red', 'remon': 'yellow'}
for key,
value in test.items():
データクラス(3.7〜)
データ保持用のクラス
(通常)
def __init__(self, x=0, y=0, name=""):
self.x = x
self.y = y
self.name = name
(データクラス使用)
@dataclass
class Hoge2:
x: int = 0
y: int = 0
name: str = ""
使い方は一緒
hoge2 = Hoge2()
メリット
クラス定義を短くかける
データの集合とその読み書き手段のみからなるシンプルなオブジェクトで、
zip
複数のイテラブルオブジェクト(リストやタプルなど)の要素をまとめる関数
names = ['Alice', 'Bob', 'Charlie']
ages = [24, 50, 18]
for name, age in zip(names, ages):
print(name, age)
# Alice 24
# Bob 50
# Charlie 18
map
仕様
配列のすべての要素にアクセスし、関数を適応させる。
使い方
list(map(関数, 配列)
i = '1 2 3'
print(list(map(int, i.split(' '))))
# [1, 2, 3]
print([int(x) for x in i.split(' ')])
# [1, 2, 3]
getattr
# <用途>
# 様々な処理のリク
エスト(クラス内で定義されていないもの)をすべていったんひとつのオブジェクトに集めて、
# そのオブジェクトから実際の処理を他の処理に委託する際に使用。こういった仕組みはdelegationデリゲーションという。
class MyCls(object):
def __init__(self):
def __getattr__(self,name):
print("__getattribute__:name=",name,"は定義されてないよ")
my_inst=MyCls()
# 定義されている
print(my_inst.x)
# 定義されていない
print(my_inst.y)
デコレータ
# 既存関数の中身を書き換えずに装飾する為の仕組み
# 呼び出される既存関数もデコレートされる前提で作成されている必要がある?
# ここで言う既存関数とはwrapperのこと
def decodeco(func):
def wrapper(*args,**kwargs):
print('======start=======')
func(*args,**kwargs)
print('======end=======')
return wrapper
# 次の行の関数(test())をデコレートしますよ
@decodeco
def test():
print('Hello Decporator')
test()
2進数作成
#15桁くらいが限界
FLG_CNT = 3
# 2のFLG_CNT乗
i = (2**FLG_CNT)
#10進数
print(i)
#10進数->2進数
i2 =int(format(i, 'b'))
lst = [format(l, 'b').zfill(FLG_CNT) for l in range(1,i)]
lst
指定箇所だけ置換
a='1234567890ABCDEFGHIJKLMN'
i = 5
print(a[0:i],'@',a[i:len(a)].lower())
pendulum
import pendulum
_date = pendulum.datetime(2020,7,3,13)
ISODate
import pendulum
import datetime
def convert_to_ISODate(dt:str) -> datetime:
dt_p = pendulum.parse(dt)
dt_ymd = dt_p.strftime('%Y-%m-%d')
dt_hm = dt_p.strftime('%H:%M:%S')
iso_date = datetime.datetime.strptime(f"{dt_ymd}T{dt_hm}Z", "%Y-%m-%dT%H:%M:%SZ") - datetime.timedelta(hours=9)
return iso_date
os.chdir(f'{os.path.dirname(__file__)}/../')
辞書一括置換
dct = [{'title':'報道1','onair_time_begin':'2020-06-25 19:00:00+00','onair_time_end':'2020-06-25 19:30:00+00','plan_id':'1,2,3'},
{'title':'報道2','onair_time_begin':'2020-06-25 20:00:00+00','onair_time_end':'2020-06-25 19:30:00+00','plan_id':'4,5,6'},
{'title':'報道3','onair_time_begin':'2020-06-25 21:00:00+00','onair_time_end':'2020-06-25 19:30:00+00','plan_id':'7,8,9'},
{'title':'報道4','onair_time_begin':'2020-06-25 22:00:00+00','onair_time_end':'2020-06-25 19:30:00+00','plan_id':'10,11,12'},
{'title':'報道5','onair_time_begin':'2020-06-25 23:00:00+00','onair_time_end':'2020-06-25 19:30:00+00','plan_id':'13,14,15'},
{'title':'報道6','onair_time_begin':'2020-06-26 00:00:00+00','onair_time_end':'2020-06-25 19:30:00+00','plan_id':'16,17,18'}
]
a = [(d['title'].replace('報道','AAA')) for d in dct]
print(a)
時間丸め
import numpy as np
import pendulum
start_time = '2020-01-02 02:25:00'
# 分単位丸めロジック
# 数字の部分は丸めたい分数を指定
dt = pendulum.parse(start_time)
print(dt.weekday())
dt = dt.subtract(minutes=(dt.minute - int(np.floor(dt.minute / 10) * 10))).strftime('%H%M')
print(dt)
# 数字の部分は丸めたい分数を指定
dt_end = pendulum.parse(start_time)
dt_end = dt_end.add(minutes=(30 - (dt_end.minute - int(np.floor(dt_end.minute / 30) * 30))))
print(dt_end)
GET
【呼び出され側】
import requests
from pprint import pformat
def fetch_
json(method: str, url: str, **kwgs) -> Dict[str, Any]:
""""""
return fetch(method, url, **kwgs).
json()
def fetch(method: str, url: str, **kwgs) -> requests.models.Response:
""""""
logger.info(f'Request: [{method.upper()}] {url}\n{pformat(kwgs)}')
m = getattr(requests, method.lower())
r = m(url, **kwgs)
logger.info(f'Response: {r.status_code} {r.reason}')
if r.status_code != requests.codes.ok:
try:
logger.info(f'ErrorDetail:\n{pformat(r.
json())}')
except Exception:
pass
raise HttpStatusERROR(f'{r} {r.reason}: {url}')
return r
-------------------------------------
【呼び出し側】
get_data = fetch_
json('GET', url)
POST
import requests
url = 'xxxxxxxxxxxxxxxxxx'
headers = {'Content-Type': 'application/
json; charset=
utf-8'}
data = [{"xxxx":"11111"},{"yyyy":"22222"}]
r = requests.post(url, headers=headers, data=
json.dumps(data))
#codes.createdやcodes.okなど
if r.status_code != requests.codes.created:
raise Exception(f'APIError: {r} {r.reason}')
パラメータストア取得
import boto3
# boto3の認証情報の渡し方メモ
def ssm_parameter(name: str, region_name: str) -> Dict[str, str] or str:
"""
パラメータストア 想定値(SecureString)
JSON文字列を想定。変換に失敗した場合はそのまま返却
```
"region_name": "ap-northeast-1"}
```
"""
ssm = boto3.client('ssm', region_name=region_name)
r = ssm.get_parameter(Name=name, WithDecryption=True)
v = r['Parameter']['
Value']
try:
except
json.decoder.JSONDecodeError:
pass
return v
実行時引数受け取り
######################################################################
######################################################################
import argparse
# ==========================================================
# 実行コマンド例
#
python etc/command_para.py --path=s3:xxxxx --fname=sample.txt
#
python etc/command_para.py --path s3:xxxxx --fname sample.txt
# ==========================================================
# ==========================================================
# コマンド引数の解析
# ==========================================================
parser = argparse.ArgumentParser()
parser.add_argument('--path', help='参照
ディレクトリ', required=True)
parser.add_argument('--fname', help='ファイル名', required=True)
# これでもOK(ハイフンあり)
# parser.add_argument('--path', help='参照
ディレクトリ', required=True)
# parser.add_argument('--fname', help='ファイル名', required=True)
parsed, extra = parser.parse_known_args()
_path = parsed.path
_fname = parsed.fname
print("パス:",_path)
print("ファイル名:",_fname)
HTTPヘッダ取得
import requests
try:
res = requests.get(url, stream=True, headers={})
print('
au',res.headers['Content-Length'])
except Exception as e:
pass
try:
res = requests.get(url, stream=True, headers={})
print('sb',res.headers['Content-Length'])
except Exception as e:
pass
try:
res = requests.get(url, stream=True, headers={})
print('
docomo',res.headers['Content-Length'])
except Exception as e:
pass
from os import path
sql_path = f"{path.dirname
*1}/test-sql/test_正常系_setup.
sql"
path = "1111/22222/3333/5555.txt"
↓
ファイル名だけにする
filename = re.sub(".*/", '',path) か
filename = re.sub("^.*/", '',path)
↓
print(filename)
>5555.txt
pytest
時間操作
JST_TIMEZONE = timezone(timedelta(hours=+9))
current_dt = current_datetime()
comp_dt = os.environ['COMP_DATETIME']
deadline_dt = datetime.datetime.strptime(comp_dt, '%Y-%m-%d %H:%M:%S')
deadline_dt = deadline_dt.replace(year=current_dt.year, month=current_dt.month, day=current_dt.day,
hour=4, minute=30, second=0, microsecond=0)
project
└src
└app
└batch
└user
└job
└・・・
└tests
└test_function1
└テスト用.py
└common.py(中身はdefだけ)
FastAPI
Python製の非同期前提なWeb APIをいい感じに作るためのFramework。flaskに近い。関連用語「RESTful API」
REST=セッション等の状態管理はせず、やり取りされる情報はそれ自体で完結。
Applicationとしてのエントリーポイントはapp.py
$ python app.py
(例)
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
def read_root():
return {"Hello": "World"}
@app.get("/items/{item_id}")
def read_item(item_id: int, q: str = None):
return {"item_id": item_id, "q": q}
実行時引数を文字列にして返却
(ex. arg1=a,arg2=b,arg3=c)
#vars:辞書属性を返却
arg_dict = vars(args)
items = list(map(lambda x: f'{x[0]}={(x[1] or "")}', list(arg_dict.items())))
return ','.join(items)
import
csvwriter =
csv.writer(
csv_file, delimiter=',', quoting=
csv.QUOTE_ALL, lineterminator='\r\n')
writer.writerow(['HEADER1', 'HEADER2', 'HEADER3', 'HEADER4'])
yobi_id_
json =
json.load(open('resources/yobi_id.
json', 'r'))
traceback
標準ライブラリ。エラー時の詳細情報が出力される。以下の2通りの書き方がある。
・traceback.print_exc()
・traceback.print_exception()
改行/空白の除去
import re
query = re.sub(r"(\n)", " ", query)
query = re.sub(r"( +)", " ", query)
終了処理
メソッドの最後で
# 正常終了
sys.exit(0)
# 異常終了
sys.exit(9)
8桁の数字であること
re.match(r'^\d{8}$', self.day)
半角英数字6桁(ではないこと)
re.match(r"^[0-9a-zA-Z]{1,6}$", text) is None
半角数字0埋め4桁(ではないこと)
re.match(r"^[0-9]{4}$", text) is None
0か1(ではないこと)
re.match(r"^[01]$", args.
kbn) is None
その他
実行ファイルのパス
__file__
使用例
os.path.basename(__file__)
os.path.dirname(__file__)
ファイル名を取得: os.path.basename()
フォルダ名を取得: base_path = os.path.dirname()
ファイル名とフォルダ名のペアを取得: os.path.split()