2020年6月30日火曜日

データ分析入門 (Jupyter notebook + pandas or matplotlib)

概要

最近というわけではないですがデータ分析をするのに Jupyter notebook + pandas を使うのが流行ったので自分も今更ながら試してみました

環境

  • macOS 10.15.5
  • Python 3.8.3
    • jupyter notebook 6.0.3

Jupyter notebook のインストール

Jupyter notebook は簡単に言えばブラウザ上で動作するエディタです
Python のコードを直接実行したり作成した notebook を公開することもできます
anaconda というツールからインストールするのが公式の手順のようです

  • brew cask install anaconda

Anaconda-Navigator.app というアプリがインストールできたら開きましょう
そして jupyter notebook を Launch すれば OK です

デフォルトだと localhost:8888 で起動するようです
そこにブラウザでアクセスすると以下のようにユーザのホームディレクトリが表示されます

Hello Jupyter notebook

とりあえず動かしてみましょう
Python が動作するので「Hello World」でも表示してみましょう
「New」->「Python3」で新しく notebook を作成します

すると Python のプログラムを入力できるエディタになるのでここに

  • print("Hello")

と入力してみましょう
あとは「Run」ボタンを押せば Hello が表示されると思います

Pandas でデータを読み込んでみる

さてここからデータ分析っぽいことをしてみましょう
まずは Pandas に適当な二次元データを読み込ませてみます
二次元データを扱う場合には DataFrame というクラスを使います
pandas のデータ作成するには numpy のデータを元にしているのでそれぞれ import します

import pandas as pd
import numpy as np

そして numpy で二次元配列を作成してみます
試しに出力もしてみましょう

data = np.array([[10, -20, 64, -54], [-18, 43, 9, -34], [-41, 0, -30, 71], [1, -40, -21, 60], [11, -21, 49, -39], [-45, -26, 9, 62], [-48, 11, -13, 50]])
data

これを使って DataFrame オブジェクトを作成します

df = pd.DataFrame(data, index=range(1, len(data) + 1), columns=["A", "B", "C", "D"])
df

index は行番号で colums はそれぞれのデータのカラムになります
今回 index は 1 からスタートさせてみました
これを Jupyter notebook で表示させるとキレイなテーブルとして表示されるのが確認できると思います

Pandas の関数を使ってみる

pandas のデータにできればあとはいろいろな関数を使ってデータの抽出や加工ができます
例えば各列のデータの合計を求めたい場合は sum() を使います

df.sum()

また A 列のみを求めたい場合はカラム名を指定します

df["A"].sum()

他にも tail や value_counts (出現数のカウント) など様々な関数が用意されています
使い方はチートシートを見るのが手っ取り早いと思います

matplotlab でグラフにしてみる

Dataframe オブジェクトにできれば簡単にグラフにすることも可能です

df.plot()

これだけで折れ線グラフが作成されます
A カラムだけ見たい場合は df["A"].plot() という感じで先程と同じようにカラムを指定すれば OK です
グラフの見た目をちょっと変えたい場合は ggplot を使いましょう
これだけでもだいぶ見やすくなります

import matplotlib.pyplot as plt
plt.style.use('ggplot')
df.plot()

画像として保存したい場合は savefig を使いましょう
以下の場合はホームディレクトリの直下に作成されます

df.plot()
plt.savefig('graph.png')
plt.close('all')

最後に

Jupyter notebook でそれっぽくデータ分析してみました
ポイントはデータを如何にして Pandas のデータに落とし込むかかなと思います
Pandas のデータにさえできればあとは好きな関数を使って好きな分析ができるというわｋです

また使ってみて感じましたが正直データ分析がしたいだけなのであればわざわざ Jupyter notebook や pandas を使わなくてもエクセルで十分だと思います
これらを使うメリットはいろいろあると思いますが一番大きいのはこの後行うであろう機械学習のフェーズで scikit-learn や Tensorflow などのツールををそのまま Python で使えることだと思います

参考サイト

時刻: 0 件のコメント:
ラベル: ,

2020年6月23日火曜日

Celery のコードをテストする方法とドキュメントを生成する方法

概要

celery のコントリビュータになりたい人向けの記事です
docker でもできるのですがイメージの作成がかなり大変なのでローカルで直接コマンドを叩いたほうが簡単です

環境

  • macOS 10.15.5
  • Python 3.8.3
    • celery 4.4.4

ユニットテスト

  • git clone https://github.com/celery/celery.git
  • cd celery
  • pipenv install -r requirements/default.txt
  • pipenv install -r requirements/test.txt
  • vim t/unit/tasks/test_result.py
    def test_prtest(self):
        uid = uuid()
        x = self.app.AsyncResult(uid)
        assert x.id == uid
  • pipenv run py.test t/unit

ドキュメント生成

  • pipenv install -r requirements/docs.txt
  • rm -rf docs/_build
  • pipenv run sphinx-build -b html -d docs/_build/doctrees docs/ docs/_build/html
  • open docs/_build/html/index.html

参考サイト

時刻: 0 件のコメント:
ラベル: ,

2020年6月22日月曜日

Firefox WebExtension マウスのクリックイベントを特定のタグにだけ発生される方法

概要

querySelectAll を使うとページ内の特定のタグをすべて取得できます
取得したタグに対して addEventListener してあげます

環境

  • Firefox 77.0.1

サンプルコード

querySelectorAll で取得したタグは forEach で回せます
あとはそれぞれに対してイベントリスナーを追加すれば OK です

function clickedLeft(e) {
  if ('object' == typeof e){
    btnCode = e.button;
    switch (btnCode){
      case 0:
        alert('Left button clicked');
        break;
    }
  }
}

document.querySelectorAll("a").forEach(function(a) {
  a.addEventListener("mousedown", clickedLeft);
})

ちなみにどのタグでもいい場合は document.addEventListener("mousedown", func); で OK です

manifest.js

content_scripts にすることを忘れないようにしましょう

{
  "manifest_version": 2,
  "name": "test",
  "version": "1.0.0",
  "description": "test",
  "content_scripts": [
    {
      "matches": ["http://*/*", "https://*/*"],
      "js": ["main.js"]
    }
  ],
  "applications": {
    "gecko": {
      "id": "test@hawksnowlog.blogspot.com"
    }
  }
}
時刻: 0 件のコメント:
ラベル:

2020年6月21日日曜日

dnspython を使って Python で名前解決してみた

概要

dnspython を使うと簡単に実現できます
今回は単純に A レコードの取得を行ってみました

環境

  • macOS 10.15.5
  • Python 3.8.3
    • dnspython 1.16.0

インストール

  • pipenv install dnspython

サンプルコード

  • vim test.py
# カスタムのリゾルバを作成
# configure=True にするとデフォルトの DNS がシステムに設定されている DNS になる
resolver = dns.resolver.Resolver(configure=True)
# システム以外の DNS リゾルバを使いたい場合は設定する
resolver.nameservers = ["8.8.8.8", "8.8.4.4"]
# リゾルバを確認するためのデバッグ
print(resolver.nameservers)

try:
    # 名前解決できない場合はエラーが発生する
    anser = resolver.query("hawksnowlog.blogspot.com", "a")
    # Anser -> RRset -> IN.A.A という順番で IP アドレスを取得します
    for item in anser.rrset.items:
        print(item.address)
