2020/04/30

Python Tips: Python のプロジェクトで GitHub Actions を使いたい

GitHub Actions

今回は Python のプロジェクトで GitHub Actions を導入する方法をご紹介します。実際に動くかんたんなサンプルを使って説明します。


GitHub Actions とは

最初に GitHub Actions についてかんたんに説明します。

GitHub Actions とは GitHub 公式のワークフロー自動化ツールです。

2018 年 10 月にパブリックベータ版が公開され、 2019 年の 11 月に一般公開されました。それまで GitHub で CI/CD を行うには外部のサービスを利用する必要がありましたが、 GitHub Actions が導入されたことでコード管理と CI/CD を GitHub 内で一元的に行えるようになりました。

GitHub Actions は、 GitHub で管理されているリポジトリにディレクトリ .github/workflows/ を作成してその下に自動化したい処理を記述した YAML 形式の設定ファイルを置けばすぐに利用し始めることができます。処理を実行するプラットフォームは Linux / macOS / Windows の中からひとつ(または複数)選ぶことができます。

GitHub Actions の特徴のひとつに、 GitHub や他の人が作成し GitHub 上に公開したアクションを利用できる点があります。たとえば、次の内容を設定ファイルに記述すると、 actions/checkoutactions/setup-python というアクションをかんたんに利用することができます。

name: pytest

on:
  push:

jobs:
  pytest:
    name: Run tests with pytest
    runs-on: ubuntu-latest
    steps:
      - name: Checkout
        uses: actions/checkout@v2
      - name: Set up Python 3.7
        uses: actions/setup-python@v1
        with:
          python-version: 3.7

汎用度の高いアクションは誰かがオープンソースとして公開してくれていたりもするので、必ずしもすべての手順を自分でゼロから書く必要はありません。 CI/CD の経験があまり無い方やプログラミング初心者の方にも始めやすいかと思います。


Python プロジェクトでの GitHub Actions の使い方

Python プロジェクトでの GitHub Actions の使い方を説明します。

といっても、公式のドキュメントを読めば大体のことは書かれているので、スラスラと読んで理解できる方は公式のドキュメントを読むのがいちばんです。ここでは動くサンプルを使ってさわりの部分だけ説明します。

お話の前提として、 GitHub にアカウントを持っていて Python プロジェクトのリポジトリを置いているものとします。

GitHub Actions を使い始める方法は 2 つあります。

  • ブラウザで Actions のページを開く
  • ファイル .github/workflows/xxx.yml をコミットする

ブラウザで Actions のページを開くにはリポジトリの「 Actions 」タブをクリックします。ファイル .github/workflows/xxx.yml をコミットする場合は、手元のテキストエディタとターミナルをそのまま使います。どちらの形を選んでも最終的にできることは同じです。

以下、テストとスタイルチェックを行うかんたんなサンプルを使って説明します。次のツールを使用します。

  • poetry: 依存パッケージを管理する
  • pytest: テストを実行する
  • black: コーディングスタイルのチェックを行う

最初に GitHub 上に空のリポジトリを作成して、次の内容のファイル hello.py をコミットします。

hello.py:

"""関数 `hello()` を提供する"""


def hello(name: str = "world") -> None:
    """`Hello, xxx.` のメッセージを出力する"""
    print(f"Hello, {name}.")

続いて、テストファイルを作成します。 tests/test_hello.py あたりに作るのがよいでしょうか。

tests/test_hello.py:

from hello import hello


def test_hello_default(capsys):
    hello()
    out, err = capsys.readouterr()
    assert out == "Hello, world.\n"


def test_hello_with_name(capsys):
    hello("サザエ")
    hello("カツオ")
    out, err = capsys.readouterr()
    assert out == "Hello, サザエ.\n" "Hello, カツオ.\n"

これは pytest 用に作ったテストなので pytest をインストールします。私は Python のパッケージには poetry をよく使いますがここは pip でもそれ以外のものでもかまいません。

poetry add --dev pytest

pytest が無事にインストールできたら、 setup.py を作成してからテストを実行します。

setup.py:

from setuptools import setup, find_packages

setup(name="hello-github-actions", version="1.0", packages=find_packages())
poetry run pytest tests/test_hello.py
============================= test session starts ---===========================
platform linux -- Python 3.7.4, pytest-5.4.1, py-1.8.1, pluggy-0.13.1
rootdir: /app
collected 2 items

