PythonとBlackSheepで爆速Webサービスを作る
はじめに
PythonでWebアプリケーションを作る場合のフレームワークとしては、DjangoやFlaskが候補にあがるかと思います。とはいえこれらのフレームワークには大きな問題があります。それは、信じられないくらい遅いという問題です。
ベンチマークを見てみると、DjangoやFlaskは遅すぎてもはや評価対象にすらならないといったところです。
そこで今回は、ベンチマーク上では最速であるBlackSheepを利用して爆速Webアプリケーションを作ってみることにしました。
BlackSheepの特徴
非同期処理を利用したWebアプリケーションフレームワークであるため、非常に高速なレスポンスが期待できます。
また、非常に軽量なフレームワークであり、起動時間も比較的早いといった特徴を持ちます。
とてもざっくり言うなれば、とても速いフレームワークといったところです。
また、その他の大きな特徴として、Automatic binding と Dependency Injection があげられます。
Automatic binding
Webアプリケーションフレームワークとしてはよくある機能ですが、HTTPリクエスト時に付与されるクエリパラメーターなどを自動で設定してくれるというものです。
@get("/foo")
async def example(
page: int = 1,
search: str = "",
):
# page はクエリパラメーターから自動で取得され、設定されていない場合の初期値は1になります
# search も同様に自動で取得され、初期値は "" となります
...
このような形で自動で設定してくれるため、処理を書くのが簡単になりますよね、といったところです。
Dependency Injection（DI）
こちらもよくある機能ですが、使い勝手が良い形で実装されているのは少し珍しいなと思いました。
from blacksheep import Application
from domain.foo import Foo
app = Application()
app.services.add_exact_scoped(Foo) # Foo をDIするように設定
@app.route("/")
def home(foo: Foo): # <-- foo が常に設定されるようになる
return f"Hello, {foo.foo}!"
このような形で常に使うような処理をDIさせることで、コードを簡易化することができます。
BlackSheepを使ってみた感想
DjangoやFlaskを触ったことがある身としては、これといって迷うことなく使うことができ、学習コストは比較的低いフレームワークであるように感じました。
他方で、フレームワークのデフォルトでは提供されていない機能が多く、少し凝ったことをしようとした場合は他のライブラリを探すか自力で実装するしかなく、少し手間だなと感じました。
例えば、ORマッパーはフレームワークの標準機能に含まれておらず、自分で選択する必要があります。この辺りはDjangoやRuby on Rails などに慣れている人にとっては、ひとつのハードルになるかもしれません。
なお、今回は非同期実行を特徴にもつ Tortoie ORM を利用したのですが、こちはこちらで少しクセのあるORマッパーでした。（機会があれば Tortoise ORM についても記事が書けたらと思います。）
とはいえ全体的には非常に良い印象を持っています。ドキュメントもしっかり揃っていますし、何よりも速いです。速さに勝るものはありません。個人的には、今後Pythonで開発するのであれば積極的に使っていこうと思っています。
ハマったところ
ここからは、実際に実装していく中でハマった点を書いていきます。
認証＆認可周り
認証＆認可周りについては最低限の機能しか提供されておらず、ほぼ自力で書く必要があります。
フレームワークで提供されている機能をざっくりといえば、リクエストごとに request.identity が自動で設定される（上述のDIですね）といったコアな機能のみとなっています。
そのため、どのような値を request.identity に設定するかなどの部分を自力で書く必要があります。
ログイン状態によってアクセス可否を判断するためには、大きく下記の3つを書く必要があります。
-
request.identityに情報を設定するためのhandler - アクセス可否を判断するための
Policy - コントローラーに対して
authデコレーターを付与
handler の作成
handler を作成してアプリケーションに設定することで、 request.identity に値を設定することが出来るようになります。
下記はセッションに設定された user_id を request.identity に自動で設定することでアクセス可能か判断する例です。（アクセス可否の判断は後述の Policy の中で行います）
# handlerの例
from guardpost import Identity, User
from guardpost.asynchronous.authentication import AuthenticationHandler
class ExampleAuthHandler(AuthenticationHandler):
def __init__(self):
pass
async def authenticate(self, context: Request) -> Optional[Identity]:
# セッションに user_id が設定されている場合は認証済みとするため、
# identity に user を設定している
session_user_id = context.session.get("user_id", "")
if session_user_id:
user = User({"user_id": session_user_id}, "AUTH")
context.identity = user
return user
return None
# handelre は下記の形でアプリケーションに登録する
app.use_authentication().add(ExampleAuthHandler())
毎回セッションを読みに行けば同じようなことは実現できますが、こちらの方が比較的楽にコードを書けるかなとは思います。他方で、DjangoやRailsであれば標準で提供されている機能であるため、自分で書くのは手間に感じました。
Policy の例
handler によって設定された内容をチェックし、当該処理にアクセス出来るかどうかを判断する処理を書きます。
さらに Requirement を継承したクラスを作成し、条件を満たす場合に context.succeed を呼び出す handle メソッドを作ることで、処理にアクセス可能かどうかを判断する処理を実現します。
その上で、上述のクラスを Policy を継承したクラスに読み込ませ、app.use_authorization().add() に追加することで設定内容をどこからでも呼び出せるようにします。
from blacksheep.server.authorization import Policy
from guardpost.authorization import AuthorizationContext
from guardpost.synchronous.authorization import Requirement
class LoginRequirement(Requirement):
def handle(self, context: AuthorizationContext) -> None:
identity = context.identity
if identity is not None:
context.succeed(self)
class LoginRequiredPolicy(Policy):
def __init__(self):
# ここでは "default" という名前を Policy に設定している
super().__init__("default", LoginRequirement())
# ポリシーをアプリケーションに設定する
app.use_authorization().add(LoginRequiredPolicy())
この辺りは少し面倒ですね。
他方で、特定の何かを持つ場合はこの操作を許可、といった細かい制御をしたい場合などは便利に使うことができそうです。
なお、 Policy には名前を設定することができ（上述の例では "default"）、これを後述の auth の引数に付与することで、利用する Policy をコントローラー側から指定できるようになります。（例： auth("default")）
コントローラーの例
コントローラーは比較的シンプルです。
auth デコレーターをつけるだけで実現できます。
from blacksheep.server.authorization import auth
@auth
@app.route("/")
async def only_auth_user(user: Identity):
if user.is_authenticated():
response = pretty_json(user.claims)
return response
このあたりは、DjangoやFlaskと同じように実現できますね。
ログイン状況をテンプレート側で判定したいときの処理
MPAでアプリケーションを作る場合のあるあるですが、テンプレート側でログイン状態による分岐を実現したいことがあるかと思います。その場合の処理がなかなか面倒でした。
実現したい処理は下記のようなものです。
なお、BlackSheepのデフォルトでは Jinja2 を拡張したものがテンプレートエンジンとして使われています。
{% if login_user %}
<p>ログインしています</p>
{% else %}
<p>未ログイン状態です</p>
{% endif %}
もちろん、すべてのリクエストの戻り値に login_user を含めれば良いのですが、流石に毎回それを書くのはナンセンスです。
ですので、今回は標準の Controller クラスを拡張した BaseController クラスを作成することで対応しました。
from blacksheep.server.controllers import Controller
class BaseController(Controller):
def __init__(self):
# リクエスト情報を退避するインスタンス変数
self._request = None
async def on_request(self, request: Request) -> None:
# on_request はリクエストの処理の開始時に常に呼ばれる
# このタイミングで request を退避
self._request = request
# view はテンプレートをレンダリングする処理
# 親クラスのメソッドをオーバーライド
def view(
self, name: Optional[str] = None, model: Optional[Any] = None, **kwargs
) -> Response:
# ログイン状況を request.identity から取得
login_user = self._request.identity if self._request else None
return super().view(
name,
model,
request=self._request,
flash_message=self._flash_message,
login_user=login_user, # 取得したログイン状況を常にテンプレートに追加
**kwargs,
)
今回はこのような形で実現しましたが、他に良い例があれば教えていただけると嬉しいです。
作ったサービス
今回はBlackSheepを利用して本のレビューサイトを作成しました。
超爆速かといわれるとデータベースのioネックで少しもっさりする部分もあるのですが、データベース読み込みのないページはなかなかに早く仕上がっていると思います。
こちらのサービスもなかなか面白い感じになっているので、ぜひ見てみてください！
Discussion