Python

こんにちは、ナナオです。 以前の記事で図書館予約管理システムの構築と自宅k8sサーバーの構築を行いました。 Seleniumで図書館の予約管理システムを作った streamlitで表を出力した おうちでk3sを使ってk8sを構築する 今回は構築したk8s環境に予約システムをデプロイしようと思います。 実装 まずはDockerイメージのプッシュです。 Dockerレジストリを何にしようか…というのが最初のポイントですが、docker.ioはプライベートリポジトリが1リポジトリしか使えないのでナシ。 ghcr.ioは自分がGithubアカウントがプロなのもあって2GBまでストレージ使えるのと月の通信量が10GBまでなら課金なしなので、ghcr.ioにします。 ということでイメージをビルドしていきます。 docker build -f docker/Dockerfile -t library-checker:0.1.0 . プッシュもやっちゃいましょう。 ここからはこのドキュメントを参照します。 https://docs.github.com/ja/packages/working-with-a-github-packages-registry/working-with-the-container-registry GitHub Container Registry(ghcr)でコンテナイメージを管理する #Docker - Qiita PATを発行し、以下のコマンドを実行します。 echo "<発行したPAT>" | docker login ghcr.io -u <username> --password-stdin ログインに成功したら、以下のコマンドでプッシュします。 docker push ghcr.io/NAMESPACE/IMAGE_NAME:latest 次にイメージのプルに必要な認証情報をクラスタに渡します。 kubectl create secret docker-registry ghcr-secret \ --docker-server=https://ghcr.io \ --docker-username=ユーザー名 \ --docker-password=PAT(ghp_xxxx...) \ --docker-email=メールアドレス また、定義していた.envもシークレットとして登録しておきます。 kubectl create secret generic library-checker-env --from-env-file=.env それでは、システムを動かすためのファイルを実装します。 apiVersion: apps/v1 kind: Deployment metadata: name: library-checker spec: replicas: 1 selector: matchLabels: app: library-checker template: metadata: labels: app: library-checker spec: imagePullSecrets: - name: ghcr-secret # 先ほどクラスタに設定したイメージのプルに必要な認証情報 containers: # --- 1. アプリ本体 (Streamlit等のUIを想定) --- - name: app image: ghcr.io/<ユーザー名>/library-checker:0.1.0 ports: - containerPort: 8501 env: - name: SELENIUM_URL value: "http://localhost:4444" # 同じPod内なのでlocalhostでOK # .envの内容は本来ConfigMapやSecretを使いますが、一旦直接指定か環境変数で対応 envFrom: - secretRef: name: library-checker-env # 先ほどデプロイしたシークレット # 生存確認 readinessProbe: httpGet: path: /healthz port: 8501 initialDelaySeconds: 5 periodSeconds: 5 livenessProbe: httpGet: path: /healthz port: 8501 initialDelaySeconds: 10 periodSeconds: 5 # --- 2. Selenium Chrome (サイドカー) --- - name: chrome image: selenium/standalone-chrome:4.39.0-20251212 ports: - containerPort: 4444 resources: limits: memory: "2Gi" cpu: "1000m" volumeMounts: - name: dshm mountPath: /dev/shm # 共有メモリ (/dev/shm) の不足によるクラッシュを防ぐ設定 volumes: - name: dshm emptyDir: medium: Memory --- # --- 3. 外部公開用サービス --- apiVersion: v1 kind: Service metadata: name: library-checker-service spec: type: NodePort selector: app: library-checker ports: - name: ui port: 8501 targetPort: 8501 nodePort: 30501 # 外部からアクセスするポート - name: selenium port: 4444 targetPort: 4444 nodePort: 30444 今回もGemini先生に頑張ってもらいました。 ...