except NXDOMAIN as e:
    print(e)

参考サイト

時刻: 1 件のコメント:
ラベル:

2020年6月20日土曜日

Celery のリトライ機能を使ってみた

概要

Celery にはタスクのリトライ機能が標準で備わっています
基本はタスク内の raise に反応してリトライ処理が走ります
様々なオプションがあるので今回はオプションをいろいろと変更して挙動を確認してみました

環境

  • macOS 10.15.5
  • Python 3.8.3
    • celery 4.4.4

リトライ処理を追加した普通のタスク

まずは普通にリトライ機能を追加しただけのタスクです
必須なのは bind=True です
これを使うとタスク内で self が参照できるようになり、リトライする場合は self.retry() をコールすることで同じタスクをリトライできるようになります

from celery import Celery

app = Celery('tasks', broker='redis://localhost', backend='redis://localhost')

@app.task(bind=True)
def add(self, x, y):
    try:
        return x + y
    except Exception as exc:
        self.retry(exc=exc)

特定の時間ウェイトしたあとにリトライする方法

default_retry_delay を使います
単位は秒になります

from celery import Celery

app = Celery('tasks', broker='redis://localhost', backend='redis://localhost')

@app.task(bind=True, default_retry_delay=5)
def add(self, x, y):
    try:
        raise
        return x + y
    except Exception as exc:
        self.retry(exc=exc)

サンプルは必ずリトライします
デフォルトだとリトライ回数は 3 回のようです
変更する場合には self.retry(exc=exc, max_retries=5) こんな感じで retry の引数に指定できます

try, except を使わずにリトライさせる

タスクないで発生するエラーのクラスがわかっているのであればデコレータの引数でエラーを指定することで自動的にリトライさせることもできます

from celery import Celery

app = Celery('tasks', broker='redis://localhost', backend='redis://localhost')

@app.task(bind=True, default_retry_delay=5, autoretry_for=(Exception,), retry_kwargs={'max_retries': 5})
def add(self, x, y):
    raise
    return x + y

リトライの回数は retry_kwargs を使って辞書形式でオプションを渡すことができます

exponential backoff を使う

指定の時間待ってリトライだと効率がよくない場合があります
その場合は exponential backoff の機能があるのでこれを使いましょう
簡単に言えば失敗するたびにウェイトする時間が増える仕組みのことです

@app.task(bind=True, autoretry_for=(Exception,), retry_kwargs={'max_retries': 5}, retry_backoff=True)
def add(self, x, y):
    raise
    return x + y

これで確認してみるとウェイトの時間が 0s -> 1s -> 2s -> 4s のように指数的に増えているのが確認できると思います
exponential backoff の場合回数ではなく時間でリトライ回数を指定することもできます
retry_backoff_max を使います
デフォルトでは 600秒が指定されており 600秒間の間 exponential backoff でリトライを繰り返してくれます

ステータスはどうなっているか

リトライ中のタスクのステータスも一応確認してみました
flower を使っています

リトライが一度でも発生すると RETRY の状態になるようです
リトライ中だからと言って STARTED や PENDING にはならないようです

最後に

Celery のリトライ処理を試してみました
基本はリトライを実装したほうが良いと思いますが無駄なリトライなどが発生してタスクの完了が遅くならないように注意しましょう

時刻: 0 件のコメント:
ラベル: ,

2020年6月19日金曜日

Python http.client でクエリストリングなパラメータを送信

概要

JSON ボディを送る場合はこちらを参照してください

環境

  • Python 3.8.3

サンプルコード

  • vim lib/__init__.py
import http.client

class MyClient:
    def __init__(self, host, port="80"):
        self.host = host
        self.port = port
        self.conn = http.client.HTTPConnection(host, port)

    def get(self, path="/"):
        self.conn.request('GET', path)
        ret = self.conn.getresponse()
        return ret.status, ret.read().decode()

    def close(self):
        self.conn.close()

動作確認コード

  • vim main.py
from lib import MyHTTPSClient

c = MyHTTPSClient("kaka-request-dumper.herokuapp.com")
print(c.get(path="/hoge?key=value"))
c.close()
時刻: 0 件のコメント:
ラベル:

2020年6月18日木曜日

Gitlab 12.9 から 13.0 にアップグレードしてみた

概要

Omnibus Install でインストールした Gitlab を最新バージョンにアップグレードしてみました

環境

  • Gitlab EE 12.9 -> 13.0

準備

  • sudo su -
  • touch /etc/gitlab/skip-auto-backup
  • apt -y update

/etc/gitlab/skip-auto-backup を配置することで DB のバックアップをスルーできます

12.10 へアップグレード

どうやら 12.9 -> 13.0 への直接のアップグレードはできないようです
一旦 12.10 を経由する必要があります

  • apt -y install gitlab-ee=12.10.11-ee.0

パッケージのダウンロード＋ reconfigure がかかるので終了まで時間がかかります
またアップグレード完了後に 502 が続きますがすべてのプロセスが立ち上がるまで待ちましょう

13.0 へアップグレード

あとは最新版をインストールすれば OK です

  • apt -y install gitlab-ee

アップグレード完了後は skip-auto-backup ファイルを削除しましょう

  • rm /etc/gitlab/skip-auto-backup

参考サイト

時刻: 0 件のコメント:
ラベル:

2020年6月17日水曜日

ギガトラWifi を解約しました

概要

Nuro光G2V に申し込んだのでギガトラWifi を解約しました
解約方法について紹介します

環境

  • ギガトラWifi 100GB SIMのみプラン

時系列

  • 2020/06/09 解約フォームから連絡
  • 2020/06/10 SIMカードとソフトバンクカードを返送
  • 2020/06/16 解約完了の連絡

解約フォームから連絡

フォームと言っても専用のフォームはなく普通の問い合わせフォームから解約する旨を連絡する感じです
問い合わせ内容の件名に「お申し込みキャンセルまたは解約」があったのでこれを選択し内容に解約する旨を記載しました

当日中にサポートから連絡が返ってくると思います
返送の期限や返送先が記載されているので確認して期限までに返送します

自分は 6/8 に解約を申請したので当月 28 日まで利用できます
が面倒なのですぐに SIM を返却しました
また容量もまだ 70GB ほど残っていましたが返却しました

SIM の発送

返却するのは以下の通りです

  • SIM カード (ゲタも)
  • ソフトバンクカード

ギガトラWifi のカードみたいなやつはおそらく返送不要ですがいらなかったのでこちらも返しました
返送は追跡可能な発送方法なので自分はクリックポストにしました
SIM だけの場合はこれで十分ですし 198 円で送れます
SIM カードだけですが念の為ぷちぷちなどの緩衝材でくるんで送りましょう

メールで連絡

追跡番号をメールで連絡します
解約の申請をした際に来たメールにそのまま返信すれば OK だと思います
クリックポストの追跡番号は登録時に確認できます
もしくは印字したラベルのバーコードにも記載されています

あとは待つだけ

ちゃんと届けばサポートから連絡があるようです
これで解約完了になります
発送から約一週間後くらいに届きました

最後に