tests/test_hello.py ..                                                   [100%]

============================== 2 passed in 0.13s ---============================

成功することが確認できました。

続いて、このテストを GitHub Actions で実行するように設定ファイルを書きます。

リポジトリのルートに .github/workflows/ci.yml というファイルを作成し以下の内容で保存します。

.github/worflows/ci.yml:

name: pytest

on:
  push:
  pull_request:

jobs:
  pytest:
    name: Run tests with pytest
    # 実行環境として `ubuntu-latest` という名前のものを選ぶ
    runs-on: ubuntu-latest
    # 複数の Python のバージョンでテストするために `strategy.matrix` を設定する
    strategy:
      matrix:
        python-version: [3.7, 3.8]
    steps:
      # リポジトリをチェックアウトする
      # See: https://github.com/actions/checkout
      - name: Checkout
        uses: actions/checkout@v2
      # Python のランタイムをセットアップする
      # バージョンは `strategy.matrix` に並べたものを指定する
      # See: https://github.com/actions/setup-python
      - name: Set up Python ${{ matrix.python-version }}
        uses: actions/setup-python@v1
        with:
          python-version: ${{ matrix.python-version }}
      # Poetry そのものをインストールする
      # See: https://github.com/dschep/install-poetry-action
      - name: Install Poetry
        uses: dschep/install-poetry-action@v1.3
      # インストールした Poetry を使って必要な Python パッケージをインストールする
      - name: Install Dependencies
        run: poetry install --no-interaction
      # pytest を実行する
      - name: Run Tests
        run: poetry run pytest tests/

メインは jobs.pytest.steps の部分です。ここに指定したアクションが上から順に実行されます。

Python プロジェクトで GitHub Actions を使う場合はおそらく最初の 2 ステップ(チェックアウトと Python ランタイムのセットアップ)はほぼ共通になると思います。

各アクションがどういうことをしているかについてはインラインのコメントを見てください。

ちなみに、ディレクトリ .github/worflows/ はそのままこの名前にする必要がありますが、 ci.yml はこの名前でないといけないわけではありません。適当にわかりやすい名前を付けるとよいと思います。

ファイルが作成できたらコミットします。 poetry を使っている場合は pyproject.tomlpoetry.lock を、 pip を使っている場合は requirements.txt も忘れずにコミットしましょう。

コミットができたら GitHub に push します。すると、リポジトリの Actions のページに yaml ファイルで設定したワークフローが追加されていてテスト用のタスクが実行されるはずです。

これがうまく行ったら、おまけで black を使ったスタイルチェックも追加します。 poetry add をして、先ほどの .github/workflows/ci.yml を編集します。

poetry add --dev black

ファイルの末尾に以下の内容を追加します( black: の行が jobs の下に来るようにします)。インデントがおかしいと正しく動かないので注意してください。

  black:
    name: Check code style with Black
    runs-on: ubuntu-latest
    steps:
      - name: Checkout
        uses: actions/checkout@v2
      - name: Set up Python 3.7
        uses: actions/setup-python@v1
        with:
          python-version: 3.7
      - name: Install Poetry
        uses: dschep/install-poetry-action@v1.3
      - name: Install Dependencies
        run: poetry install --no-interaction
      - name: Check code style with Black
        run: poetry run black --check --diff .

コードに問題がなければ、このワークフローも GitHub Actions で実行され、結果が success として報告されるはずです。

ということで、かんたんではありましたが、 Python プロジェクトでの GitHub Actions の使い方についてでした。 GitHub Actions について深く知りたい方は公式のドキュメントその他を確認していただければと思います。

今回使ったサンプルを少し整理して GitHub にあげているので、興味のある方は参考にしてみてください:

この記事を書いている時点で GitHub Actions は無料で利用できますが、金額等は時間が経つと変わると思うので、ここには載せません。 GitHub 公式の情報を確認してみてください。


参考リンク

2020/04/17

Python ライブラリ: Pony ORM

Pony ORM


今回は Python の ORM ライブラリである Pony ORM を紹介します。

動作確認に使ったバージョンは次のとおりです。

  • python 3.7.7
  • pony 0.7.12

Pony ORM とは

Pony ORM (以下「 Pony 」)は Python の ORM ライブラリです。 ORM をご存知でない方のために念のために説明すると、 ORM とはオブジェクト・リレーショナル・マッパーの略で、プログラミング言語の「オブジェクト」とリレーショナルデータベースのテーブルとの間のデータのやりとりをよきようにやってくれる機能のことです。