こんにちは、ナナオです。 前回の記事で図書館の予約管理システムを構築しました。 データ取得まではできましたが、データ出力するところが実装できなかったので今回実装していこうと思います。 React使うか～とかも思ったのですが、Geminiと相談した結果Streamlitというライブラリを使うことにしました。 Streamlitとは グラフデータや表データなどを含むWebアプリを超簡単に構築できるライブラリです。 Streamlit • A faster way to build and share data apps もちろん自由度はReactでフロントエンドを作るよりも低くはなるのですが、大体の機能が揃っているのでとりあえずデータ出力したいだけであればこれで十分だと思います。 実装 ということで実装していきます。 今回は以下のデータクラスを表で出力します。 @dataclass class LoanStatusItem: deadline_at: dt.date borrower_name: str library_name: str book_title: str @dataclass class ReserveItem: status: Literal["reserved", "already"] borrower_name: str library_name: str book_title: str 必要なライブラリを追加します。 uv add streamlit pandas 実装はとりあえずダミーデータを使ってやりました。 def get_dummy_data(): # 貸出状況のデータ loans = [ LoanStatusItem(dt.date(2023, 10, 15), "佐藤", "中央図書館", "Python入門"), LoanStatusItem(dt.date(2023, 10, 20), "鈴木", "北図書館", "Streamlit実践"), LoanStatusItem(dt.date(2023, 10, 12), "佐藤", "中央図書館", "機械学習の基礎"), ] # 予約状況のデータ reserves = [ ReserveItem("reserved", "佐藤", "南図書館", "Web設計パターン"), ReserveItem("already", "田中", "中央図書館", "デザイン思考"), ReserveItem("reserved", "鈴木", "北図書館", "Docker活用"), ] current_time = dt.datetime.now().strftime('%H:%M:%S') return loans, reserves, current_time def main(): st.title("図書館 利用状況ダッシュボード") # --- 更新ボタンの配置 --- # col1, col2 を使うことで、タイトルやボタンの配置を調整できます col1, col2 = st.columns([3, 1]) with col1: st.write("最新の貸出・予約状況を表示します。") with col2: # ボタンが押されたらキャッシュをクリアして再実行 if st.button("最新情報を取得 🔄"): st.cache_data.clear() # キャッシュを削除 st.rerun() # スクリプトを再実行（画面更新） # データの取得 loans, reserves, fetched_time = get_dummy_data() # 更新時刻の表示 st.caption(f"データ取得時刻: {fetched_time}") # --- 1. 貸出状況の表示 --- st.subheader("📅 貸出状況 (Loan Status)") if loans: # dataclassのリストをDataFrameに変換 # dataclasses.asdictを使うと辞書に変換され、DataFrame化しやすくなります df_loans = pd.DataFrame([asdict(item) for item in loans]) # カラム名の見た目を整える（任意） df_loans = df_loans.rename(columns={ "deadline_at": "返却期限", "borrower_name": "利用者名", "library_name": "図書館", "book_title": "書名" }) # テーブル表示 (use_container_width=Trueで横幅いっぱいに表示) st.dataframe(df_loans, use_container_width=True) else: st.info("貸出中の本はありません。") st.markdown("---") # 区切り線 # --- 2. 予約状況の表示 --- st.subheader("🔖 予約状況 (Reserve Status)") if reserves: df_reserves = pd.DataFrame([asdict(item) for item in reserves]) # カラム名の整理 df_reserves = df_reserves.rename(columns={ "status": "状態", "borrower_name": "利用者名", "library_name": "図書館", "book_title": "書名" }) # 状態（reserved/already）に応じて色をつけたりも可能です st.dataframe(df_reserves, use_container_width=True) else: st.info("予約中の本はありません。") if __name__ == "__main__": main() これを実行します。 ...

こんにちは、ナナオです。 最近、いろんな図書館からいろんな本を借りることが多いのですが、書籍の予約管理を今はメモで行っており、ちょっとこれだと見づらいので一元管理できるシステムを作ることにしました。 ということで、早速実装していきましょう。 （Seleniumの詳細な説明は省略します、詳細はこちらから！） 仕様 とりあえず必要な仕様は以下の通りです。最初はシンプルにいきましょう。 WebUIを出力 書籍の予約を一覧で表示する 「名前」と「借りている図書館」を表示する とりあえずこれでいきましょう。 実装にはSeleniumと相性がよく、私がすぐ実装できるpythonを使用します。 実装 とりあえずローカルでseleniumを動作させられる環境を構築します。 まずは土台になるパッケージを作成します。 uvを使います。 uv init --package --project library-checker uv add selenium seleniumの動作環境構築はDockerでやるのが一番楽なので、以下のコマンドを実行します。 docker run -d -p 4444:4444 --rm --shm-size="2g" selenium/standalone-chrome:4.39.0-20251212 （実行イメージはこちらを参考にしました） これでとりあえず動作させる準備が整いました。 seleniumを動作させる基本実装は以下の通りです。 from selenium import webdriver def main(): # Chrome のオプションを設定する options = webdriver.ChromeOptions() options.add_argument('--headless') # Selenium Server に接続する driver = webdriver.Remote( command_executor='http://localhost:4444/wd/hub', options=options, ) # ブラウザを終了する driver.quit() if __name__ == "__main__": main() これで特にエラーなく動作すれば最初のステップは問題ないです。 ...

