classmethod を使います
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
gopls という公式が出している language server があるのでそれを使います
基本は公式のドキュメントを参考にすれば OK です
必要に応じて PATH を通しましょう
クライアントは 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 です
json2xml を使うと簡単です
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 とも言えます
python から fluentd にメッセージを送信してみました
簡単なサンプルコードを紹介します
とりあえずメッセージを受け取り標準出力に表示するだけの fluentd を起動します
<source>
type forward
port 24224
</source>
<match app.**>
type stdout
</match>
AppleSillicon なので arm64 版のイメージを起動しています
起動した fluentd にメッセージを送信するサンプルコードです
タグは app.* で指定します
from fluent import sender
logger = sender.FluentSender('app', host='localhost', port=24224)
logger.emit('follow', {'from': 'userA', 'to': 'userB'})
で fluentd 側のログに以下のようなログが表示されることを確認します
2023-06-30 05:55:55.000000000 +0000 app.follow: {"from":"userA","to":"userB"}
Python から fluentd に直接メッセージを送信してみました
ちょっとユースケースが思い浮かばないのですがログ以外でメッセージブローカとして fluentd を使うケースなどには使えそうです
そうしないと ConnectionError や Too many open files になり途中で必ず終了してしまいます
今回は 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()
fastapi でフォームデータを取得する方法を紹介します
クラス化して扱う方法も紹介します
from fastapi import FastAPI, Form
app = FastAPI()
@app.post("/login/")
async def login(username: str = Form(), password: str = Form()):
return {"username": username}
BaseModel を継承してそのクラス内で Form -> str への変換を行うクラスメソッドを定義します
更にそのクラスメソッドを Depends でコールすることで変換します
クラス化することでバリデーションも管理しやすくなります
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}
=> InternalError
=> 成功
fastapi でフォームデータを扱ってみました
ちゃんと BaseModel ベースで定義することもできるのでそこまで特殊なことをする必要はなさそうです
Sidekiq のキューイングを cron のように定期的に実行できるのが sidekiq-scheduler です
今回はインストールから簡単な動作確認までやってみました
gem 'sidekiq-scheduler'
エンキューはスケジューラが実行するため必要なのはワーカーになります
ワーカーは普通の sidekiq のように perform を実装するだけです
# frozen_string_literal: true
require 'sidekiq-scheduler'
# ワーカークラス
class HelloWorld
include Sidekiq::Worker
def perform
puts 'Hello world'
end
end
この設定ファイルを元にスケジューラがエンキューします
sidekiq-scheduler の cron 設定は秒単位まで指定できます
:scheduler:
:schedule:
hello_world:
cron: '0 * * * * *' # 毎分0秒に実行
class: HelloWorld
あとはワーカーとともにスケジューラを起動するだけです
ワーカーのログに毎分 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 から動的に登録することも可能です
# frozen_string_literal: true
require 'sidekiq-scheduler'
Sidekiq.set_schedule('hello', { 'every' => ['1m'], 'class' => 'HelloWorld' })
これを起動することでスケジューラを登録できます
redis を確認するとスケジューラが登録されています
redis-cli keys '*'
1) "schedules_changed"
2) "schedules"
redis-cli hgetall 'schedules'
1) "hello"
2) "{\"every\":[\"1m\"],\"class\":\"HelloWorker\"}"
また設定ファイルでは dynamic というオプションを true にします
:scheduler:
:dynamic: true
あとはワーカーを起動して毎分実行されることを確認しましょう
ワーカーがスケジューラを認識しない場合はサイド job_create.rb を実行してみてください
同一スケジューラ名の場合は新規で追加はせずスケジューラが更新されるだけになります
またスケジューラの変更はワーカーは動的に行ってくれます
とりあえず動かすところまでやってみました
スケジュールの設定が yaml ファイルなのが特徴的なところかなと思います
また秒単位まで指定できるので細かい時間指定も可能かなと思います