(イディオム)Python

python仮想環境作成(venv)

https://qiita.com/sychocola1/items/93c4e043e4c8dbc60cad

新しい環境の作成
$ cd [project dir]
$ python3 -m venv [newenvname]

Activate
(Mac)
$ source [newenvname]/bin/activate

(Windows)
$ .\[newenvname]\Scripts\activate

Deactivate
([newenvname])$ deactivate

 

可変長引数

引数の数が決まっていない引数。汎用的なメソッドを作成したい場合に登場する。

*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

u"文字列" ... unicode文字列 / unicode 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()

 

環境変数

$ export AWS_ACCESS_KEY_ID=YOURACCESSKEY

$ export AWS_SECRET_ACCESS_KEY=YOURSECRETKEY

$ python3

Python 3.6.9 (default, Nov 7 2019, 10:44:02)

[GCC 8.3.0] on linux

Type "help", "copyright", "credits" or "license" for more information.

>>> import boto3

>>> client = boto3.client('iam')

>>> client.list_users()

 

ここまでの認証情報がない場合、認証情報ファイル（~/.aws/credentials）内に default プロファイルとして構成されている認証情報が使われる。

(aws configure コマンドで設定)

 

ここまでの認証情報がない場合、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()

 

JSON出力

辞書をjson出力

import json

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')

 

ディレクトリ操作(Windows用？　Linuxは「HOMEDIR」ではなく、おそらく「PATH」)

# このファイルの一つ上のディレクトリをホームディレクトリってことにする

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(

Bucket=bucket,

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(

Bucket=bucket,

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():

print('key:', key, 'value:', value, ',', end=' ')

 

データクラス(3.7〜)

データ保持用のクラス

 

(通常)

class Hoge:

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 = ""

 

使い方は一緒

hoge = Hoge()

hoge2 = Hoge2()

 

メリット

クラス定義を短くかける

 

DTO

データの集合とその読み書き手段のみからなるシンプルなオブジェクトで、

そのデータを利用して行われる具体的な処理（ビジネスロジックなど）は含まない。

 

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デリゲーションという。

# https://agongji.hatenadiary.jp/entry/2020/02/25/224632

 

class MyCls(object):

def __init__(self):

self.x="x_value"

 

def __getattr__(self,name):

print("__getattribute__:name=",name,"は定義されてないよ")

return "undefine_value"

 

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

 

ディレクトリ移動

(参考)https://note.nkmk.me/python-os-getcwd-chdir/

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

-------------------------------------

【呼び出し側】

import fetch_json

 

get_data = fetch_json('GET', url)

 

POST

import requests

import json

 

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の認証情報の渡し方メモ

# https://cloudpack.media/24423

 

def ssm_parameter(name: str, region_name: str) -> Dict[str, str] or str:

"""

パラメータストア 想定値(SecureString)

JSON文字列を想定。変換に失敗した場合はそのまま返却

 

```

{"aws_access_key_id": "XXXXXX",

"aws_secret_access_key": "XXXXX",

"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:

v = json.loads(v)

except json.decoder.JSONDecodeError:

pass

return v

 

実行時引数受け取り

######################################################################

# Pythonコマンド 引数受け取り

######################################################################

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:

url = 'https://www.au.com/ezfactory/tec/spec/ezsava_ip.html'

res = requests.get(url, stream=True, headers={})

print('au',res.headers['Content-Length'])

except Exception as e:

pass

 

try:

url = 'https://www.support.softbankmobile.co.jp/partner/home_tech1/index.cfm'

res = requests.get(url, stream=True, headers={})

print('sb',res.headers['Content-Length'])

except Exception as e:

pass

 

try:

url = 'https://www.nttdocomo.co.jp/service/developer/smart_phone/spmode/index.html'

res = requests.get(url, stream=True, headers={})

print('docomo',res.headers['Content-Length'])

except Exception as e:

pass

 

同じディレクトリにあるSQLファイルのパスを取得

from os import path

sql_path = f"{path.dirname*1}/test-sql/test_正常系_setup.sql"

 

正規表現

https://www.debuggex.com/

 

path = "1111/22222/3333/5555.txt"

↓

ファイル名だけにする

filename = re.sub(".*/", '',path) か

filename = re.sub("^.*/", '',path)

↓

print(filename)

>5555.txt

 

 

pytest

python -m pytest tests

 

時間操作

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)

 

CSV出力(ヘッダ)

import csv
writer = csv.writer(csv_file, delimiter=',', quoting=csv.QUOTE_ALL, lineterminator='\r\n')
writer.writerow(['HEADER1', 'HEADER2', 'HEADER3', 'HEADER4'])

 

json読み込み

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() 

*1:__file__