ギガトラWifi を解約したので実際の手順を紹介しました
月末に解約申請すると翌月分も請求されるので解約申請は月初にしましょう
また返送期日もありこれを超えても翌月請求になってしまうので余裕を持って返送しましょう
ギガを全部使い切ってもいいですが使い切るのに気を取られて発送が遅れないようにしましょう

時刻: 1 件のコメント:
ラベル:

2020年6月16日火曜日

リモートのサーバに対してリモートからコマンドを発行させる方法を考える

概要

リモートから外部のサーバに対してコマンドを発行する方法をいろいろと考えてみました
要するにわざわざリモートサーバにログインしてコマンドを発行しなくて済むような方法を考えてみます

環境

  • macOS 10.15.5 (client)
  • Ubuntu 16.04 (server)
    • golang 1.14.4

ノンパス ssh を使う方法

一番オーソドックスな方法かなと思います
ログインする側で公開鍵を作成します
作成のパスワードは空にします

  • ssh-keygen -t rsa
  • cat ~/.ssh/id_rsa.pub

生成された pub ファイルをログイン対象のサーバの authorized_keys に登録します

  • vim ~/.ssh/authorized_keys

これで ssh 経由でコマンドを発行できます

  • ssh vagrant@192.168.100.10 sudo gitlab-ctl status

また対象のサーバに対してダイレクトに実行するのが嫌だという場合は途中にプロキシなどを挟んで netcat 越しに実行しても良いと思います

メリット/デメリット

メリットは ssh なので暗号化はされています
また ssh 自体かなり使われている技術なので導入障壁は低いかなと思います
デメリットしてはサーバの ssh ポートを外部から LISTEN しておく必要があります
IP でアクセス制御したり ssh するユーザをある程度絞る必要があります

shell2http を使う

shell2http は登録したコマンドを http 経由で発行することができる Golang 製のツールです
Ubuntu へのインストールは以下のように行います

  • snap install shell2http

ただ一部権限がなく動作しないコマンドがあるようなので github にあるバイナリファイルを使いましょう (参考)

  • go get -u github.com/msoap/shell2http

例えば gitlab-ctl status というコマンドを /status のパスでコールするには以下のようにします

  • shell2http /status "gitlab-ctl status"
  • curl 192.168.100.10:8080/status

こんな感じで :8080/status にリクエストするとコマンドの実行結果が返ってきます
LISTEN させるポートや動作させるシェルなどオプションで様々な指定が可能です
--add-exit を指定すると /exit にアクセスした際に shell2http のプロセス自体を kill してくれます

  • shell2http --shell bash --port 8888 --add-exit /status "gitlab-ctl status"

--form オプションを使えばパラメータを受け取ることもできます
コマンド部分はシングルクォートで囲う必要があるので注意してください

  • shell2http--form /status 'gitlab-ctl $v_cmd'
  • curl "192.168.100.10:8080/status?cmd=status"

こんな感じでサブコマンドを変数にすると便利です

メリット/デメリット

メリットは ssh/22 をオープンしなくて良い点です
代わりに 8080 ポートをオープンしましょう
あとは curl なり HTTP プロトコルを使ってコマンドを実行できます
ssh 同様コマンドの結果はそのまま返ってくるのでパースなどは自分でやる必要があります

デメリットは shell2http のプロセス管理をする必要がある点です
また golang のインストールも必要になるので作業手順も若干増えます
暗号化に関してはオプションで証明書を指定可能なので https を使うことで通信を暗号化できます

Tips: No such file or directory や Cannot access などのエラーが出る場合

おそらく snap などでインストールしているのが原因です
go get を使って github から最新版を入手してください

その他案

その他使えそうな技術をピックアップしてみました

chef や ansible のプロビジョニングツールを使う

というのがモダンで良いとは思うのですが結局プロビジョニング対象のサーバに対して ssh できる必要があります
管理的なコストを考えるとソースに落とせるのでそれだけでメリットはありますが学習コストなどは増えてしまいます

クラウドサービスを使う

例えば GCP には gcloud コマンドという CLI が付属しておりこれをつかうことで簡単にリモートからコマンドを発行することができます
内部的には ssh を使っているのでプロトコル的には同じなのですがファイアウォールの設定やインスタンスの設定をしないで済むのは嬉しい点かなと思います

また AWS には AWS System Manager というサービスがありこれは画面からコマンドを実行することもできるのでかなり理想に近い感じはします
ちなみに SSM エージェントと呼ばれるエージェントをインストール必要がありこれはソースも公開されています

自作する

あとは気に入ったものがなかったり要件を満たさない場合には自作するしかないと思います
自作であればプロトコルやインタフェースは自由に選べるしエージェントを使って pull 側の仕組みを作ったりすることもできます

ですが当然開発コストやノウハウがそれなりに必要にはなるので障壁は一番高いかなと思います

最後に

リモートのサーバに対してログインせずにコマンドを実行する方法を考えてみました
over HTTP でコマンドを実行する方法は他にもいろいろとありそうだったので興味があれば調べてみてください

AWS のようにエージェント形式で気軽に使えるツールはあまりないようでやはりエージェント形式の場合はサーバも必須になるので構成も複雑になるかなと思います

時刻: 0 件のコメント:
ラベル:

2020年6月15日月曜日

curl で JSON を複数行にして実行する方法

概要

タイトルの通り
クエリストリングはこちら

環境

  • macOS 10.15.5
  • zsh 5.7.1

方法

curl -v -XPOST localhost:8000/path \
-H 'Content-Type: application/json' \
-d @- << EOF
{
  "key1": "value1",
  "key2": {
    "nested_key1": "nested_value1"
  },
  "key3": [
    "array_value1"
  ],
  "key4": false
}
EOF

@- で標準入力が使えます
それに << EOF でヒアドキュメントを与えることで複数行にしています

時刻: 0 件のコメント:
ラベル:

2020年6月14日日曜日

flack で設定ファイルに python ファイルを使う方法

概要

設定ファイルに Python ファイルを使うことでデフォルトの値や環境変数を使って柔軟に設定を定義することができます

環境

  • Python 3.8.3
  • flask 1.1.2

準備

  • pipenv install flask

アプリ作成

  • mkdir test
  • vim test/__ini__.py 
from flask import Flask
app = Flask(__name__)
app.config.from_object("test.config.Config")

app.config.from_object にクラスを指定することでそのクラスが持つクラス定数を参照することができます

  • vim run.py
from test import app

@app.route("/")
def index():
    return "%s,%s" % (app.config["NAME"], app.config["AGE"])

app.run(debug=True)

設定ファイル作成

  • mkdir test/config
  • vim test/config/__init__.py
import os

class Config:
    NAME = "hawksnowlog"
    AGE = os.environ.get("AGE", 10)

スーパークラスで共通の設定を管理する

上書きしたい設定だけ継承したクラスで再定義してあげましょう
AGE は Config クラスのものが使われます

import os

class Config:
    NAME = "hawksnowlog"
    AGE = os.environ.get("AGE", 10)

class DevConfig(Config):
    NAME = "dev_hawksnowlog"

config ProductionConfig(Config):
    NAME = "pro_hawksnowlog"

動作確認

  • pipenv run python run.py
  • curl localhost:5000

=> hawksnowlog,10

  • AGE=50 pipenv run python run.py
  • curl localhost:5000

=> hawksnowlog,50

参考サイト

