2023年7月31日月曜日

fastapi で XML なレスポンスを考える

fastapi で XML なレスポンスを考える

概要

今回は pydantic_xml を使って fastapi で XML レスポンスを返却する方法を考えます

環境

  • Python 3.11.3
  • fastapi 0.100.1
  • pydantic-xml 1.0.0

サンプルコード

  • vim ./app.py
from fastapi import FastAPI, Response
from pydantic_xml import BaseXmlModel, element

app = FastAPI()


class User(BaseXmlModel, tag="User"):
    name: str = element(tag="Name")
    age: int = element(tag="Age")


@app.get("/", response_class=Response)
def xml():
    data = """<?xml version="1.0"?>
    <User>
      <Name>hawksnowlog</Name>
      <Age>10</Age>
    </User>
    """
    result = User.from_xml(data)
    return Response(content=result.to_xml(), media_type="application/xml")

動作確認

  • pipenv run uvicorn app:app --reload
  • curl -v localhost:8000/

=> <User><Name>hawksnowlog</Name><Age>10</Age></User>

解説

XML の構造を pydantic で扱うには pydantic_xml.BaseXmlModel を使います
各エレメントごとにクラスの属性として定義することができます
今回は単純な文字列と数字のフィールドなので element を使っています
タグ内の属性などを扱うこともできるので必要に応じて BaseXmlModel の定義を変更してください

実際にオブジェクトを作成する際は今回は文字列の XML 情報から from_xml メソッドを使って User オブジェクトを作成しています
更にそこから to_xml というシリアライズ用のメソッドが BaseXmlModel に生えているのでそれを使ってレスポンス情報を生成しています

あとは fastapi の Response クラスを使ってデータを application/xml として返却しています

最後に

pydantic-xml を使って fastapi で XML レスポンスを返却する方法を考えてみました

結局この方法だとレスポンスモデルは pydantic で管理できているが最終的なレスポンスは fastapi の Response クラスを使ってしまっているので微妙な気はします (openapi に載らないなど)

本当は response_class に XMLResponse などのクラスを定義して自動でシリアライズしてくれるような便利クラスを作成するほうがいいかなと思います (また検証次第紹介するかも)

参考サイト

2023年7月28日金曜日

Pythonでどんな名前の属性でも受け入れることができる便利クラスを考える

Pythonでどんな名前の属性でも受け入れることができる便利クラスを考える

概要

__setattr__, __getattribute__ をオペレーションオーバライドしてどんな名前の属性でも設定/取得できるような便利クラスを作ってみます

環境

  • Python 3.11.3

サンプルコード

from typing import Any


class AnythingOK:
    def __setattr__(self, __name: str, __value: Any) -> None:
        super().__setattr__(__name, __value)

    def __getattribute__(self, __name: str) -> Any:
        try:
            return super().__getattribute__(__name)
        except AttributeError:
            return None


if __name__ == "__main__":
    aok = AnythingOK()
    aok.name = "hawksnowlog"
    print(aok.name)
    print(aok.age)
    aok.age = 10
    print(aok.age)

指定の属性がない場合は None を返却します

使い道

  • とりあえず値を保持しておきたい場合に dict や list ではなくオブジェクトとして保存したい場合に使えるかも
  • 属性名などが動的に変わる場合に使えるかも
  • クラスとして定義しているので to_dict などの便利メソッドを生やせば追加した属性の一覧や加工などが簡単に行えるかも

注意点

__setattr__ 内で setattr() をコールすると RecursionError になるのでコールしないようにしましょう

__getattribute__ 内で hasattr() や dir() をコールすると RecursionError になるのでコールしないようにしましょう

最後に

便利なのだろうか

参考サイト

2023年7月27日木曜日

SQLAlchemyのautomapを使って自動で生成されたクラスの定義を表示する方法

SQLAlchemyのautomapを使って自動で生成されたクラスの定義を表示する方法

概要