Pony の特徴として公式では次の 4 つが謳われています。

中でも目を引く特徴は直感的なシンタックスで、次のような Python らしいコードで SQL クエリを生成することができます。

select(c for c in Customer if sum(c.orders.total_price) > 1000)
Customer.select(lambda c: sum(c.orders.total_price) > 1000)

実際に使ってみましょう。

Pony をインストールする

pip でインストールします。

python -m pip install pony

インストールが完了したら Python のコンソールでバージョンが確認できるはずです。

import pony

print(pony.__version__)
# => 0.7.12

Pony の基本的な使い方

データベースに接続する・テーブルを定義する

Pony を使うときには、最初にデータベースオブジェクトを生成し、それを使ってテーブルに相当するクラスを定義した上で、そのクラスを使ってデータベースを操作します。

from pony import orm

DB_NAME = 'db.sqlite'

# データベースオブジェクトを生成する
db = orm.Database()
db.bind(provider='sqlite', filename=DB_NAME, create_db=True)


# データベーステーブルに相当するクラスを定義する
class Article(db.Entity):
    """記事データを格納するテーブル"""
    slug = orm.Required(str)
    title = orm.Required(str)
    body = orm.Optional(str)
    data = orm.Optional(orm.Json)

    def __str__(self):
        return 'Article(title="{}")'.format(self.title)

テーブルのカラムは主に pony.orm.Requiredpony.orm.Optional を使用して定義します。引数には対象のカラムの型を渡します。ここでの説明は割愛しますが、その他のカラム設定を行いたい場合は第 2 引数以降で指定することができます。

その他カラムの定義に使えるものとして次のクラスが用意されています。

  • PrimaryKey
  • Set
  • Discriminator

参考:

テーブルに相当するクラスの定義ができたら、続いて実際のテーブルを作成します。テーブルの作成は pony.orm.Databasegenerate_mapping() メソッドを使用します。

db.generate_mapping(create_tables=True)
テーブルにレコードを挿入する

テーブルが作成できたら、テーブルに相当するクラス( Article )を使ってレコードを登録することができます。

with orm.db_session:
    a1 = Article(slug='hello', title='こんにちは')
    a2 = Article(slug='good-bye', title='さようなら')
    orm.commit()

レコードを登録するには、データベースセッションを開いてテーブルに相当するクラスのインスタンスを生成してから、変更をコミットします。データベースセッションを開くにはコンテキストマネージャ pony.orm.db_session を使います。変更のコミットには pony.orm.commit() を使用します。コンテキストマネージャ pony.orm.db_session はコンテキスト脱出時に自動的にコミットを行ってくれるので、上の例の場合は pony.orm.commit() による明示的なコミットは不要です。

pony.orm.db_session は、上の例のようにコンテキストマネージャとして使う方法に加えて関数のデコレータとして使う方法が用意されています:

@orm.db_session
def update_related_data():
    ...

コンテキストマネージャとしてもデコレータとしても使える pony.orm.db_session ですが、上述の「変更のコミット」の他にも「例外があがったときのロールバック」「コネクションプールへのデータベースコネクションの返却」「データベースセッションキャッシュのクリア」等を自動で行ってくれる機能が備わっています。

テーブルのレコードを取得する

テーブルにレコードが登録できたらそれを取得してみましょう。おそらく Pony の最大の特徴はこのレコードの取得方法です。pony.orm.select() とジェネレータ式を使ってシンプル・直感的に SELECT クエリを生成することができます。

全レコードの全カラムを順番に取得する:

with orm.db_session:
    for article in orm.select(a for a in Article):
        print(article)
# => Article(title="こんにちは")
# => Article(title="さようなら")
クラスメソッド select() を使った方法も用意されています。
with orm.db_session:
    for article in Article.select():
        print(article)
# => Article(title="こんにちは")
# => Article(title="さようなら")

全レコードの特定のカラムだけ順番に取得する:

with orm.db_session:
    for slug, title in orm.select((a.slug, a.title) for a in Article):
        print(slug, title)
# => hello こんにちは
# => good-bye さようなら

全要素の特定のカラムを取得しリスト化する:

with orm.db_session:
    slugs = orm.select(a.slug for a in Article)[:]
    print(slugs)
# => ['hello', 'good-bye']