時刻: 0 件のコメント:
ラベル:

2020年6月13日土曜日

Ruby のオブザーバパターンを試す

概要

Ruby のデザインパターンの一つであるオブザーバパターンをサンプルコードを実装しながら試してみました
ポイントは Observable の include と実際に何か処理をした際に notify を出すところです

環境

  • macOS 10.15.5
  • Ruby 2.7.1p83

subject.rb

まずは subject を作成します
subject はメイン処理側で通知を飛ばす側になります
まず include Observable が必須になります
これを行うことでオブザーバに通知にするために必要なインスタンスメソッドが使えるようになります

流れとしては先に add_observer でオブザーバとなるオブジェクトを登録しその後メイン処理を実行したあとで通知 (changed -> notify_observers)します
changed がコールされないとサブジェクト側の状態が変わったことにならず通知してもオブザーバ側のメソッドがコールされないので必ず changed をコールしましょう

notify_observers の引数は何でも OK です
通知時にオブザーバ側に渡したい値を指定すれば OK です

  • vim subject.rb
# coding: utf-8
require 'observer'

class Subject
  # Observable を include が必須
  # changed や notify_observers, add_observer や delete_observer が使えるようになる
  include Observable

  def initialize(name)
    @name = name
  end

  attr_reader :name

  def add_my_observer(observer)
    add_observer(observer)
  end

  def delete_my_observer(observer)
    delete_observer(observer)
  end

  def main_method
    puts "メイン処理を開始します"
    # subject 側の処理が完了したら登録したオブザーバに通知します
    changed
    notify_observers(self)
  end
end

observer1.rb

次に通知を受け取る側のオブザーバを作成します
ここでは update という名前のメソッドを実装しておく必要があります
サブジェクト側で notify_observers された際にこの update が呼ばれます
引数は notify_observers で渡されたものがそのまま入ります

  • vim observer1.rb
# coding: utf-8
class Observer1
  # notify_observers が呼ばれるとこのメソッドが発火する
  def update(subject)
    puts "#{subject.name} からの通知が届きました"
  end
end

テスト用メインスクリプト

では動作確認します
まずはサブジェクトとオブザーバを作りましょう
そして add_my_observer でオブザーバを登録します
そしてサブジェクト側のメインのメソッドを実行するとオブザーバの update メソッドがコールされているのが確認できると思います

また delete_observer でオブザーバを削除するとメインの処理を実行しても update がコールされなくなるのが確認できると思います

  • vim main.rb
require './observer1'
require './subject'

subject = Subject.new('my_subject')
observer1 = Observer1.new

subject.add_my_observer(observer1)
subject.main_method

subject.delete_my_observer(observer1)
subject.main_method

=>

メイン処理を開始します
my_subject からの通知が届きました
メイン処理を開始します

最後に

Ruby のオブザーバパターンを試してみました
メイン処理に対して Observable を追加するだけなのでかなり簡単に使えます
オブザーバを複数作成すれば登録/削除を繰り返すだけで状況に合わせて好きなオブザーバが使えるのが嬉しい点かなと思います

参考サイト

時刻: 0 件のコメント:
ラベル:

2020年6月12日金曜日

Ruby で Python のデコレータっぽいことをする

概要

Python のデコレータは関数の前後処理を実装することができる便利な機能です
Ruby でも同じことがしたいと思い試してみました

環境

  • macOS 10.15.5
  • Ruby 2.7.1p83

サンプルコード

  • vim hooks.rb
module Hooks
  def before(*method_names)
    to_prepend = Module.new do
      method_names.each do |name| 
        define_method(name) do |*args, &block|
          puts "before #{name}"
          super(*args,&block)
        end
      end
    end
    prepend to_prepend
  end
end

説明

最大のポイントは prepend を使っているところです
クラス (Example) が持っているメソッドをモジュール (Module.new したモジュール) で再度 define_method します
そしてそのモジュールを prepend することでクラスで呼ばれる同名のメソッドが呼ばれた際に prepend したモジュールで定義されたメソッドがコールされるという仕組みです

確認用スクリプト

  • vim test.rb
require './hooks'

class Example
  extend Hooks
  before :foo, :bar

  def foo
    puts "in foo"
  end

  def bar
    puts "in bar"
  end
end

a = Example.new
a.foo
  • ruby test.rb

=> before foo in foo

after の場合は

おそらく以下のような感じでいいと思います

module Hooks
  def after(*method_names)
    to_prepend = Module.new do
      method_names.each do |name| 
        define_method(name) do |*args, &block|
          super(*args,&block)
          puts "after #{name}"
        end
      end
    end
    prepend to_prepend
  end
end

最後に

Ruby で Python のデコレータっぽい処理を試してみました
Ruby にはデコレータパターンと呼ばれるデザインパターンが存在しますがこれは既存クラスの拡張っぽい機能なので少し異なります

参考サイト

時刻: 0 件のコメント:
ラベル:

2020年6月11日木曜日

Python デコレータ入門

概要

Python のデコレータは関数の前後に処理を入れたり関数の引数のチェックなどが行なえます
今回は自作のデコレータを作成して挙動を確認してみました

環境

  • macOS 10.15.5
  • Python 3.8.3

最初のステップ

まずはデコレータの雛形を作成します
デコレータ自体は特に何もしていないため test メソッドの内容がそのまま出力されます

  • vim 1.py
def first_deco(func):
    def wrapper(*args, **kwargs):
        func(*args, **kwargs)
    return wrapper

@first_deco
def test():
    print("hello")

test()
  • python 1.py

=> hello

func(*args, **kwargs) が test メソッド本体になります
なのでこの前後に何か処理を書くことで前後の処理を定義できます

デコレータに引数をつける

次にデレコータに引数をもたせてみます
一つ階層を上げてまず引数を受け取る関数を定義するのがポイントです
今回はデコレータに指定されたメッセージを本体の関数がコールされたあとに表示するようにしています

  • vim 2.py
def first_deco(msg):
    def _first_deco(func):
        def wrapper(*args, **kwargs):
            func(*args, **kwargs)
            print(msg) # 本体の関数のあとに msg を表示する
        return wrapper
    return _first_deco

@first_deco(msg="world")
def test():
    print("hello")

test()

=> hello world

デコレータを付与した関数の引数を扱う

デコレータを付与した関数をデコレータ内で扱ってみます
関数の引数は args と kwargs にそのまま入ってきます

  • vim 3.py
def first_deco(func):
    def wrapper(*args, **kwargs):
        func(*args, **kwargs)
        print(args[0])
        print(kwargs["post_msg"])
    return wrapper

@first_deco
def test(msg, post_msg=""):
    print("hello")

test("world", post_msg="!!")
# test("world") => KeyError: 'post_msg'

=> hello world !!

kwargs は本体の関数を呼び出した際に指定がないとデコレータ側には届かないようです
要するにで kwargs のデフォルト値は関数本体にしか影響を与えないようです

レスポンスがある場合

関数本体に return がある場合はデコレータ内でもちゃんと return を書く必要があります

  • vim 4.py
def first_deco(msg):
    def _first_deco(func):
        def wrapper(*args, **kwargs):
            return func(*args, **kwargs) + msg
        return wrapper
    return _first_deco

@first_deco(msg="world")
def test():
    return "hello"

