2023年9月29日金曜日

pydanticV2 で自作クラス側の field_validator に info.data が渡らない問題をどう解決するか

pydanticV2 で自作クラス側の field_validator に info.data が渡らない問題をどう解決するか

概要

以下にサンプルコードを掲載しますが pydanticV2 を使っていて自作クラスを型に指定しその自作クラス側で FieldValidationInfo を使ってバリデーションした場合に info.data が空になるという問題があります

解決する方法としてクラスをまとめるという方法がありますがそれだとクラスが膨大になってしまいます
今回はその場合の解決策を紹介します

環境

  • Python 3.11.3
  • fastapi 0.100.1
  • pydantic 2.1.1

サンプルコード (info.data が空になる)

以下の場合 name -> age という順番でバリデーションされます
pydantic V2 の場合、バリデーションを通過した値はすべて FieldValidationInfo の info.data というオブジェクトに格納されます

しかし以下のように独自クラス側に age を分けている場合に info.data の値が空になるためバリデーションに成功した name を参照することができません

これを解決する方法として Age クラスを使わずに User 側にバリデーションを寄せる方法がありますがそれだとクラスが大きくなり微妙です

from pydantic import BaseModel, Field, FieldValidationInfo, field_validator


class Age(BaseModel):
    age: int = Field()

    @field_validator("age")
    @classmethod
    def validate_age(cls, v: int, info: FieldValidationInfo):
        # print(info.data["name"]) ## KeyError になる
        if info.context is None:
            return v
        if "age_limit" not in info.context or not isinstance(
            info.context["age_limit"], int
        ):
            raise ValueError("info.context.age_limit が正しく設定されていません")
        print("validate_age_with_context")
        limit = int(info.context["age_limit"])
        if v > limit:
            raise ValueError(f"age が limit={limit} を超えています")
        return v


class User(BaseModel):
    name: str = Field()
    age: Age = Field()

    @field_validator("name")
    @classmethod
    def validate_name(cls, v: str, info: FieldValidationInfo):
        if info.context is None:
            return v
        if "allowed_name" not in info.context or not isinstance(
            info.context["allowed_name"], str
        ):
            raise ValueError("info.context.allowed_name が正しく設定されていません")
        print("validate_name")
        if v not in [info.context["allowed_name"]]:
            raise ValueError()
        return v


if __name__ == "__main__":
    user = User.model_validate(
        {"name": "hawksnowlog", "age": {"age": 0}}, context={"allowed_name": "hawksnowlog", "age_limit": 9}
    )