こんにちは。ナナオです。 アービトラージに興味があり、開発してみたいな～と思いつつなかなか手が出せなかったのですが、この度重い腰をあげて開発してみました。 使用技術 監視にはPrometheus + Grafanaを使用しました。 実装はPythonを使い、HTTPリクエストにrequestsを使っています。 実装 早速実装です。 メインになる実装は以下の通りです。 ここの実装が大体100行くらいです。 import time import logging import threading from itertools import combinations # リクエストに使うAPI（自作） from vc_bot import exchange_api from prometheus_client import start_http_server, Gauge, Counter # 必要に応じてログの設定（これは標準エラー出力に出す設定です） logging.basicConfig(level=logging.ERROR) logger = logging.getLogger(__name__) PROFIT_GAUGE = Gauge("profit", "利益", ["exchange", "symbol"]) PROFIT_RATE_GAUGE = Gauge("profit_rate", "利益率", ["exchange", "symbol"]) REQUEST_ERROR_GAUGE = Counter("request_error", "リクエストエラー", ["exchange"]) def worker(): # 対象の取引所 exchange_pair = { "BTC_JPY": ["gmo", "coincheck", "binance", "bitflyer", "zaif", "bitbank", "okcoin"], "ETH_JPY": ["gmo", "coincheck", "binance", "bitflyer", "zaif", "bitbank", "okcoin"], "XRP_JPY": ['gmo', 'coincheck', 'binance', 'bitflyer', 'bitbank', 'okcoin'], "MONA_JPY": ['coincheck', 'bitflyer', 'zaif', 'bitbank'], } while True: for pair, exchanges in exchange_pair.items(): prices = {} # リクエストが成功した取引所のみ格納 success_exchanges = [] for exchange in exchanges: try: api = getattr(exchange_api, exchange)() prices.update({exchange: api.fetch_ticker(pair=pair)}) success_exchanges += [exchange] except Exception: logger.exception("リクエスト中にエラーが発生しました") REQUEST_ERROR_GAUGE.labels( exchange=exchange, ).inc() continue # 各取引所の比較結果を格納するGaugeオブジェクトを初期化 exchange_combination = list(combinations(success_exchanges, 2)) for exchange1, exchange2 in exchange_combination: ex1 = prices[exchange1] ex2 = prices[exchange2] # パターンA: Ex1で買って(Ask)、Ex2で売る(Bid) profit_a = ex2["bid"] - ex1["ask"] profit_rate_a = (profit_a / ex1["ask"]) * 100 print(f"{pair} {exchange1}-{exchange2} profit : {profit_a}") print(f"{pair} {exchange1}-{exchange2} profit rate: {profit_rate_a}") PROFIT_GAUGE.labels( exchange=f"{exchange1}_{exchange2}", symbol=pair ).set(profit_a) PROFIT_RATE_GAUGE.labels( exchange=f"{exchange1}_{exchange2}", symbol=pair ).set(profit_rate_a) # パターンB: Ex2で買って(Ask)、Ex1で売る(Bid) profit_b = ex1["bid"] - ex2["ask"] profit_rate_b = (profit_a / ex2["ask"]) * 100 print(f"{pair} {exchange2}-{exchange1} profit : {profit_b}") print(f"{pair} {exchange2}-{exchange1} profit rate: {profit_rate_b}") PROFIT_GAUGE.labels( exchange=f"{exchange2}_{exchange1}", symbol=pair ).set(profit_b) PROFIT_RATE_GAUGE.labels( exchange=f"{exchange2}_{exchange1}", symbol=pair ).set(profit_rate_b) def main(): # Prometheus ExporterのHTTPサーバーをポート8000で起動 start_http_server(8000) # メトリクス更新ワーカーを別スレッドで実行 worker_thread = threading.Thread(target=worker, daemon=True) worker_thread.start() print("Prometheus metrics server running on port 8000") # メインスレッドを維持 try: while True: time.sleep(1) except KeyboardInterrupt: print("Exiting.") if __name__ == "__main__": main() exchange_apiの実装は以下の通りです。 ...