print(test())

=> hello world

普通の関数としても使える

デコレータとして関数に付与しなくても普通の関数として使うこともできます
first_deco にキーワード引数を与えてその戻り値に関数本体である test を渡して最後に test に必要な引数を与えています　

  • vim 5.py
def first_deco(msg):
    def _first_deco(func):
        def wrapper(*args, **kwargs):
            return func(*args, **kwargs) + msg
        return wrapper
    return _first_deco

def test():
    return "hello"

print(first_deco(msg="world")(test)())

=> hello world

最後に

Python のデコレータの機能を使ってみました
とりあえずは関数の前後処理を実装するためのものだと思っておけば良さそうです
調べるといろいろとサンプルコードが出てくるのでデコレータを作ること自体そんなに大変ではなりですが実際に手を動かしてサンプルを作ってみることでより理解が深まるかなと思います

参考サイト

時刻: 0 件のコメント:
ラベル:

2020年6月9日火曜日

Nuro光 G2 V に申し込んでみた

概要

どんなときもWifiを契約してからまだ 1 年も経過していないのですがどんなときもWifi が 2020/03 - 04 で使い物にならなくなってしまったので固定回線を引きました
いろいろ考慮した結果 Nuro光を引くことにしたので申込みから開通までの流れを紹介したいと思います
先に言っておきますが申込みから回線工事まで最短で 1 ヶ月以上かかるので開通までに 1-2 ヶ月以上かかるので注意しましょう
自分はほぼ 2 ヶ月かかりましたので開通までの流れを紹介します

環境

時系列

  • 2020/04/01 申し込み
  • 2020/04/01 第一回の宅内回線工事の予約
  • 2020/04/05 So-net アカウントが記載された契約内容のハガキ到着
  • 2020/05/11 第一回の宅内回線工事
  • 2020/05/17 第二回の屋外回線工事日の予約
  • 2020/06/08 第二回の屋外回線工事
  • 2020/06/08 開通完了

申し込み手順

1. 価格.com から申し込み

キャッシュバックの金額が多いのでオススメかなと思います
「お申し込み公式サイトへ」を押すと価格.com 経由で申し込んだことになり特典を受けれることができるようです

2. 新規申し込みを選択

So-net アカウントがある場合は既存のアカウントを使っても OK です
今回はアカウントがないので新規で作成しました

3. 回線が引けるエリアか確認する

回線を引く住所を検索して提供可能エリアか確認しましょう

以下のように入力した住所が提供エリア内であれば OK です
おそらくですがこの段階で NG の場合は申込みできないのかなと思います

4. キャンペーンの状況などの確認

先に確認するようです
問題なければ個人情報の登録と決済情報の登録に進みます

5. 契約者情報の入力

契約者の情報を入力します

6. 決済情報の入力

カードのみのようです
クレジットカードの情報を入力しましょう

7. 契約情報に同意して契約する

申込内容を確認し同意する項目にチェックを入れ契約しましょう

8. 登録メールアドレスの確認

マイページにログインできる会員情報などは後日書面で送られてくるようです
とりあえず登録したメールアドレスにアドレス登録用の URL が記載されたメールが届くのでメールアドレスだけ登録しましょう

9. 登録電話番号の確認

なぜかこれだけ LINE でやらされます
登録した電話番号から LINE アカウントを引っ張って LINE に電話番号を確認する連絡がきます
SMS に確認用の PIN 番号が送信されるのでそれで電話番号も確認しましょう

10. 回線工事日の決定

最後に回線工事日を決定します
執筆時点 (2020/04/01) では最短で 2020/05/11 が選択可能だったのでそこを選択しました
これはできれば先に言ってほしかったです

また後日メールでも連絡が来ました

会員証到着

申し込みから 1 週間後あたりに郵送で So-net アカウントの ID/PW が記載されたハガキが届きます
それに記載された ID/PW を使ってマイページにログインできることを確認しましょう
また住所や電話番号などの登録情報が正しいことを確認しましょう

ID/PW を変更したい場合はマイページのお客様情報ページからできるようです
自動で付与される「So-net メールアドレス/メールアドレスパスワード」と「接続用ID/接続用パスワード」も変更できるようです

第一回宅内回線工事

まずは第一回の宅内回線工事がありました
Nuro光の工事担当者が来て 宅内の配線口から室外にある MDF 盤に対して配線します
マンションタイプなどは配線のための配線口のようなものが備え付けられているのでそこから配線します
MDF 盤には通常鍵がかかっていますが Nuro光の工事の方が鍵を持っていたので不動産会社から借りる必要はありませんでした

立ち会いのもと大体 1 時間ほどで終了しました
MDF 盤はない場合は直接線を引っ張ることになるのでもっと大変なのかもしれません
また宅内回線工事の前に賃貸であれば不動産会社に連絡しておきましょう
連絡の際に確認することとしては

  • そもそも Nuro 光を引いても問題ないか
  • 壁に穴を開けても問題ないか
  • MDF 盤があるかないか

あたりを確認すると良いと思います

またこの段階で Nuro光のルータ (ONU) も置いていかれます
まだこの段階では使えません
ルータは 6 種類あり自分は「HG8045Q」でした

第二回屋外回線工事

一回目の工事のあと約一週間で二回目の工事の日程を予約する連絡が SMS で来ました
5/17 (日) に連絡が来て最短が 6/7 (日) でした
自分は6/8 の午後に予約しました

二回目の工事は屋外工事になります
家などに業者が入ってくることはなく完全に屋外での作業になります
インターフォン越しに「作業開始します」という連絡と「作業終了しました」というやり取りくらいになります

作業が終了すればあとはルータの電源を入れて無線なり優先で接続すれば完了です
ちなみに無線の場合本体裏にデフォルトの SSID/PW が記載されているので確認しましょう

無事開通、スピードテスト

Google のスピードテストの結果です
端末は macOS 10.15.5 になります

HG8045Q には 11ac が使える 5GHz 帯と 2.4GHz 帯の 2 つがあります
一応それぞれで速度を測ってみました
時間帯にもよりますが夕方 16 時くらいの計測値です

  • 5GHz 帯

  • 2.4GHz 帯

まぁ当然と言えば当然ですが 11ac のほうが高速です
時間帯によっては 300Mbps ほど出るときもありました
マンションタイプであれば十分ではないでしょうか
おそらく有線接続であればもう少し速いでしょう
もう一つの 11ac 対応の「ZXHN F660A」でも計測したくなります

月額料金やキャンペーンなど

開通初月はどうやら無料のようです
そもそもコースの開始月は翌月 1 日になるようです

また月額料金は一年間だけ料金が安くなるキャンペーンが適用されているので 1 年後は料金が通常に戻るはずです

キャンペーンに関しては一年間使ったあとに振り込みの連絡が来るようなのでそれまで待ちましょう
連絡が来たら忘れないように振り込み口座などを登録する感じだと思います

またデフォルトでカスペルスキーのセキュリティソフトのライセンスに登録しています
不要であれば解約しましょう

ただ注意事項と見るとマンションミニのタイプは無料で使えるとあるのと初月解約すると料金が発生するとあるのでとりあえず放置しておくことにしました

その他解約時の違約金や契約内容はマイページで確認できるのでそこで確認するのがいいと思います

  • 2020/07 3,300 円 (事務手数料のみ)

