目的の話

最近のシステムは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/