条件に合致する 1 レコードだけ取得する:

with orm.db_session:
    article = orm.select(a for a in Article if a.slug == 'hello').get()
    print(article)
# => Article(title="こんにちは")
with orm.db_session:
    article = Article.get(slug='hello')
    print(article)
# => Article(title="こんにちは")

get() は、対象のレコードが複数件見つかったときには MultipleObjectsFoundError という例外をあげます。

複数件マッチしうるクエリで最初の要素のみ取得したい場合は次の first() を使用します。

条件に合致する最初のレコードを取得する:

with orm.db_session:
    article = orm.select(a for a in Article if 'kkk' in a.slug).first()
    print(article)
# => Article(title="こんにちは")

first() は該当するレコードが 0 件だったときは None を返します。

ORDER BY 句を使う:

with orm.db_session:
    for article in orm.select(a for a in Article).order_by(Article.slug):
        print(article)
# => Article(title="さようなら")
# => Article(title="こんにちは")

WHERE 句を使う:

with orm.db_session:
    for article in orm.select(a for a in Article if a.title.startswith('さ')):
        print(article)
# => Article(title="さようなら")
with orm.db_session:
    for article in orm.select(a for a in Article).where(lambda a: a.title.startswith('さ')):
        print(article)
# => Article(title="さようなら")

WHERE 句はジェネレータ式の ifでも where() メソッドでも書くことができます( filter() というまた別のメソッドも用意されています)。

発行されるクエリを確認する:

with orm.db_session:
    print(orm.select(a for a in Article).order_by(Article.slug).get_sql())
# => SELECT "a"."id", "a"."slug", "a"."title", "a"."body", "a"."data"
# => FROM "Article" "a"
# => ORDER BY "a"."slug"

発行されるクエリを確認したいときは get_sql() メソッドが使えます。

レコードを操作する

データベーステーブルの各レコードに相当するインスタンスにも便利なメソッドが用意されています。

primary key を取得する:

article.get_pk()
# => 2

複数のフィールドの値をまとめてセットする:

article.set(body='sayonara sayonara', data={'message': 'onara'})

dict に変換する:

article.to_dict()
# => {'id': 2, 'slug': 'good-bye', 'title': 'さようなら', 'body': 'sayonara sayonara', 'data': {'message': 'onara'}}

これらの他にも SQL と Python にある程度馴染みがあれば直感的に使える機能が多数用意されています。

この記事を書いたときに確認したかぎりでは API は 1 ページにまとまっているので、「こんなもの無いかな」と思ったときにも探しやすいと思います。

ということで、 Python のライブラリ Pony ORM の紹介でした。実際に利用してみたいと思う方は公式のドキュメント・リポジトリを見てみてください。

参考:

2020/04/10

Python Tips: Python で FTP のダウンロードを自動化したい

Python を使って FTP でファイルのダウンロードを行う方法についてです。以前次のような記事を書きましたが、今回はそのダウンロード版です。
   

Python には FTP を使うためのそのままずばり ftplib というパッケージが同梱されており、これを使えば Python でかんたんに FTP を使ったファイル送受信が行なえます。しかし、提供されているインタフェースが低レイヤーな感じで、 FTP に詳しい人以外にはあまり直感的ではありません。特にフォルダを扱う処理はかんたんなものでも数行〜のコードを書く必要があります。 

今回はそんな ftplib を使ってファイルやフォルダをダウンロードする方法について説明します。アップロードについては上の記事に書いているので興味のある方は参考にしてください。
   

ファイルをダウンロードする

まずは単一のファイルをダウンロードする方法からです。
from ftplib import FTP_TLS


config = {
    'host': 'xx.xx.xx.xx',
    'user': 'username',
    'passwd': 'password',
}

# sample.txt ファイルをダウンロードする
with FTP_TLS(**config) as ftp:
    with open('sample.txt', 'wb') as fp:
        ftp.retrbinary('RETR sample.txt', fp.write)

ファイルをダウンロードするには ftplib.FTP_TLS (以下 FTP_TLS )のインスタンスを生成してサーバーに接続した上で retrbinary() メソッドを使います。

FTP_TLSftplib.FTP (以下 FTP )のサブクラスで TLS をサポートした FTP (いわゆる「 FTPS 」)を行うためのクラスです。いまどき FTP を使うべき理由も無いので原則 ftplib.FTP_TLS 一択と思ってよいでしょう。 FTP_TLS の初期化時には host user passwd の引数を渡します。