ルータ設定

管理画面は 192.168.1.1 でアクセスできます
ID/PW は admin/admin でした

SSID 変更

デフォルトはブルートフォースなどであっさり入られてしまうので変更しましょう

  • WLAN タグ -> 5G基本ネットワーク設定

で変更できます
ちなみに SSID や PW を変更して適用するたびに ONU が再起動するので面倒です

基本は 5GHz を使えばいいので 2.4Ghz はオフにしてもいいと思います
各チャネルで端末の接続台数は 32 台が限界なので台数がマックスになったらオンにするとかでいいと思います

あとはポートフォーワードや Mac アドレスフィルタリング、URL フィルタなどの基本的なセキュリティ機能のほか Dos 対策として SYN フラッドや Smurf 攻撃の防止機能がデフォルトで有効になっていました
何か悪さしそうな気もしますがそのときに設定を見直す感じでいいと思います

最後に

NURO 光 G2V(マンションミニ) の申し込み方法と工事から開通までの流れを紹介しました
戸建ての場合は流れは同じですが穴を開けたりする必要が可能性として出てくるのでもっと面倒かなと思います
MDF 盤のあるマンションであれば時間はかかりますが手間はかからず開通できると思います

また解約時ですが SC ケーブルの回収などが必要な場合は 1 万円かかる可能性があるそうです
このあたりは賃貸の場合不動産会社との相談になります

時刻: 0 件のコメント:
ラベル: ,

2020年6月8日月曜日

pydoc 超入門

概要

Python の標準で付属されている pydoc を使ってみました
docstring は Google Style Guide を参考にしています

環境

  • macOS 10.15.5
  • Python 3.7.3

サンプルコード

  • vim module/__init__.py
"""
main.py
"""
from module import TestClass

if __name__ == '__main__':
    tc = TestClass("Hello")
    tc.say("Python!")
  • vim main.py
from datetime import datetime
"""
This is a test module.
"""

__author__  = "hawksnowlog <hawksnowlog@gmail.com>"
__version__ = "0.1.0"
__date__    = datetime.now()

class TestClass:
    """
    This is a test class

    Attributes:
      default_msg (str): Default message
    """

    def __init__(self, default_msg):
        """
        Constructor

        Args:
          default_msg (str): Default message
        """
        self.default_msg = default_msg

    def say(self, msg):
        """
        Print a message

        Args:
          msg (str): A message you like
        Returns:
          None
        Raises
          None
        """
        print("%s, %s" % (self.default_msg, msg))

これらを実行する方法以下の通りです

  • python3 main.py

ドキュメント生成

作成したソースコードからドキュメントを生成します

  • python3 -m pydoc -w main module

.py などの拡張子は指定しません
-w で HTML ファイルを出力します
生成される HTML ファイルは以下の通りです

Package Contents はサブディレクトリがある場合に出力されます
例えば module/sub/__init__.py がある場合に表示されます

残念な点

  • Data descriptors defined here: の部分を変更する方法が不明
  • index.html は出力されない
  • 外部のクラスなどにリンクできない (参考)
  • docstring と呼ばれるドキュメントの書き方がたくさんあるのでどれを選択すればいいのかわからない

最後に

Python 標準のドキュメントツール pydoc を使ってみました
rdoc や yard に比べるとちょっと機能が劣りますが簡単に使えるのは良い点かなと思います
Sphinx を使えばデザインなども設定できますがいろいろと準備が必要なのが難点です

参考サイト

時刻: 0 件のコメント:
ラベル:

2020年6月7日日曜日

Celery で chain 実行したタスクの状態を確認する方法

概要

celery で chain を使った際の親子タスクの状態を確認する方法を紹介します

環境

  • macOS 10.15.5
  • Python 3.7.3
    • celery 4.4.3

ポイント

  • タスクの状態を確認するタスクを登録する
  • AsyncResult.as_tuple() を使ってすべてのタスク情報を返却する

タスク作成

まずタスクを作成します
ここで1つ目のポイントですがタスクの状態を確認するのに AsyncResult というクラスを使います
これに task_id を指定することでタスクを生成し状況を確認するのですが同一の celery オブジェクトから作成した AsyncResult でないとうまくタスクが生成できませんでした
なので「指定したタスクの状態を確認するタスク」として登録します

  • vim sub_tasks.py
import time
from celery import Celery
from celery.result import AsyncResult

app = Celery('sub_tasks', backend='redis://localhost', broker='redis://localhost')

@app.task(bind=True)
def check(self, req_id):
    task = self.AsyncResult(req_id)
    return (task.id, task.state, task.info, task.name)

@app.task
def add(x, y):
    time.sleep(10)
    return x + y

@app.task
def multi(x, y):
    return x * y

また celery.result.AsyncResult は指定された tasks_id が存在しない task_id でもエラーなどにはなりません
task.state もまだ実行していない状態と同じ PENDING が返ってきます
なのでもし存在しないタスク ID かどうかを調べる場合には tasks.state 以外に task.info や task.name など他の要素も使って判断する必要がありそうです
※ただいろいろ調べた結果現状だと判断する要素はないようです (参考1, 参考2)

add, multi は今回実行するタスクです

実行メインスクリプト

作成したタスクを実行するスクリプトです
chain の結果から親子関係のタスク情報を as_tuple() で取得してすべてのタスク情報を返却します
なお as_tuple() が謎の入れ子になっているので必要なタスク ID の情報だけ取得して返却してます

  • vim chain_test.py
import json
from celery import chain
from sub_tasks import add, multi

tasks = chain(
    add.s(1, 2),
    add.s(4),
    multi.s(10)
).apply_async()

ret = tasks.as_tuple()
print("%s,%s,%s" % (ret[0][0], ret[0][1][0][0], ret[0][1][0][1][0][0]))

ステータス確認スクリプト

あとは取得したタスク ID を使ってチェックするスクリプトを実行します

  • vim chain_check.py
import sys
from sub_tasks import check

state = check.delay(sys.argv[1]).get()

print(state)

動作確認

まずワーカーを起動します

  • pipenv run celery -A sub_tasks worker -l info

次にメインスクリプトを実行します

  • pipenv run python3 chain_test.py

すると cb6c0264-6d8f-4105-ae4d-e2d8c21cb9c1,a61b8f1d-1c82-4b04-b354-f39e80f66ffc,a5c59713-7579-4155-befd-917df16d65e7 親子関係のタスク情報が返ってくるのでこれを使って状態を確認します
右に行くほど一番最初に実行される上位のタスク ID になります

例えば最上位のタスクの状態を調べたい場合は以下のように実行します

  • pipenv run python3 chain_check.py a5c59713-7579-4155-befd-917df16d65e7

=> ['a5c59713-7579-4155-befd-917df16d65e7', 'SUCCESS', 3, None]

最後に

Celery でタスクの状態を取得する方法を試してみました
今回は chain から各種タスクの状態を取得しましたが group や chord でも同じようにできるはずです
GroupResult というクラスもあるのでこちらも使えると思います

Tips

celery オブジェクト作成時に backend= を 指定しないと AttributeError: 'DisabledBackend' object has no attribute '_get_task_meta_for' が発生します

参考サイト