解決方法 (context を無理やり使う

自分が発見した方法は info.context を使う方法です
これをすることで info.data ではなく info.context を Age 側で参照することで検証済みの name を使うことができます

from pydantic import BaseModel, Field, FieldValidationInfo, field_validator


class Age(BaseModel):
    age: int = Field()

    @field_validator("age")
    @classmethod
    def validate_age(cls, v: int, info: FieldValidationInfo):
        # info.context を使うように変更
        if info.context is None:
            raise ValueError()
        print(info.context["validated_name"])
        if info.context is None:
            return v
        if "age_limit" not in info.context or not isinstance(
            info.context["age_limit"], int
        ):
            raise ValueError("info.context.age_limit が正しく設定されていません")
        print("validate_age_with_context")
        limit = int(info.context["age_limit"])
        if v > limit:
            raise ValueError(f"age が limit={limit} を超えています")
        return v


class User(BaseModel):
    name: str = Field()
    age: Age = Field()

    @field_validator("name")
    @classmethod
    def validate_name(cls, v: str, info: FieldValidationInfo):
        if info.context is None:
            return v
        if "allowed_name" not in info.context or not isinstance(
            info.context["allowed_name"], str
        ):
            raise ValueError("info.context.allowed_name が正しく設定されていません")
        print("validate_name")
        if v not in [info.context["allowed_name"]]:
            raise ValueError()
        return v

    # mode=before にすることで Field() に記載されている評価よりも前にこっちを実行させる
    @field_validator("age", mode="before")
    @classmethod
    def validate_age(cls, v: Age, info: FieldValidationInfo):
        if info.context is None:
            raise ValueError()
        # 同じクラス内では info.data が引き継がれるので検証済みの name が参照できる
        info.context["validated_name"] = info.data["name"]
        Age.model_validate(v, context=info.context)
        return v


if __name__ == "__main__":
    user = User.model_validate(
        {"name": "hawksnowlog", "age": {"age": 0}},
        context={"allowed_name": "hawksnowlog", "age_limit": 9},
    )

最後に

他に自作クラス側に info.data を渡す方法はあるのだろうか

info.data の説明を見ると同一クラス内で参照可能とあるので Age クラスと User が継承するようにすれば参照できるようになるはず

なので継承しない独自クラスを作る場合には別のモデルクラスとフィールドが相互参照するようなモデルの設計をしないほうが無難な気がする

2023年9月28日木曜日

pydanticV2 の model_validate は逐次評価ではなくすべてを評価してから ValidationError が raise される

pydanticV2 の model_validate は逐次評価ではなくすべてを評価してから ValidationError が raise される

概要

特に context (FiledInfoValidator) と組み合わせる場合には注意が必要かなと思います

結論としてはエラーとなりそうなところがあればちゃんと事前に None チェックをしましょうという話です

環境

  • Python 3.11.3
  • fastapi 0.100.1
  • pydantic 2.1.1

サンプルコード

from fastapi import FastAPI
from pydantic import BaseModel, Field, FieldValidationInfo, field_validator

app = FastAPI()


class User(BaseModel):
    name: str = Field()
    age: int = Field()

    @field_validator("name")
    @classmethod
    def validate_name(cls, v: str, info: FieldValidationInfo):
        if info.context is None:
            return v
        print("validate_name")
        if v not in ["hawksnowlog"]:
            raise ValueError()
        return v

    @field_validator("age")
    @classmethod
    def validate_age_with_context(cls, v: int, info: FieldValidationInfo):
        if info.context is None:
            return v
        if "age_limit" not in info.context or not isinstance(
            info.context["age_limit"], int
        ):
            raise ValueError("info.context.age_limit が正しく設定されていません")
        # ここがポイント
        # 前の validate_name で ValueError になると raise されない
        # 更に info.data["name"] には何も設定されずに来るのでここで KeyError が発生してしまう
        if info.data["name"] not in ["hawksnowlog"]:
            raise ValueError()
        # もしやるなら以下のように None チェックしてから info.data["name"] を扱うようにする
        if info.data.get("name") is None:
            raise ValueError()
        if info.data["name"] not in ["hawksnowlog"]:
            raise ValueError()
        print("validate_age_with_context")
        limit = int(info.context["age_limit"])
        if v > limit:
            raise ValueError(f"age が limit={limit} を超えています")
        return v


@app.post("/")
async def test_context(user: User):
    # ルーティング時の User 生成時にコールされるバリデーションはどこかで raise すればエラーになる
    # model_validate の場合は raise 時にエラーとならずすべてが評価されたあとで raise される
    user = User.model_validate(
        {"name": "hawksnowlog2", "age": 0}, context={"age_limit": 9}
    )
    return user

ちょっと解説

基本はコードのコメント部分に書いてあるとおりです

model_validate と context を使ってバリデーションした場合には途中で raise ValueError したものが後ろに回されてまとめて raise されます
なのでバリデーションのコードの中で KeyError など予期せぬエラーが発生した場合にはそこでバリデーションが止まってしまい本当に取得したい ValidationError が取得できなくなってしまいます

なので None 参照や Key 参照をする場合はちゃんと存在するかチェックした上でバリデーションを組む必要があります

2023年9月27日水曜日

SQLAlchemy でクエリキャッシュを使う方法 (dogpile.cache編)

SQLAlchemy でクエリキャッシュを使う方法 (dogpile.cache編)

概要

SQLAlchemy は以前 Beker cache という仕組みを使ってクエリキャッシュしていました
しかし SQLAlchemy 2.0 の現在ではその方法は廃止され dogpile.cache という汎用キャッシュの仕組みを使ってクエリキャッシュするのが王道になっています

前回 dogpile.cache の簡単な使い方を紹介しました
今回は SQLAlchemy と組み合わせて使ってみます

環境

  • macOS 13.5.2
  • Python 3.11.5
    • dogpile.cache 1.2.2
    • SQLAlchemy 2.0.21

サンプルコード

from dogpile.cache import make_region
from sqlalchemy import Column, Integer, String, create_engine
from sqlalchemy.orm import Session, declarative_base, sessionmaker

engine = create_engine("mysql+pymysql://root@localhost/test?charset=utf8mb4")
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()


class User(Base):
    __tablename__ = "user"

    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    age = Column(Integer)


# Dogpileキャッシュリージョンを作成
cache_region = make_region().configure(
    "dogpile.cache.memory",  # メモリキャッシュを使用しますが、他のバックエンドも利用可能です
    expiration_time=3600,  # キャッシュの有効期限を設定(秒単位)
)


# SQLAlchemyセッションをキャッシュ対象にする
@cache_region.cache_on_arguments()
def get_user_by_id(user_id):
    db: Session = SessionLocal()
    print("Querying database for user:", user_id)
    result = db.query(User).filter_by(id=user_id).first()
    db.close()
    return result


# ユーザーを取得し、キャッシュを利用
cached_user = get_user_by_id(1)
print("User 1:", cached_user.name)

# 同じユーザーを再度取得し、キャッシュを利用
cached_user = get_user_by_id(1)
print("User 1 (cached):", cached_user.name)

ちょっと解説

dogpile.cache は同一パラメータを使った関数呼び出しの場合はキャッシュを参照するようになります
なのでその仕組を使ってクエリを投げる関数を作成しその関数に対して cache_on_arguments アノテーションを付与することでキャッシュ化することができます

2回関数をコールしていますが2回目はキャッシュを参照しているため「Querying database for user: 1」が表示されないのが確認できます

こんな感じで既存のクエリを発行する関数を簡単にキャッシュ化することができます

最後に

dogpile.cache を使って SQLAlchemy のクエリキャッシュをする方法を紹介しました
アノテーション付与だけで使えるのでかなり簡単です

dogpile.cache には他にもキャッシュ戦略やキャッシュを特定するキーの生成方法などいろいろとカスタマイズすることができるので興味があれば参考サイトにある公式ドキュメントを見ることをおすすめします

参考サイト

2023年9月26日火曜日

dogpile.cache 超入門

dogpile.cache 超入門

概要

dogpile.cache は SQLAlchemy のクエリキャッシュのために作られたライブラリでその名の通りクエリキャッシュに使います
しかし dogpile.cache 自体は単なる関数キャッシュとしても使えるので今回は SQLAlchemy なしで簡単に動かせるサンプルコードを紹介します

環境

  • macOS 13.5.2
  • Python 3.11.5
  • dogpile.cache 1.2.2

サンプルコード

from dogpile.cache import make_region

# Dogpileキャッシュリージョンを作成
cache_region = make_region().configure(
    "dogpile.cache.memory",  # メモリキャッシュを使用しますが、他のバックエンドも利用可能です
    expiration_time=3600,  # キャッシュの有効期限を設定(秒単位)
)


# キャッシュする関数を定義
@cache_region.cache_on_arguments()
def expensive_function(x, y):
    print("Function executed!")
    return x + y


# 関数を呼び出してキャッシュを利用
result1 = expensive_function(2, 3)
print("Result 1:", result1)  # この行が実行されると、関数が実際に呼び出されます

# 同じ引数で関数を再度呼び出し
result2 = expensive_function(2, 3)
print("Result 2:", result2)  # この行はキャッシュから結果を取得します

# 別の引数で関数を呼び出し
result3 = expensive_function(4, 5)
print("Result 3:", result3)  # この行が実行されると、新しい結果がキャッシュされます

ちょっと解説

result1 と result3 は関数を実行しています
result2 は関数を実行せずキャッシュ (メモリ) から結果を取得しているため「Function executed!」が表示されないのが確認できます

またdogpile.cache.memoryでキャッシュ先のバックエンドを指定することもできます
今回はメモリを使っていますが他にも redis や memcached などいろいろなバックエンドがキャッシュ先として使えます

最後に

アノテーションを一つ付けるだけで簡単に関数の結果をキャッシュすることができます
大量にコールされる関数でかつ結果が非同期でも問題ない場合は使えるかなと思います
結果が非同期というかどんな結果でも問題ないような関数の場合には使えると思います

次回は SQLAlchemy と組み合わせて使ってみたいと思います

参考サイト

2023年9月25日月曜日

PydanticV2 の FieldValidationInfo にはデータベースへのセッションも渡すことができる

PydanticV2 の FieldValidationInfo にはデータベースへのセッションも渡すことができる

概要

前回 FieldValidationInfo の基本的な使い方を紹介しました

今回は db のセッションが渡せることが判明したのでそのサンプルコードを紹介します

環境

  • Python 3.11.3
  • fastapi 0.100.1
  • pydantic 2.1.1

サンプルコード

from fastapi import Depends, FastAPI
from pydantic import BaseModel, Field, FieldValidationInfo, field_validator
from sqlalchemy import Column, Integer, String, create_engine
from sqlalchemy.orm import Session, declarative_base, sessionmaker

app = FastAPI()

engine = create_engine("mysql+pymysql://root@localhost/test?charset=utf8mb4")
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()


class UserTable(Base):
    __tablename__ = "user"

    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    age = Column(Integer)


def get_db():
    db: Session = SessionLocal()
    try:
        yield db
        db.commit()
    except Exception:
        db.rollback()
    finally:
        db.close()


class User(BaseModel):
    name: str = Field()
    age: int = Field()

    @field_validator("name")
    @classmethod
    def validate_name(cls, v: str, info: FieldValidationInfo):
        if info.context is not None:
            # context がある場合のバリデーションは無視する
            return v
        print("validate_name")
        if v not in ["hawksnowlog"]:
            raise ValueError()
        return v

    @field_validator("age")
    @classmethod
    def validate_age_with_context(cls, v: int, info: FieldValidationInfo):
        if info.context is None:
            # context がない場合のバリデーションは無視する
            return v
        if "age_limit" not in info.context or not isinstance(
            info.context["age_limit"], int
        ):
            # context の値をチェック、正しくセットされていない場合はエラー
            raise ValueError("info.context.age_limit が正しく設定されていません")
        if "db" not in info.context or not isinstance(info.context["db"], Session):
            # context の値をチェック、正しくセットされていない場合はエラー
            raise ValueError("info.context.db が正しく設定されていません")
        print("validate_age_with_context")
        # db オブジェクトが使えるかテスト
        users = info.context["db"].query(UserTable).all()
        for user in users:
            print(user.name)
        limit = int(info.context["age_limit"])
        if v > limit:
            raise ValueError(f"age が limit={limit} を超えています")
        return v


@app.post("/")
async def test_context(user: User, db: Session = Depends(get_db)):
    # ルーティングで一度 User のバリデーションをコールする
    user_dict = user.model_dump()
    # ここで context を付与して再度バリデーションをコールする
    user = User.model_validate(user_dict, context={"age_limit": 9, "db": db})
    return user

ポイント

ルーティングの箇所で Depdends + get_db でセッションを取得します
取得したセッションは model_validate の context 引数にそのまま渡します
あとは FieldValidationInfo を使う際に存在チェックや型チェックを行ったあとで普通にクエリを発行すればデータベースにアクセスすることができます

最後に

pickle できるオブジェクトじゃないとダメかなと思ったいましたがそんなことはありませんでした

基本的には何でも渡せるのかなと思います

2023年9月21日木曜日

Python の validators でサポートしていないプロトコルの URL をバリデーションする方法

Python の validators でサポートしていないプロトコルの URL をバリデーションする方法

概要

無理やり対応しているプロトコルに変換して validators.url に投げます

環境

  • macOS 13.5.2
  • Python 3.11.3
    • validators 0.22.0

サンプルコード

import validators


def replace_protocol(url):
    if url.startswith("mqtt"):
        return mqtt_url.replace("mqtt", "http")
    elif url.startswith("mqtts"):
        return mqtt_url.replace("mqtts", "https")
    else:
        raise ValueError()
    

mqtt_url = "mqtts://localhost:1884"
http_url = replace_protocol(mqtt_url)

print(validators.url(http_url, simple_host=True, may_have_port=True))

最後に

苦肉の策としてこんな感じで対応していないプロトコルでも URL バリデーションできます
本当は validators の対応を待つのがいいかなと思います

2023年9月20日水曜日

Pythonのvalidatorsを使ってみる

Pythonのvalidatorsを使ってみる

概要

validators ライブラリは複雑な正規表現を使わずにいろいろなフォーマットのバリデーションが行える便利ライブラリです

今回は URL のバリデーションだけですが使って見ました

環境

  • macOS 13.5.2
  • Python 3.11.3
    • validators 0.22.0

インストール

  • pipenv install validators

サンプルコード

成功している場合は True で失敗している場合は False ではなく ValidationError のオブジェクトが返ってきます

まだまだ開発段階ですべての URL スキームに対応しているわけではないので注意してください

import validators

# OK
print(validators.url("https://kakakikikeke.com"))
print(validators.url("http://localhost.dev"))
print(validators.url("http://localhost", simple_host=True))
print(validators.url("http://localhost:8000", simple_host=True, may_have_port=True))

# Error
print(validators.url("localhost"))  # schema がない
print(validators.url("localhost:8000", simple_host=True, may_have_port=True))  # schema がない
print(validators.url("http://localhost:8000"))  # ポートが指定されている (my_have_port=True がない場合
print(validators.url("mqtt://localhost:1883", simple_host=True, may_have_port=True))  # mqtt スキームはサポートしていない https://github.com/python-validators/validators/blob/master/src/validators/url.py#L41

最後に

他にもいろいろなバリデーションに対応しています
メールフォーマットをチェックした IP アドレスフォーマットのチェック、ホスト名チェックなども行えます

正規表現を愚直に書くよりかはこちらに寄せるほうがコードとしては読みやすくなるかなと思います

そのうち uri というメソッドができるのでそっちでいろいろなスキームに対応できるかなと思います 参考

参考サイト

2023年9月14日木曜日

当サイトにおけるプライバシポリシーおよび免責事項など

当サイトにおけるプライバシポリシーおよび免責事項など

プライバシーポリシー

個人情報の利用目的

当サイトでは、お問い合わせや記事へのコメントの際、名前やメールアドレス等の個人情報を入力いただく場合がございます。 取得した個人情報は、お問い合わせに対する回答や必要な情報を電子メールなどをでご連絡する場合に利用させていただくものであり、これらの目的以外では利用いたしません。

広告について

当サイトでは、第三者配信の広告サービス(Googleアドセンス)を利用しており、ユーザーの興味に応じた商品やサービスの広告を表示するため、クッキー(Cookie)を使用しております。 クッキーを使用することで当サイトはお客様のコンピュータを識別できるようになりますが、お客様個人を特定できるものではありません。

Cookieを無効にする方法やGoogleアドセンスに関する詳細は「広告 – ポリシーと規約 – Google」をご確認ください。

また、当サイトは、Amazon.co.jpを宣伝しリンクすることによってサイトが紹介料を獲得できる手段を提供することを目的に設定されたアフィリエイトプログラムである、Amazonアソシエイト・プログラムの参加者です。

アクセス解析ツールについて

当サイトでは、Googleによるアクセス解析ツール「Googleアナリティクス」を利用しています。このGoogleアナリティクスはトラフィックデータの収集のためにクッキー(Cookie)を使用しております。トラフィックデータは匿名で収集されており、個人を特定するものではありません。

コメントについて

全てのコメントは管理人が事前にその内容を確認し、承認した上での掲載となります。あらかじめご了承ください。

免責事項

当サイトからのリンクやバナーなどで移動したサイトで提供される情報、サービス等について一切の責任を負いません。

また当サイトのコンテンツ・情報について、できる限り正確な情報を提供するように努めておりますが、正確性や安全性を保証するものではありません。情報が古くなっていることもございます。

当サイトに掲載された内容によって生じた損害等の一切の責任を負いかねますのでご了承ください。

著作権について

当サイトで掲載している文章や画像などにつきましては、無断転載することを禁止します。

当サイトは著作権や肖像権の侵害を目的としたものではありません。著作権や肖像権に関して問題がございましたら、Twitter およびコメントなどでご連絡ください。迅速に対応いたします。

リンクについて

当サイトは基本的にリンクフリーです。リンクを行う場合の許可や連絡は不要です。

ただし、インラインフレームの使用や画像の直リンクはご遠慮ください。

2023年9月13日水曜日

Pythonでサロゲートペア文字が含まれているか確認する方法

Pythonでサロゲートペア文字が含まれているか確認する方法

概要

今回は文字数を元にチェックしてみました

環境

  • macOS 13.5.2
  • Python 3.11.3

サンプルコード

def has_surrogate(s: str) -> bool:
    for ts in s:
        if len(ts.encode('utf-16-be')) == 2:
            continue
        else:
            return True
    return False


print(has_surrogate("𠀋"))  # True
print(has_surrogate("𡗗𦰩"))  # True
print(has_surrogate("Hello😀"))  # True
print(has_surrogate("含まれていない"))  # False
print(has_surrogate("含"))  # False
print(has_surrogate("Hello"))  # False

最後に

この方法だと2バイト文字列以外は全部弾くので 1 or 3 以上の文字も弾いてしまいます

正規表現でチェックする方法はないだろうか

参考サイト

2023年9月12日火曜日

fastapi で pydantic V2 の FieldValidationInfo (context) を使う方法

fastapi で pydantic V2 の FieldValidationInfo (context) を使う方法

概要

fastapi の BaseModel を使ってリクエストモデルを定義している場合に field_validator は自動でコールされます
ただ context (FieldValidationInfo) を使った場合には context が必ず None になってしまうので fastapi では context が使えません

今回は fastapi で context を使ったバリデーションを実現する方法を考えます

環境

  • Python 3.11.3
  • fastapi 0.100.1
  • pydantic 2.1.1

サンプルコード

from fastapi import FastAPI
from pydantic import BaseModel, Field, FieldValidationInfo, field_validator

app = FastAPI()


class User(BaseModel):
    name: str = Field()
    age: int = Field()

    @field_validator("name")
    @classmethod
    def validate_name(cls, v: str, info: FieldValidationInfo):
        if info.context is not None:
            # context がある場合のバリデーションは無視する
            return v
        print("validate_name")
        if v not in ["hawksnowlog"]:
            raise ValueError()
        return v

    @field_validator("age")
    @classmethod
    def validate_age_with_context(cls, v: int, info: FieldValidationInfo):
        if info.context is None:
            # context がない場合のバリデーションは無視する
            return v
        if "age_limit" not in info.context or not isinstance(
            info.context["age_limit"], int
        ):
            # context の値をチェック、正しくセットされていない場合はエラー
            raise ValueError("info.context.age_limit が正しく設定されていません")
        print("validate_age_with_context")
        limit = int(info.context["age_limit"])
        if v > limit:
            raise ValueError(f"age が limit={limit} を超えています")
        return v


@app.post("/")
async def test_context(user: User):
    # ルーティングで一度 User のバリデーションをコールする
    user_dict = user.model_dump()
    # ここで context を付与して再度バリデーションをコールする
    user = User.model_validate(user_dict, context={"age_limit": 9})
    return user

ちょっと解説

ルーティングのリクエストモデルを受け取るところで一度各種バリデーションがコールされます
そしてルーティング内で model_validate することで再度バリデーションをコールします
そしてここで context を付与してコールします

バリデーション側では context が設定されているかどうかでバリデーションを実行するか判断しています
context が不要なバリデーションを二度コールしても一度通っているので必ず成功するため実行しても問題ないですが処理効率的に二回実行しても意味がないのでスルーするようにしています

最後に

これ以外に context ありで fastapi からバリデーションを実行する方法があるのだろうか

参考サイト

2023年9月8日金曜日

書籍を自炊する手順

書籍を自炊する手順

概要

物理的な本を電子化する手順を紹介します
最終的に kindle のパーソナルドキュメントで管理します

環境

  • macOS 13.5.1
  • iPhone
  • Amazon アカウント

本を裁断

まずは裁断します
何でも OK ですがカッターと定規でも簡単にできます

こんな感じでバラバラになれば OK です

各ページを電子化

電子化していきます
スキャナでスキャンするときれいに電子化できます
面倒な場合は写真でも大丈夫です

これが一番大変な作業かなと思います
参考書など 400 ページ以上ある場合には両面電子化する必要があるので場合によっては 800 ファイルほど作成する必要があります

カメラを使う場合は三脚などで固定して紙だけ特定の位置におけばいいような感じで撮影するとやりやすいかもしれません

自分は 500 枚ほどを 1 時間で撮影しました

Mac 上で結合

電子化したデータを1つのファイルに結合します
画像ファイルの結合は簡単で cat すれば OK です

  • cat 0.jpg 1.jpg 2.jpg > book.png

HEIC で Mac 上の場合はファインダーからファイルを選択してクイックアクションから PDF 作成でも OK です 参考

圧縮

ただしページが多くファイルサイズが最終的に 25MB を超えるようであれば各ページを圧縮する必要があります
その場合は imagemagick などを使って各ページファイルを圧縮しましょう

  • for file in *.HEIC; do magick convert -quality 70 -define jpeg:extent=50KB -resize 20% "$file" jpg/"$file".jpg; don

これで各ファイルが50KBあたりまで小さくなります
ファイル容量を下げてファイルサイズがそのままだとノイズがすごいので resize でファイルサイズも調整します

それでも25MB以内に収まらない場合は pdf 化するときに pdf を分割してあげましょう

pdf 化

結合した画像ファイルを pdf に変換します

Mac であれば先程紹介したようにクイックアクションから簡単にできます

パーソナルドキュメントに送信

Kindle のパーソナルドキュメントはメールで pdf を送信することで登録できます ここ にパーソナルドキュメントのアドレスが表示されていると思うので確認してください
またメール送信したパーソナルドキュメントを Kindle のライブラリに保存するオプションがあるので有効にしておきましょう

パーソナルドキュメントは50MBまで対応しています
送信側のメーラが50MBまで対応しているのであればいいのですが Gmail では添付は最大25MBなので注意しましょう

最後に

物理本を電子化する方法を紹介しました
この場合だと kindle 上ではただの pdf なので目次やリンクなどはありません
もし目次なども作成したい場合は epub 形式などにする必要がありますが結構たいへんなので今回の方法は簡易的にやる方法になります

2023年9月5日火曜日

rq-scheduler の enqueue_in を使ってみた

rq-scheduler の enqueue_in を使ってみた

概要

enqueue_in はスケジューラの起点の時間から指定時間後にジョブをスケジューリングする機能です

具体的にどんな動きになるのか試してみました

環境

  • Python 3.11.3
  • flask-rq2 18.3
  • rq-scheduler 0.13.1

サンプルコード

from datetime import timedelta

from flask import Flask
from flask_rq2 import RQ

app = Flask(__name__)
app.config["RQ_REDIS_URL"] = "redis://localhost:6379/0"

rq = RQ(app)


class User:
    def __init__(self, name) -> None:
        self.name = name

    def say(self, msg):
        print(msg)


@app.route("/register")
def register_job():
    scheduler = rq.get_scheduler(queue="bar")
    job = scheduler.enqueue_in(timedelta(seconds=5), User("hawksnowlog").say, "HELLO")
    return job.get_id()

動作確認

まずはスケジューラを起動します

  • pipenv run rqscheduler
13:40:50 Registering birth

ここの時間が起点時間になります
次にワーカーを起動します

  • pipenv run rq worker bar

あとはアプリを起動しジョブを登録します

  • pipenv run flask run
  • curl localhost:5000/register

するとワーカーのジョブに起点時間から次の 1 分後にログが流れることが確認できると思います

13:41:50 bar: say('HELLO') (0378cd1a-274f-46f7-8484-00d513b98413)
HELLO
13:41:50 bar: Job OK (0378cd1a-274f-46f7-8484-00d513b98413)
13:41:50 Result is kept for 500 seconds

つまり timedelta の最小値は 1min になることがわかります
1min 以下を指定しても結局次の 1 分後にジョブが実行されることになります

最後に

もし秒単位で実行したい場合は少しトリッキーな使い方をする必要があるようです https://github.com/rq/rq-scheduler/issues/74

もしくは素直に rq の機能の enqueue を使う感じかなと思います

また enqueue_in を使った場合は redis 側への結果の保存は 500sec がデフォルトになるのでそのうちキーごと削除されます

参考サイト

2023年9月4日月曜日

rq のジョブに独自のメタ情報を付与する方法

rq のジョブに独自のメタ情報を付与する方法

概要

ジョブ登録時に meta 引数を指定することができます
辞書で好きな値を指定できます
pickle するので pickle 可能なオブジェクトにしましょう

環境

  • Python 3.11.3
  • flask-rq2 18.3
  • rq-scheduler 0.13.1

サンプルコード

from flask import Flask, request
from flask_rq2 import RQ

app = Flask(__name__)
app.config["RQ_REDIS_URL"] = "redis://localhost:6379/0"

rq = RQ(app)


def hello():
    raise Exception()


@app.route("/")
def get_jobs():
    # 検索する id をクエリストリングで渡せる
    id = request.args.get("id")
    scheduler = rq.get_scheduler(queue="bar")
    jobs_ = scheduler.get_jobs()
    # id が設定されていない場合は全件取得、そうでない場合は meta 情報を使って id 検索
    if id is None:
        jobs = jobs_
    else:
        jobs = [
            j for j in jobs_ if j.meta.get("id") is not None and j.meta.get("id") == id
        ]
    return [
        {
            "id": job.id,
            "meta": job.meta.get("id"),
        }
        for job in jobs
    ]


@app.route("/register")
def register_job():
    # id を meta 情報として設定する
    id = request.args.get("id")
    scheduler = rq.get_scheduler(queue="bar")
    job = scheduler.cron(
        "*/1 * * * *",
        func=hello,
        args=[],
        meta={"id": id},  # meta 属性に辞書を設定可能
    )
    return job.get_id()

最後に

ジョブを特定する際に使えると思います
ただ meta 情報で直接検索する機能がないので全件取得してから検索するしかないので redis のデータが膨大になる場合にはパフォーマンスなどの注意が必要です

参考サイト

2023年9月1日金曜日

dicttoxml で namespace を付与する方法

dicttoxml で namespace を付与する方法

概要

公式的には xmler を使えとのことですが面倒なので直接変更します

環境

  • macOS 13.4.1
  • Python 3.11.3

サンプルコード

from dicttoxml import dicttoxml

dict = {
  "User": {
    "Name": "hawksnowlog",
    "Age": 10,
  }
}

xml = dicttoxml(dict, root=False, attr_type=False)
print(xml.decode("utf-8").replace("<User>", '<User xmlns:mx="http://schema.url.goes.here">'))

参考サイト