2023年8月3日木曜日

fastapi で XML なレスポンスを考える (ドキュメント生成編)

fastapi で XML なレスポンスを考える (ドキュメント生成編)

概要

前回まででカスタムレスポンスクラスを扱う方法を紹介しました
ただ openapi にレスポンスの XML 情報が掲載されない問題がありました
今回はその問題に対処してみます

環境

  • Python 3.11.3
  • fastapi 0.100.1
  • pydantic-xml 1.0.0

サンプルコード

  • vim ./app.py
import textwrap
from typing import Union

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")


class XmlResponse(Response):
    def __init__(self, content: Union[str, bytes], status_code: int = 200):
        super().__init__(content, status_code=status_code, media_type="application/xml")


@app.get(
    "/",
    response_class=XmlResponse,
    response_model=User,
    responses={
        200: {
            "description": "Success",
            "content": {
                "application/xml": {
                    "schema": {"$ref": "#/components/schemas/User"},
                    "example": textwrap.dedent(
                        """
                        <?xml version="1.0" encoding="UTF-8"?>
                        <User>
                          <Name>hawk</Name>
                          <Age>20</Age>
                        </User>
                    """
                    )[1:-1],
                },
                "application/json": None,
            },
        },
    },
)
def xml():
    result = User(name="hawksnowlog", age=10)
    return XmlResponse(content=result.to_xml())

解説

openapi のドキュメントの作成は responses 属性を使って行います
content 配下に application/xml を定義し更に schema と example を定義します
schema に直接 Python のクラスを指定すると ValueError: [TypeError("'ModelField' object is not iterable"), TypeError('vars() argument must have __dict__ attribute')] というエラーになるので openapi に出力された際にスキーマを参照する形式で指定します

そして example はヒアドキュメントで直接していします
ここが dict などにしていしまうとうまく XML が表示されないので注意が必要です

application/json のレスポンスはデフォルトで表示されてしまうので不要であれば None を指定します

動作確認

  • pipenv run uvicorn app:app --reload

で localhost:8000/docs にアクセスると以下のようにちゃんとレスポンスのサンプルが XML で表示されているのが確認できると思います

最後に

fastapi でXML のレスポンスを扱う場合には最終的には以下の組み合わせがいいかなと思います

  • モデルの表現 -> pydantic-xml
  • レスポンスの表現 -> Response クラスを継承して XmlResponse クラスを作成する
  • ドキュメントの表現 -> responses 属性を使って content に直接 application/xml のレスポンスを定義する

参考サイト

2023年8月2日水曜日

fastapi で XML なレスポンスを考える (カスタムレスポンス編)

fastapi で XML なレスポンスを考える (カスタムレスポンス編)

概要

前回 pydantic-xml を使ってレスポンスの構造を XML で表現してみました

今回は Response ではなく独自で定義した Response クラスを使ってレスポンスを生成してみたいと思います

環境

  • Python 3.11.3
  • fastapi 0.100.1
  • pydantic-xml 1.0.0

サンプルコード

  • vim ./app.py
from typing import Optional, Union

from fastapi import FastAPI, Response
from pydantic_xml import BaseXmlModel, element
from starlette.background import BackgroundTask

app = FastAPI()


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


class XmlResponse(Response):
    def __init__(
        self,
        content: Union[str, bytes],
        status_code: int = 200,
        background: Optional[BackgroundTask] = None,
    ):
        super().__init__(
            content,
            status_code=status_code,
            media_type="application/xml",
            background=background,
        )


@app.get("/", response_class=XmlResponse, response_model=User)
def xml():
    user = User(name="hawksnowlog", age=10)
    return XmlResponse(content=user.to_xml())

動作確認

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

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

解説

Response を継承した XmlResponse を定義しています
XmlResponse は response_class に指定できます
また response_model に User も指定できます

json を返却したい場合には以下のように response_class の指定を外せば OK です

@app.get("/", response_model=User)
def xml():
    user = User(name="hawksnowlog", age=10)
    return user

最後に

カスタム Response を使って XML レスポンスを返却する方法を紹介しました
ただこの方法でも openapi に載らないので結局 responses 属性を指定することになります

XmlResponse を使った場合に openapi にレスポンス情報を掲載するにはどうすればいいだろうか

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