時刻: 0 件のコメント:
ラベル: ,

2020年6月6日土曜日

celery でワークフローやパイプラインを実現するための chain や group を試してみた

概要

celery でワークフローを実現するための仕組みとして Canvas という仕組みがあります
これはワークフローを実現するための機能で様々なメソッドが提供されています
今回は chain や group と言ったワークフローを実現するためのメソッドを試してみました

環境

  • macOS 10.15.5
  • Python 3.7.3
    • celery 4.4.3

chain を試す

まずは chain を試してみます
chain は簡単に言えば前に実行したタスクの結果を使って次のタスクを実行することができる機能です

まずはいくつかのタスクを作成します
ここで定義したタスクを chain でつなぎ合わせてみます

  • vim sub_tasks.py
from celery import Celery

app = Celery('sub_tasks', backend='redis://localhost', broker='redis://localhost')

@app.task
def add(x, y):
    return x + y

@app.task
def multi(x, y):
    return x * y

ワーカーとして起動しておきましょう

  • pipenv run celery -A sub_tasks worker -l info

次にタスクを chain でつなぎ合わせ実際に実行するスクリプトを作成します

  • vim chain_test.py
from celery import chain
from sub_tasks import add, multi

tasks = chain(add.s(1, 2), multi.s(10)).apply_async()

print(tasks.get())

ポイントは multi というタスクの引数が 1 つになっているところです
これは add の結果 (3) を 1 つ目の引数として渡すためで multi の引数を 2 つにしてしまうと引数が合わないよというエラーになってしまいます
あとは実行すればワーカーで実行された 2 つのタスクの結果が返ってくると思います

  • pipenv run python3 chain_tasks.py

=> 30

[2020-06-02 15:37:59,798: INFO/ForkPoolWorker-2] Task sub_tasks.add[5e35ecc1-19e5-405a-b355-21578f1d85cd] succeeded in 0.03750605799999107s: 3 [2020-06-02 15:37:59,814: INFO/ForkPoolWorker-4] Task sub_tasks.multi[081a469e-03ea-4273-8884-2a43a4988dcb] succeeded in 0.014889055999987022s: 30

group を試す

group はまとめたタスクを並列で実行することができる機能です
先程の chain は前のタスクの結果に依存するため前のタスクが終わらないとダメですが group はすべてのタスクが並列に実行されます
sub_task.py は先程作成したものを使います

  • vim group_test.py
from celery import group
from sub_tasks import add, multi

tasks = group(add.s(1, 2), multi.s(10, 10)).apply_async()

print(tasks.get())

group の場合前のタスクの結果を参照しないのでちゃんとそれぞれのタスクの引数は 2 つ指定してあげます
これで実行すると各タスクで実行された結果が配列で取得できます

  • pipenv run python3 group_test.py

=> [3, 100]

[2020-06-02 15:41:21,307: INFO/ForkPoolWorker-4] Task sub_tasks.multi[129a3b43-ab45-4578-b659-f50c6bd8de5c] succeeded in 0.0007359960000030696s: 100 [2020-06-02 15:41:21,307: INFO/ForkPoolWorker-2] Task sub_tasks.add[c6d50e58-abb5-469f-8106-0fbe219cbe11] succeeded in 0.000807351000048584s: 3

chord を試してみる

コードと読みます
chord は group に似ていますが並列処理の最後に指定したコールバックメソッド用のタスクを実行することができます
sub_tasks.py を少し書き換えます

  • vim sub_tasks.py
from celery import Celery

app = Celery('sub_tasks', backend='redis://localhost', broker='redis://localhost')

@app.task
def add(x, y):
    return x + y

@app.task
def multi(x, y):
    return x * y

@app.task
def total(ary):
    return sum(ary)
  • vim chord_test.py
from celery import chord
from sub_tasks import add, multi, total

tasks = chord((add.s(1, 2) for i in range(5)), total.s()).apply_async()

print(tasks.get())

1 + 2 = 3 を 5 回繰り返して生成される配列に対してそれぞれの要素を足し合わせた結果 (15) を返します

  • pipenv run python3 chord_test.py

=> 15

結果を確認するとわかりますが add タスクがすべて実行されたあとに total タスクが実行されていることがわかります

[2020-06-02 15:59:52,309: INFO/MainProcess] Received task: sub_tasks.add[18d99f43-02bf-4fac-a051-20ad627eaa73] [2020-06-02 15:59:52,312: INFO/ForkPoolWorker-2] Task sub_tasks.add[18d99f43-02bf-4fac-a051-20ad627eaa73] succeeded in 0.0014969600000256378s: 3 [2020-06-02 15:59:52,313: INFO/MainProcess] Received task: sub_tasks.add[262d1e52-a80f-49f4-a486-5dd43b24866d] [2020-06-02 15:59:52,316: INFO/ForkPoolWorker-2] Task sub_tasks.add[262d1e52-a80f-49f4-a486-5dd43b24866d] succeeded in 0.001473454999995738s: 3 [2020-06-02 15:59:52,317: INFO/MainProcess] Received task: sub_tasks.add[e0d01505-7791-4336-aeee-75dad07babd6] [2020-06-02 15:59:52,320: INFO/MainProcess] Received task: sub_tasks.add[f9bd8bc1-054c-4efb-838e-80f69d3fe128] [2020-06-02 15:59:52,320: INFO/ForkPoolWorker-2] Task sub_tasks.add[e0d01505-7791-4336-aeee-75dad07babd6] succeeded in 0.0011679110000102355s: 3 [2020-06-02 15:59:52,323: INFO/MainProcess] Received task: sub_tasks.add[600bc612-8d29-4950-aaa1-d019b26176d8] [2020-06-02 15:59:52,325: INFO/ForkPoolWorker-4] Task sub_tasks.add[600bc612-8d29-4950-aaa1-d019b26176d8] succeeded in 0.001488192000010713s: 3 [2020-06-02 15:59:52,362: INFO/ForkPoolWorker-2] Task sub_tasks.add[f9bd8bc1-054c-4efb-838e-80f69d3fe128] succeeded in 0.038503546000015376s: 3 [2020-06-02 15:59:52,363: INFO/MainProcess] Received task: sub_tasks.total[0b704079-30b5-4a00-9083-8027f010603e] [2020-06-02 15:59:52,365: INFO/ForkPoolWorker-2] Task sub_tasks.total[0b704079-30b5-4a00-9083-8027f010603e] succeeded in 0.0006079719999831923s: 15

map を試してみる

ここから紹介する map と starmap はタスクに対して実行する機能です
map は引数が 1 つのタスクに対して引数で与えた Iteratable な値を順番に実行します

  • vim map_test.py
from sub_tasks import total

ret = total.map([(1, 2, 3, 4, 5)]).apply_async()
print(ret.get())
  • pipenv run python3 map_test.py

=> [15]

starmap を試してみる

map が単一の引数のタスクに対する命令でしたが starmap は引数が複数あるタスクに対して実行できます

  • vim starmap_test.py
from sub_tasks import add

ret = add.starmap([(1, 2), (3, 4)]).apply_async()
print(ret.get())
  • pipenv run python3 starmap_test.py

=> [3, 7]

chunks を試してみる

chunks は与えられた Iteratable な引数から実行するべきタスク数を算出し最後に与えられた引数分エンキューするジョブを分割します

  • vim chunks_test.py