どうも、ナナオです。 最近PythonからRustを呼び出す実装をすることがありまして、pyo3にお世話になることがありました。 pyo3のビルドにはmaturinというCLIを使うのですが、これをRustのワークスペース機能と併用できるのかどうか気になったので、検証してみたいと思います。 準備 とりあえず適当にpyo3を使用したライブラリを作ります。 ryeを使っていれば以下のコマンドでmaturinをビルダーに指定したプロジェクトを作成できます。 rye init maturin-workspace-playground --build-system maturin maturinをインストールしていなかったので、以下のコマンドでインストールしておきます。 rye install maturin 作ったプロジェクトに移動して、ワークスペースのメンバーになるプロジェクトを作成しておきます。 cd maturin-workspace-playground mkdir rust && cd rust cargo init --lib app cargo init --lib python-api 作成したプロジェクトをワークスペースのメンバーになるように設定をしていきます。 ルートディレクトリのCargo.tomlを以下のように編集します。 [workspace.package] name = "maturin-workspace-playground" version = "0.1.0" edition = "2021" # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html [workspace] members = ["rust/python-api", "rust/app"] rust/appのCargo.tomlは以下のように編集します。 [package] name = "app" version.workspace = true edition.workspace = true # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html [dependencies] rust/python-apiのCargo.tomlは以下のように編集します。 ...

前回、prisma-client-pythonに入門しました。 しかし、どうやってNodeが動いているのかちょっとわからなかったので調べてみました。 Nodeの実態は？ 現状、グローバルにprismaが使える状態ではないため、普通にprismaコマンドを実行しても実行できません。 % prisma zsh: command not found: prisma しかし、前回作ったパッケージ上であればprismaコマンドは実行できます。 % poetry run prisma This command is only intended to be invoked internally. Please run the following instead: prisma <command> e.g. prisma generate 説明を見るかぎり、ここで実行されているprismaは本家のCLIのラッパーとして実装されているようです。 Prisma Client Python exposes a CLI interface which wraps the Prisma CLI. This works by downloading a Node binary, if you don’t already have Node installed on your machine, installing the CLI with npm and running the CLI using Node. ...

今、自分の中でORMツールとしてアツいのがprismaです。 Prisma | Instant Postgres plus an ORM for simpler db workflows モデル定義もマイグレーションも、ドキュメント読んだり実装してみたりすればするほどかなり使える書き方ができるので、すごく可能性を感じています。 そんなprismaですが、python用のクライアントがあるということで、勉強がてら触ってみたいと思います。 Prisma Client Python pythonプロジェクトの初期化 まずはpoetryでプロジェクトを初期化します。 poetry new prisma-client-python-playground 作成されたプロジェクトに依存性を追加します。 poetry add -D prisma さて、これでいつものprisma CLIが使えるようになりました。 早速Prismaの初期化をしていきましょう。 今回もsqliteを使用していきます。 % poetry run prisma init --datasource-provider sqlite ✔ Your Prisma schema was created at prisma/schema.prisma You can now open it in your favorite editor. Next steps: 1. Set the DATABASE_URL in the .env file to point to your existing database. If your database has no tables yet, read https://pris.ly/d/getting-started 2. Run prisma db pull to turn your database schema into a Prisma schema. 3. Run prisma generate to generate the Prisma Client. You can then start querying your database. More information in our documentation: https://pris.ly/d/getting-started 初期化できました。 ...

markdownをplain textにしたうえで、その文章をginzaでリスト化しました。 必要なライブラリをインストールします。 pip install Markdown beautifulsoup4 ginza ja-ginza 以下のような実装を行いました。 from bs4 import BeautifulSoup from markdown import markdown import re import spacy def markdown_to_text(markdown_string): """マークダウンを平文に変換します""" # md -> html -> text since BeautifulSoup can extract text cleanly html = markdown(markdown_string) # remove code snippets html = re.sub(r'<pre>(.*?)</pre>', ' ', html) html = re.sub(r'<code>(.*?)</code >', ' ', html) # extract text soup = BeautifulSoup(html, "html.parser") text = ''.join(soup.find_all(string=True)).replace("\n", "　") return text def split_sentences(text): nlp = spacy.load('ja_ginza') doc = nlp(text) sentences = [sent.text for sent in doc.sents] return sentences if __name__ == "__main__": markdown_string = """# タイトル ## サブタイトル これは段落です。 これは別の段落です。 """ text = markdown_to_text(markdown_string) print(text) sentences = split_sentences(text) print(sentences) 出力は以下のとおりです。 ...

いつも通り開発していたのに、poetry installを行ったら急に全く動かなくなりました。 そんな場合の対処法について発見したので共有します。 対処法 poetry cacheコマンドを利用してキャッシュの削除を行います。 まず、poetry cache listでキャッシュの一覧を表示します。 次に、先程表示したキャッシュ一覧の名前を使用してpoetry cache clear --all [キャッシュ名]とタイプします。 すべてのキャッシュ一覧に対してこの操作を実行後、poetry installを実施したらうまく動きました。 めでたしめでたし。