目的の話
最近のシステムはRestAPIを使用して、作成されている。
多くの場合はWEB-UIを使用してこのサービスを使用するが、下記のようにそれだけでは不都合なシーンがある。
- 定型作業をプログラムのように実行したい
あるサービスに100人のユーザーを登録するという作業をWEB-UIを用いる場合、入力は手入力となり時間を要する。
健全で怠惰なプログラマならCSVから登録ユーザの一覧を読み込み、プログラムで登録したいと考えるはずだ。 - 特定の作業を、操作手順ではなくコマンドの形式で表現したい
操作手順をドキュメントに記録する際も、文章やスクリーンショットで説明するよりも、下記のようなコマンドで説明するほうが、エンジニアのやり取りでは誤解が少ない。
$ mycli workspace list | jq -c '.[] | {name:.name, id:.id}'
{"name":"TEST_WS_1","id":"AAAAA"}
{"name":"TEST_WS_2","id":"AAAAB"}
AWSの場合AWSCLI、Azureの場合AzureCliというように多くの場合、こうした用途のためにコマンドラインインターフェイスが提供されている。
本記事では、pythonのClickというライブラリを使用して、独自のサービスにコマンドラインインターフェイスを作成する方法について紹介する。
ディレクトリ構造
.
├── README.md
├── mycli
│ ├── __init__.py
│ ├── __main__.py
│ ├── mycli.py
│ ├── webclient
│ │ ├── __init__.py
│ │ └── webclinet.py
│ └── modules
│ ├── __init__.py
│ ├── login.py
│ ├── user.py
│ └── workspace.py
├── requirements.txt
├── setup.cfg
└── setup.py
実装の話
今回コマンドラインインターフェスを作成するために、PythonのライブラリClickを使用した。
https://click.palletsprojects.com/en/8.1.x/
コマンドの実装
コマンドはClickの機能を利用して下記のように実装できる。
mycli/modules/user.py
import click
import json
from click.core import Context
from webcli.webclient import WebClient
@click.group
@click.pass_context
def user(ctx: Context):
"""User Operations."""
@user.command
@click.pass_context
def list(ctx: Context):
"""List Users."""
web_client: WebClient = ctx.obj['web_client']
result = web_client.list_user()
print(json.dumps(result, indent=4))
実際の処理はクラスWebClientで実装されている。
mycli/webclient/webclinet.py
class WebClient():
def list_user(self) -> dict:
res = self.get(self.__urljoin('/v1/users/users'))
return json.loads(res.text)
def get(self, path: str, params: dict = {}, cookies: dict = {}) -> Response:
res = requests.get(
self.__urljoin(path),
params=params,
cookies=cookies,
headers=self.__create_headers()
)
return res
共通処理の実装はclickのroot-commandであるcliで実装される。これにpass_contextを付与することで実現できる。
https://click.palletsprojects.com/en/8.1.x/complex/#the-root-command
今回の実装ではload_configという機能を使用して、接続先の設定やワークスペースと呼ばれる作業範囲をデフォルト値をロードする機能が共通処理として実装されている。
これらの設定は下記のように、Clickが提供するContext上に保存される。WebClientは後述で詳しく説明するが、認証処理や認証情報を読み込みがここで行われる。
ctx.obj['default_workspace'] = config.default_workspace
ctx.obj['web_client'] = WebClient(config.web_service_host)
mycli/mycli.py
import json
import os
import os.path
import click
from click.core import Context
from dotenv import load_dotenv
from mycli.webclient import WebClient
from mycli.modules import (login, user)
DEFAULT_CONFIG_FILEPATH = os.path.expanduser('~/.mycli/config.json')
DEFAULT_WEB_SERVICE_HOST = "https://localhost"
@click.group()
@click.option('--service_host', '-h', 'host_name')
@click.pass_context
def cli(ctx: Context, host_name: str):
ctx.obj = dict()
try:
config = load_config()
except (FileNotFoundError, json.JSONDecodeError) as e:
config = Config()
if config.default_workspace:
ctx.obj['default_workspace'] = config.default_workspace
ctx.obj['web_client'] = WebClient(config.web_service_host)
load_dotenv()
cli.add_command(login.login)
cli.add_command(user.user)
class Config:
def __init__(self, web_service_host: str = DEFAULT_WEB_SERVICE_HOST, default_workspace: str = None, refresh_token: str = None):
self.web_service_host = web_service_host
self.default_workspace = default_workspace
def load_config(config_filepath: str = DEFAULT_CONFIG_FILEPATH) -> Config:
with open(config_filepath, 'rb') as config_file:
config_json = json.load(config_file)
return Config(
web_service_host=config_json['web_service_host'],
default_workspace=config_json['default_workspace']
)
def save_config(config: Config, config_filepath: str = DEFAULT_CONFIG_FILEPATH) -> Config:
os.makedirs(os.path.dirname(DEFAULT_CONFIG_FILEPATH), exist_ok=True)
with open(config_filepath, 'w+') as config_file:
config_rawjson = json.dumps(config.__dict__, indent=4)
config_file.write(config_rawjson)
return config
ログイン処理の実装
今回のシステムではアクセストークンとリフレッシュトークンを使用している。
jwt形式であり、今回はライブラリpyjwtを使用する。
setup.cfg
pyjwt[crypto]==2.6.0
下記のように実装することで、accesstokenの有効期限切れを検出することができる。
mycli/webclient/webclinet.py
class Credentials:
def __init__(self, access_token: str = None, refresh_token: str = None):
self.access_token = access_token
self.refresh_token = refresh_token
def is_expired_accesstoken(self) -> bool:
try:
jwt.decode(self.access_token, options={
'verify_signature': False, 'verify_exp': True})
except jwt.ExpiredSignatureError as e:
return True
else:
return False
アクセストークンはユーザーがログインコマンドを実行することで発行される。
clickを使用することで、コマンドは下記のように実装できる。
mycli/modules/user.py
@click.command
@click.option('--mailaddress', '-m', 'mail_address', required=True, prompt=True)
@click.option('--password', '-p', 'password', prompt=True, hide_input=True)
@click.pass_context
def login(ctx: Context, mail_address: str, password: str):
"""login Operations."""
web_client: WebClient = ctx.obj['web_client']
result = web_client.login(mail_address, password)
credentials = Credentials(
access_token=result['accessToken'], refresh_token=result['refresh'])
WebClient.save_credentials(credentials)
mycli/webclient/webclinet.py
class WebClient():
def __urljoin(self, path) -> str:
return urljoin(self.__web_service_host, path)
def __create_headers(self) -> str:
return {'Content-Type': 'application/json', 'Authorization': f'Bearer {self.__access_token}', 'correlation-id': 'DEV_CORRELATION_ID'}
def login(self, email: str, password: str) -> dict:
body = json.dumps({'email': email, 'password': password})
response = requests.post(
self.__urljoin('/v1/users/login'),
data=body,
headers=self.__create_headers_without_token()
)
result = json.loads(response.text)
result['refresh'] = response.cookies.get('refresh')
return result
@classmethod
def save_credentials(cls, credentials: Credentials, credentials_filepath: str = DEFAULT_CREDENTIALS_FILEPATH) -> Credentials:
os.makedirs(os.path.dirname(DEFAULT_CREDENTIALS_FILEPATH), exist_ok=True)
with open(credentials_filepath, 'w+') as credentials_file:
credentials_rawjson = json.dumps(credentials.__dict__, indent=4)
credentials_file.write(credentials_rawjson)
return credentials
ログインをコマンドを実行するたびに行うのは非効率ですし、誰もそんなことはしたくないはずです。
なのでWebClient.save_credentials(credentials)において保存されます。
保存箇所はブラウザクッキーに保存するときと同様に、慎重に扱う必要があります。
今回ははユーザーのホームディレクトリ上の、ドットファイルのように保存する手段を選択しました。
pythonの場合下記のように簡単に実装できます。
DEFAULT_CREDENTIALS_FILEPATH = os.path.expanduser('~/.mycli/credentials.json')
アクセストークンは有効期間が短く、アクセストークンの有効期間が切れた際はリフレッシュトークンを使用して再取得する。
この処理は、WebClientクラスのコンストラクタで行われる。
class WebClient():
def __init__(self, web_service_host: str, credentials: Credentials = None):
self.__web_service_host = web_service_host
if not credentials:
try:
credentials = WebClient.load_credentials()
except (FileNotFoundError, json.JSONDecodeError) as e:
pass
if credentials and credentials.is_expired_accesstoken() and credentials.refresh_token:
self.__access_token = credentials.access_token
self.__refresh_token = credentials.refresh_token
try:
res = self.refrash_token()
credentials = Credentials(
access_token=res['accessToken'], refresh_token=self.__refresh_token)
WebClient.save_credentials(credentials)
except requests.HTTPError as e:
pass
if credentials:
self.__access_token = credentials.access_token
self.__refresh_token = credentials.refresh_token
@classmethod
def load_credentials(cls, credentials_filepath: str = DEFAULT_CREDENTIALS_FILEPATH) -> Credentials:
with open(credentials_filepath, 'rb') as credentials_file:
credentials_json = json.load(credentials_file)
return Credentials(
access_token=credentials_json['access_token'],
refresh_token=credentials_json['refresh_token']
)
@classmethod
def save_credentials(cls, credentials: Credentials, credentials_filepath: str = DEFAULT_CREDENTIALS_FILEPATH) -> Credentials:
os.makedirs(os.path.dirname(DEFAULT_CREDENTIALS_FILEPATH), exist_ok=True)
with open(credentials_filepath, 'w+') as credentials_file:
credentials_rawjson = json.dumps(credentials.__dict__, indent=4)
credentials_file.write(credentials_rawjson)
return credentials
def __urljoin(self, path) -> str:
return urljoin(self.__web_service_host, path)
def __create_headers(self) -> str:
return {'Content-Type': 'application/json', 'Authorization': f'Bearer {self.__access_token}', 'correlation-id': 'DEV_CORRELATION_ID'}
def __create_headers_without_token(self) -> str:
return {'Content-Type': 'application/json'}
def login(self, email: str, password: str) -> dict:
body = json.dumps({'email': email, 'password': password})
response = requests.post(
self.__urljoin('/v1/users/login'),
data=body,
headers=self.__create_headers_without_token()
)
result = json.loads(response.text)
result['refresh'] = response.cookies.get('refresh')
return result
def refrash_token(self) -> dict:
res = requests.put(
self.__urljoin('/v1/users/refreshToken'),
data='',
headers=self.__create_headers_without_token(),
cookies={'refresh': self.__refresh_token}
)
res.raise_for_status()
return json.loads(res.text)
setup.pyを作成してパッケージ化する
setup.pyおよびsetup.cfgを準備することで、パッケージ化して配布することができる。
setup.py
# setup.py
from setuptools import setup
setup()
setup.cfg
[metadata]
name = mycli
version = 0.1.0
classifiers =
Framework :: Click
Programming Language :: Python :: 3
Programming Language :: Python :: 3.8
[options]
packages = find:
install_requires =
requests==2.31.0
click==8.1.3
pyjwt[crypto]==2.6.0
python-dotenv==1.0.0
python_requires = >=3.8
[options.entry_points]
console_scripts =
mycli=mycli.mycli:cli
インストールと実行
pip install -e .でインストールすることでコマンドのように扱うことができる
$ git clone https://github.com/XXXX
$ cd mycli
$ pip install -e .
$ mycli login
Mail address: user_name@domain.dummy
Password: ******
参考資料
https://click.palletsprojects.com/en/8.1.x/
https://packaging.python.org/ja/latest/guides/distributing-packages-using-setuptools/
https://pyjwt.readthedocs.io/en/stable/