過去に automap の使い方を紹介しました
automap を使った場合クラスのモデル定義がないためコード内でデータベースの定義を確認するすべがありません
またコードエディタなどで補完なども行えず pyright ではそのような属性がないと言われてエラーになってしまいます
なのでクラス定義がないと困るケースが多いです

今回はそんな場合に使えそうな automap からクラスの情報を標準出力に吐き出す方法を紹介します

環境

  • macOS 13.4.1
  • Python 3.11.3
  • sqlalchemy 2.0.19

サンプルコード

  • vim ./app.py
from sqlalchemy import create_engine, inspect
from sqlalchemy.ext.automap import automap_base

# automap を使用するための準備
Base = automap_base()

# エンジンの作成
engine = create_engine("mysql+pymysql://root@localhost/test?charset=utf8mb4")

# テーブル定義の読み込み
Base.prepare(autoload_with=engine)

# user テーブルから user モデルの抽出
User = Base.classes.user

# インスペクタの作成
inspector = inspect(User)
# import pprint
# pprint.pprint(dir(inspector))

# クラス定義を標準出力に表示
print(f"class {User.__name__.title()}(Base):")
print(f"    __tablename__ = '{User.__table__}'")

# カラム情報を表示
for column in inspector.columns:
    column_params = []
    if column.primary_key:
        column_params.append("primary_key=True")
    if column.foreign_keys:
        foreign_key = list(column.foreign_keys)[0]
        column_params.append(
            f"ForeignKey('{foreign_key.column.table.name}.{foreign_key.column.name}')"
        )
    if column.unique:
        column_params.append("unique=True")
    if column.index:
        column_params.append("index=True")
    if column.default is not None:
        default_value = column.default.arg
        column_params.append(f"default={default_value}")
    else:
        default_value = column.default
        column_params.append(f"default={default_value}")

    type = repr(column.type)
    if type.startswith("VARCHAR"):
        type = type.replace("VARCHAR", "String")
    else:
        type = type.title()
    column_str = f"{column.name} = Column({type}"
    if column_params:
        column_str += f", {', '.join(column_params)}"
    column_str += ")"
    print(f"    {column_str}")

# ユニークキー情報を表示
if "unique_constraints" in dir(inspector):
    for uc in inspector.unique_constraints:
        uc_columns = ", ".join([column.name for column in uc["column_names"]])
        print(f"    __table_args__ = (UniqueConstraint({uc_columns}),)")

# インデックス情報を表示
if "indexes" in dir(inspector):
    for index in inspector.indexes:
        index_columns = ", ".join([column.name for column in index["column_names"]])
        print(f"    {index['name']} = Index('{index['name']}', {index_columns})")

# プライマリーキー情報を表示
primary_keys = inspector.primary_key
if primary_keys:
    pk_names = ", ".join([pk.name for pk in primary_keys])
    print(f"    __mapper_args__ = {{'primary_key': [{pk_names}]}}")

inspector を使って automap により生成されたクラスに対してカラムやキーの情報を取得しそれを print する流れになります
ユニークキーやインデックスがない場合はそもそも属性自体が inspector に生えないので属性があるかのチェックが必要です

また今回表示している情報がすべてでないケースがあるので表示したい情報は既存のテーブル定義を調べて表示させる必要があります
例えば enum や text, blob 型への変換は対応していないので上記コードに追加する必要があります

これを実行すると以下のようなクラス定義が表示されます

class User(Base):
    __tablename__ = 'user'
    id = Column(Integer(), primary_key=True, default=None)
    name = Column(String(collation='utf8mb4_general_ci', length=50), default=None)
    age = Column(Integer(), default=None)
    profile = Column(String(collation='utf8mb4_general_ci', length=50), default=None)
    __mapper_args__ = {'primary_key': [id]}

動作確認

生成されたクラス情報を使って実際に CRUD 処理してみます

  • vim ./test.py