from sub_tasks import add

ret = add.chunks([(1, 2), (3, 4), (5, 6), (7, 8), (9, 10), (11, 12)], 2).apply_async()
print(ret.get())
# ret = add.starmap([(1, 2), (3, 4), (5, 6), (7, 8), (9, 10), (11, 12)]).apply_async()
# print(ret.get())
  • pipenv run python3 starmap_test.py

=> [[3, 7], [11, 15], [19, 23]]

結果を見るとわかるのですが 1 つのタスクの結果としてではなく 3 つのタスクの結果として受け取れます (2 つずつに分割したため)

ちなみに starmap 側を実行するとわかりますが 1 つのタスクの結果としてすべての結果が受け取れます

=> [3, 7, 11, 15, 19, 23]

最後に

celery の Canvas 機能を試してみました
chain や group, chord と言った便利なメソッドとタスクを組み合わせることでワークフローを実現することができます
map, starmap, chunks はワークフローとは少し毛色が違いますがタスクを反復的に実行したい場合に使えるメソッドです

Canvas は 1 つのシンプルなタスクのみを実行する場合には気にする必要はありませんが複数のタスクを組み合わせて 1 つの処理にする場合には必須な機能だと思います

参考サイト

時刻: 0 件のコメント:
ラベル: ,

2020年6月5日金曜日

celery でキューのルーティング機能を試してみた

概要

キューのルーティング機能とは簡単に言えばキューとワーカープロセスの紐付けです
「このキューに入った場合はこのワーカーに処理させる」ということができるので優先度付けなどができるようになります
今回は簡単なデモアプリを作成して試してみました

環境

  • macOS 10.15.5
  • Python 3.7.3
    • celery 4.4.3

2 つのタスクを作成する

キューがごとに処理させるタスクを分けるのでファイルも複数に分けます
タスクを作成するときにはキューは指定しません

  • vim tasks.py
from celery import Celery

app = Celery('tasks', broker='redis://localhost')

@app.task
def add(x, y):
    print(x + y)
  • vim tasks2.py
from celery import Celery

app = Celery('tasks2', broker='redis://localhost')

@app.task
def multi(x, y):
    print(x * y)

2 つのタスクを起動する

  • pipenv run celery -A tasks worker

tasks2 は -Q オプションでキューを指定します

  • pipenv run celery -A tasks2 worker -Q tasks

エンキューして動作確認する

エンキューする際に delay ではなく apply_async を使います
そして引数に queue= を指定することで特定のキューに対してメッセージをエンキューできます

  • vim test.py
from tasks import add
from tasks2 import multi

add.delay(100, 1)
multi.apply_async((100, 1), queue="tasks2")

これで動作確認するとそれぞれのワーカープロセスでログが出力されているのが確認できると思います

最後に

celery でキューのルーティングを試してみました
一点ポイントなのは同じキューにエンキューさた際にタスクに定義されていないメソッドがあると KeyError になるので注意が必要です
今回の場合ですと tasks は multi メソッドがないのでもし multi.delay でデフォルトのキューにエンキューしてしまった場合 pipenv run celery -A tasks worker で起動したワーカー側に KeyError が表示されてしまいます

参考サイト

時刻: 0 件のコメント:
ラベル: ,

2020年6月4日木曜日

Celery + flask 超入門

概要

flask で celery を使って簡単な非同期処理を試してみました

環境

  • macOS 10.15.5
  • Python 3.7.3
    • celery 4.4.3
    • flask 1.1.2

インストール

  • pipenv install celery redis flask

準備: まずは簡単な flask アプリの作成

まずは何も考えず flask アプリを作成します
その後で celery と連携することで具体的な連携方法を理解します
なお flask アプリはちゃんとモジュールなども考慮して作成しています

完成後のファイル構造は以下の通りです

. ├── Pipfile ├── Pipfile.lock ├── runserver.py └── testapp ├── __init__.py ├── celery_maker.py ├── route.py └── tasks.py   1 directory, 7 files

flask アプリの作成

flask アプリを作成します

  • vim testapp/__init__.py
from flask import Flask

app = Flask(__name__)

import testapp.route

testapp.route はあとで後述しています

アプリケーションを起動するスクリプトの作成

__init__.py で作成したアプリを起動するだけのスクリプトです

  • vim runserver.py
from testapp import app

if __name__ == "__main__":
    app.run(debug=True)

ルーティングを管理するファイルの作成

ここにルーティング情報を記載していきます
celery のタスクもここで実行することになります

  • vim testapp/route.py
from testapp import app

@app.route('/')
def index():
    return "ok"

ここから本番: celery と連携する

celery オブジェクトを作成するメソッドを定義

flask アプリを元に celery オブジェクトを作成します

  • vim testapp/celery_maker.py
from celery import Celery

def make_celery(app):
    celery = Celery(
        app.import_name,
        backend=app.config['CELERY_RESULT_BACKEND'],
        broker=app.config['CELERY_BROKER_URL']
    )
    celery.conf.update(app.config)

    class ContextTask(celery.Task):
        def __call__(self, *args, **kwargs):
            with app.app_context():
                return self.run(*args, **kwargs)

    celery.Task = ContextTask
    return celery

__init__.py に celery の設定を記載する

celery が必要とするブローカの情報を記載します
この app.conf の情報から celery オブジェクトを作成し管理します

  • vim testapp/__init__.py
from flask import Flask
from testapp.celery_maker import make_celery

app = Flask(__name__)
app.config.update(
    CELERY_BROKER_URL='redis://localhost:6379',
    CELERY_RESULT_BACKEND='redis://localhost:6379'
)
celery = make_celery(app)

import testapp.route

タスクの作成

非同期で行う celery タスクを作成します
わかりやすいようにタスクは個別のファイルで管理できるようにしています
__init__.py で管理している celery オブジェクトを元にタスクを作成します
非同期タスクはアノテーションを付与することで生成できます

  • vim testapp/tasks.py
from testapp import celery

@celery.task
def delay_print(msg):
    print(msg)

route.py でタスクをコールする処理を追加する

あとは実際にタスクをコールする処理を記載すれば OK です

  • vim testapp/route.py
from testapp import app
from testapp.tasks import delay_print

@app.route('/')
def index():
    delay_print.delay("hello")
    return "ok"

動作確認

アプリを起動します

  • pipenv run python runserver.py

次にタスクを実行するワーカープロセスを起動します
ポイントはアプリ名に「testapp.celery」を指定することです
testapp だけだと flask オブジェクトも読み込んでしまいエラーになります
「'flask' object has no attribute 'user_options'」

  • pipenv run celery -A testapp.celery worker

あとは curl localhsot:5000 をしてみるとレスポンスが返ってくるのと同時にワーカープロセス側のログに「hello」も表示されるのが確認できると思います

最後に

flask + celery を試してみました
試すこと自体は非常に簡単でしたがちゃんとモジュールや管理するファイルの構造を考えないとソースがぐちゃぐちゃになっちゃいそうです

今回の構造であればタスクを増やしたい場合は tasks.py だけを修正すればいいのでキレイに管理できていると思います

参考サイト

時刻: 2 件のコメント:
ラベル: ,
登録: 投稿 (Atom)