retrbinary() メソッドの第一引数には 'RETR ファイルパス' という形式の文字列を渡します。「ファイルパス」というのはサーバー側でのファイルパスです。パスセパレータには / が使えます。

retrbinary() の第二引数には保存先のファイルオブジェクトの write() メソッドを渡します。 retrbinary() はデータをバイナリモードで取得してくるので、保存先のファイルオブジェクトは open('ファイル名', 'wb') で開いておく必要があります。 ちなみに、 retrbinary() と似たメソッドに retrlines() というものもあり、こちらは ASCII モードでデータを取得します。対象のファイルがテキストデータであることがわかっている場合は retrlines() を使ってもよいかもしれません。
   

ファイル名に日本語を含むファイルをダウンロードする

ファイル名に日本語を含む(正確には文字コード latin-1 では正しく扱えない文字を含む)ファイルを扱う場合は注意が必要です。なぜなら FTP_TLS は( FTP も同じ)デフォルトで文字エンコーディングに latin-1 を使うからです。ファイル名に日本語を含むファイルを扱うときは適切なエンコーディングを使うように設定する必要があります。

 参考:

from ftplib import FTP_TLS


ENCODING = 'utf8'

config = {
    'host': 'xx.xx.xx.xx',
    'user': 'username',
    'passwd': 'password',
}

# サンプル.txt ファイルをダウンロードする
with FTP_TLS(**config) as ftp:
    ftp.encoding = ENCODING
    with open('サンプル.txt', 'wb') as fp:
        ftp.retrbinary('RETR サンプル.txt', fp.write)

ポイントは ftp.encoding = ENCODING の行です。ここでは UTF8 に設定していますがサーバーの設定に合わせて適切なものを使うようにします。
   

ディレクトリをダウンロードする

続いて特定のディレクトリ(=フォルダ)以下のファイルをまとめてダウンロードしたい場合です。 Python 3.8 の時点ではこの用途にそのまま使えるメソッドは用意されていないため「ファイルをリストアップしてひとつずつダウンロードする」処理を自分で書く必要があります。

次の関数 pull_files() は特定のディレクトリ以下のファイルを一式ダウンロードしてくるものです。
 
def pull_files(ftp: FTP_TLS, remote_dir: str, local_root: Path) -> None:
    """特定のディレクトリ以下のファイルを指定されたパスに一式ダウンロードする"""
    ftp.cwd(remote_dir)

    def get_full_path(path, filename):
        return '{}/{}'.format(path, filename)

    # BFS ですべてのファイルをダウンロードする
    paths = ['.']
    while paths:
        path = paths.pop(0)
        for filename, info in ftp.mlsd(path):
            # カレントディレクトリ or ペアレントディレクトリ
            if info['type'] in ('cdir', 'pdir'):
                continue

            # ディレクトリ
            if info['type'] == 'dir':
                paths.append(get_full_path(path, filename))

            # ファイル
            if info['type'] == 'file':
                full_path = get_full_path(path, filename)
                print(full_path)
                local_path = local_root / full_path
                local_path.parent.mkdir(parents=True, exist_ok=True)
                with local_path.open('wb') as fp:
                    ftp.retrbinary('RETR {}'.format(full_path), fp.write)

mlsd() は指定されたディレクトリ直下のファイル・ディレクトリ一式を返してくれるメソッドです。 nlst() dir() 等同様のメソッドがいくつか用意されていますが、サーバー側が対応していれば mlsd() が最も豊富な情報を返してくれるので使えるならこれを使うのがよいと思います。 pull_files() は次のように使用します。

from ftplib import FTP_TLS
from pathlib import Path


ENCODING = 'utf8'

config = {
    'host': 'xx.xx.xx.xx',
    'user': 'username',
    'passwd': 'password',
}

with FTP_TLS(**config) as ftp:
    ftp.encoding = ENCODING
    remote_dir = 'home'
    local_root = Path('archive').resolve()
    local_root.mkdir()
    pull_files(ftp, remote_dir, local_root)

以上です。

GitHub Gist に私が実際に作って使ったスクリプトを汎用化したサンプルを置いているので興味のある方はそちらも参考にしてみてください。
   

当然のことながら、中身を理解せずにコピペで利用するのはお控えください。参考にされる際は自己責任でお願いします :D