from sqlalchemy import create_engine
from sqlalchemy.orm import declarative_base, sessionmaker
from sqlalchemy.schema import Column
from sqlalchemy.types import Integer, String

engine = create_engine("mysql+pymysql://root@localhost/test?charset=utf8mb4")
SessionClass = sessionmaker(engine)
db_session = SessionClass()

Base = declarative_base()


class User(Base):
    __tablename__ = "user"

    id = Column(Integer(), primary_key=True, default=None)
    name = Column(String(collation="utf8mb4_general_ci", length=50), default=None)
    age = Column(Integer(), default=None)
    profile = Column(String(collation="utf8mb4_general_ci", length=50), default=None)

    __mapper_args__ = {"primary_key": [id]}

    def __repr__(self) -> str:
        return f"id: {self.id}, name: {self.name}, age: {self.age}, profile: {self.profile}"


users = db_session.query(User).all()
for u in users:
    print(u)

改行や repr 関数は手動で追加しています
一応自動生成されたモデルを使ってデータを取得できることは確認できました

最後に

リフレクションを使って愚直にやっていくしか方法がないので完璧にモデルを定義するのであれば結構労力がかかりそうです
もしかすると最初から手動で頑張って既存テーブルの構造を解析してクラスを定義するほうが正確で早いかもしれません

automap は便利なのですがクラス定義がないのでコードの可読性が下がるほかエディタや静的型チェックとの相性も悪いのでプロダクションのコードで採用するのは微妙なのかもしれません

2023年7月25日火曜日

rqschedulerでワーカーとスケジューラをスクリプトとして定義する方法

rqschedulerでワーカーとスケジューラをスクリプトとして定義する方法

概要

rq worker コマンドと rqscheduler コマンドを Python スクリプトとして定義する方法を紹介します

環境

  • Python 3.11.3
    • rq 1.15.0
    • rq-scheduler 0.13.1

ワーカーサンプルコード

rq worker コマンドの代用スクリプトです

  • vim ./worker.py
from redis import Redis
from rq.connections import Connection
from rq_scheduler import Scheduler

redis_conn = Redis(host="localhost", port=6379)
scheduler = Scheduler(connection=redis_conn)


if __name__ == "__main__":
    with Connection(connection=scheduler.connection):
        scheduler.run()

スケジューラサンプルコード

rqscheduler コマンドの代用スクリプトです

  • vim ./scheduler.py
from redis import Redis
from rq.queue import Queue
from rq.worker import Worker

redis_conn = Redis(host="localhost", port=6379)
queue = Queue("default", connection=redis_conn)


if __name__ == "__main__":
    worker = Worker(queues=[queue], connection=redis)
    worker.work(with_scheduler=True)

注意事項

実行する際はワーカー、スケジューラともに PYTHONPATH を設定してあげましょう

  • PYTHONPATH=./ pipenv run python ./worker.py
  • PYTHONPATH=./ pipenv run python ./scheduler.py

そうしないとジョブを登録するときとスケジューリングするときに実行する関数が見つからず NoModuleNameError になってしまいます

最後に

これで更に Worker や Scheduler クラスを継承することで独自のワーカークラスやスケジューラ実行クラスが作成できます

例えば複数のキューを扱うスケジューラを作成したり接続先の redis を動的に変更したりすることができます

参考サイト

2023年7月19日水曜日

Python でシングルトンクラスを実装する方法

Python でシングルトンクラスを実装する方法

概要

classmethod を使います

環境

  • macOS 13.4.1
  • Python 3.11.3

サンプルコード

  • vim ./app.py
class SingletonClass:
    _instance = None

    @classmethod
    def get_instance(cls):
        if not cls._instance:
            cls._instance = SingletonClass()
        return cls._instance


if __name__ == "__main__":
    instance1 = SingletonClass.get_instance()
    instance2 = SingletonClass.get_instance()
    print(instance1 is instance2)  # True

2023年7月7日金曜日

golangのlsp設定方法(emacs編)

golangのlsp設定方法(emacs編)

概要

gopls という公式が出している language server があるのでそれを使います
基本は公式のドキュメントを参考にすれば OK です

環境

  • Ubuntu 18.04
  • golang 1.20.5
  • gopls 0.12.4
  • emacs 27.1
    • lsp-mode 20230625

gopls のインストール

必要に応じて PATH を通しましょう

  • export PATH=$PATH:/home/user01/go/bin/

.emacs の編集

クライアントは lsp-mode を使っています
go-mode をフックして lsp を起動するだけです
golsp に対応しているバージョンを使いましょう

; for golang
(require 'lsp-mode)
(add-hook 'go-mode-hook #'lsp-deferred)

(defun lsp-go-install-save-hooks ()
  (add-hook 'before-save-hook #'lsp-format-buffer t t)
  (add-hook 'before-save-hook #'lsp-organize-imports t t))
(add-hook 'go-mode-hook #'lsp-go-install-save-hooks)

動作確認

念のためシェルを抜けて gopls がちゃんとコマンドとして起動することを確認してから emacs を起動しましょう
あとは go ファイルを開くだけで OK です

参考サイト

2023年7月6日木曜日

Python で dict to xml する方法

Python で dict to xml する方法

概要

json2xml を使うと簡単です

環境

  • macOS 13.4
  • Python 3.11.3
    • json2xml 3.21.0

サンプルコード

from json2xml import json2xml
from json2xml.utils import readfromurl

data = readfromurl("https://request-dumper.kakakikikeke.com")
j2x = json2xml.Json2xml(data, wrapper="hoge")
print(j2x.data)  # dict を出力
print(j2x.to_xml())  # xml (str) を出力

data には dict を渡せばいいので dict to xml とも言えます

2023年7月5日水曜日

Python fluent-logger 超入門

Python fluent-logger 超入門

概要

python から fluentd にメッセージを送信してみました
簡単なサンプルコードを紹介します

環境

  • macOS 13.4.1
  • Python 3.11.3
  • fluentd 1.16.1

準備

  • pipenv install fluent-logger

fluentd 起動

とりあえずメッセージを受け取り標準出力に表示するだけの fluentd を起動します

  • vim fluent.conf
<source>
  type forward
  port 24224
</source>

<match app.**>
  type stdout
</match>
  • docker run -it --rm -v $(pwd):/fluentd/etc -p 24224:24224 -p 24224:24224/udp fluent/fluentd:edge-debian-arm64 -c /fluentd/etc/fluent.conf

AppleSillicon なので arm64 版のイメージを起動しています

サンプルコード

起動した fluentd にメッセージを送信するサンプルコードです
タグは app.* で指定します

  • vim ./app.py
from fluent import sender

logger = sender.FluentSender('app', host='localhost', port=24224)
logger.emit('follow', {'from': 'userA', 'to': 'userB'})

動作確認

  • pipenv run python app.py

で fluentd 側のログに以下のようなログが表示されることを確認します

2023-06-30 05:55:55.000000000 +0000 app.follow: {"from":"userA","to":"userB"}

最後に

Python から fluentd に直接メッセージを送信してみました
ちょっとユースケースが思い浮かばないのですがログ以外でメッセージブローカとして fluentd を使うケースなどには使えそうです

2023年7月4日火曜日

requests + Thread する場合は必ず close するようにしなければならない

requests + Thread する場合は必ず close するようにしなければならない

概要

そうしないと ConnectionError や Too many open files になり途中で必ず終了してしまいます

環境

  • macOS 13.4.1
  • Python 3.11.3

サンプルコード

今回は Session を使っています
Session を使う場合は必ずスレッド作成時に Session も作成する必要があります
ポイントは response を必ず閉じるような処理にしている点です

import threading

import requests


class Downloader(threading.Thread):
    def __init__(self, url: str, filename: str):
        super().__init__()
        self.url = url
        self.filename = filename

    def run(self):
        response = None
        try:
            print(f"Target file url: {self.url}")
            with requests.Session() as session:
                response = session.get(self.url)
                if response.status_code == 200:
                    download_path = f"./imgs/{self.filename}"
                    with open(download_path, "wb") as file:
                        file.write(response.content)
                        print(f"\033[0;32mDonload complete: {download_path}\033[00m")
        except Exception:
            if response:
                response.close()


if __name__ == "__main__":
    threads = []
    for filename in ["file1", "file2", "file3"]:
        url = f"http://localhost/{filename}"
        dl = Downloader(url, filename)
        dl.start()
        threads.append(dl)
    for t in threads:
        t.join()

2023年7月3日月曜日

fastapiでフォームデータを受け取る方法

fastapiでフォームデータを受け取る方法

概要

fastapi でフォームデータを取得する方法を紹介します
クラス化して扱う方法も紹介します

環境

  • macOS 13.4.1
  • Python 3.11.3
  • fastapi 0.98.0

インストール

  • pipenv install fastapi
  • pipenv install uvicorn
  • pipenv install python-multipart

とりあえず動かすサンプルコード

  • vim ./app.py
from fastapi import FastAPI, Form

app = FastAPI()


@app.post("/login/")
async def login(username: str = Form(), password: str = Form()):
    return {"username": username}
  • pipenv run uvicorn app:app --reload
  • curl localhost:8000/login/ -XPOST -H "content-type: application/x-www-form-urlencoded" -d "username=hawk&password=pass"

クラスにする

BaseModel を継承してそのクラス内で Form -> str への変換を行うクラスメソッドを定義します
更にそのクラスメソッドを Depends でコールすることで変換します

クラス化することでバリデーションも管理しやすくなります

  • vim ./app.py
from typing import Self

from fastapi import Depends, FastAPI, Form
from pydantic import BaseModel, validator

app = FastAPI()


class UserForm(BaseModel):
    username: str
    password: str

    @classmethod
    def as_form(cls, username: str = Form(...), password: str = Form(...)) -> Self:
        return cls(username=username, password=password)

    @validator("username")
    def validate_username(cls, v):
        if v != "hawksnowlog":
            raise ValueError("Invalid username.")
        return v


@app.post("/login/")
async def login(form_data: UserForm = Depends(UserForm.as_form)):
    return {"username": form_data.username, "password": form_data.password}
  • curl localhost:8000/login/ -XPOST -H "content-type: application/x-www-form-urlencoded" -d "username=hawk&password=pass"

=> InternalError

  • curl localhost:8000/login/ -XPOST -H "content-type: application/x-www-form-urlencoded" -d "username=hawksnowlog&password=pass"

=> 成功

最後に

fastapi でフォームデータを扱ってみました
ちゃんと BaseModel ベースで定義することもできるのでそこまで特殊なことをする必要はなさそうです

参考サイト

2023年7月2日日曜日

ruby sidekiq-scheduler を試す

ruby sidekiq-scheduler を試す

概要

Sidekiq のキューイングを cron のように定期的に実行できるのが sidekiq-scheduler です
今回はインストールから簡単な動作確認までやってみました

環境

  • macOS 13.4
  • redis 7.0.11
  • Ruby 3.2.2
  • sidekiq-scheculer 5.0.3

インストール

  • vim Gemfile
gem 'sidekiq-scheduler'
  • bundle config path vendor
  • bundle install

サンプルコード (ワーカー)

エンキューはスケジューラが実行するため必要なのはワーカーになります
ワーカーは普通の sidekiq のように perform を実装するだけです

  • vim worker.rb
# frozen_string_literal: true

require 'sidekiq-scheduler'

# ワーカークラス
class HelloWorld
  include Sidekiq::Worker

  def perform
    puts 'Hello world'
  end
end

スケジュール設定 (yaml ファイル)

この設定ファイルを元にスケジューラがエンキューします
sidekiq-scheduler の cron 設定は秒単位まで指定できます

  • vim config/sidekiq.yml
:scheduler:
  :schedule:
    hello_world:
      cron: '0 * * * * *'  # 毎分0秒に実行
      class: HelloWorld

スケジューラ起動

あとはワーカーとともにスケジューラを起動するだけです

  • bundle exec sidekiq -r ./worker.rb

動作確認

ワーカーのログに毎分 0 秒のときに以下のようなログが出力されれば OK です

2023-06-12T01:28:09.357Z pid=87946 tid=1x4e INFO: Running in ruby 3.2.2 (2023-03-30 revision e51014f9c0) [arm64-darwin22]
2023-06-12T01:28:09.357Z pid=87946 tid=1x4e INFO: See LICENSE and the LGPL-3.0 for licensing details.
2023-06-12T01:28:09.357Z pid=87946 tid=1x4e INFO: Upgrade to Sidekiq Pro for more features and support: https://sidekiq.org
2023-06-12T01:28:09.357Z pid=87946 tid=1x4e INFO: Sidekiq 7.1.1 connecting to Redis with options {:size=>10, :pool_name=>"internal", :url=>nil}
2023-06-12T01:28:09.359Z pid=87946 tid=1x4e INFO: Sidekiq 7.1.1 connecting to Redis with options {:size=>5, :pool_name=>"default", :url=>nil}
2023-06-12T01:28:09.360Z pid=87946 tid=1x4e INFO: Loading Schedule
2023-06-12T01:28:09.360Z pid=87946 tid=1x4e INFO: Scheduling hello_world {"cron"=>"0 * * * * *", "class"=>"HelloWorld", "queue"=>"default"}
2023-06-12T01:28:09.527Z pid=87946 tid=1x4e INFO: Schedules Loaded
2023-06-12T01:28:09.527Z pid=87946 tid=1x4e INFO: Starting processing, hit Ctrl-C to stop
2023-06-12T01:29:00.036Z pid=87946 tid=1wwa INFO: queueing HelloWorld (hello_world)
2023-06-12T01:29:00.037Z pid=87946 tid=1wwu class=HelloWorld jid=a2522a3f2f3de9a7c706f1f9 INFO: start
Hello world
2023-06-12T01:29:00.038Z pid=87946 tid=1wwu class=HelloWorld jid=a2522a3f2f3de9a7c706f1f9 elapsed=0.001 INFO: done

ちょっと応用: スケジューラの登録を動的に行う

yaml ファイルでスケジュールを管理するのは大変なので ruby から動的に登録することも可能です

  • vim job_create.rb
# frozen_string_literal: true

require 'sidekiq-scheduler'

Sidekiq.set_schedule('hello', { 'every' => ['1m'], 'class' => 'HelloWorld' })

これを起動することでスケジューラを登録できます

  • bundle exec ruby job_create.rb

redis を確認するとスケジューラが登録されています

redis-cli keys '*'
1) "schedules_changed"
2) "schedules"

redis-cli hgetall 'schedules'
1) "hello"
2) "{\"every\":[\"1m\"],\"class\":\"HelloWorker\"}"

また設定ファイルでは dynamic というオプションを true にします

  • vim config/sidekiq.yml
:scheduler:
  :dynamic: true

あとはワーカーを起動して毎分実行されることを確認しましょう

  • bundle exec sidekiq -r ./worker.rb

ワーカーがスケジューラを認識しない場合はサイド job_create.rb を実行してみてください
同一スケジューラ名の場合は新規で追加はせずスケジューラが更新されるだけになります
またスケジューラの変更はワーカーは動的に行ってくれます

最後に

とりあえず動かすところまでやってみました
スケジュールの設定が yaml ファイルなのが特徴的なところかなと思います
また秒単位まで指定できるので細かい時間指定も可能かなと思います

参考サイト

https://github.com/sidekiq-scheduler/sidekiq-scheduler