はじめに
最近はあまり触れていないが、PythonでDiscordのBOTを作成しようとすると、discord.pyやPycordのようなライブラリを使うのが一般的と思う。
その中でよく使われる構文の中に@bot.commandのような構文を目にすることだろう。
これはデコレータと呼ばれる構文なのだが、私は中身の構造は関数型引数を実装した関数のような感じなんだろうなと予想はしながらも、なんとなしで利用していた。
そんなわけで、調べていきながら便利な使い方なんかを探してみる。
下記の方法だとPylanceによる静的型解析の際に指摘される。
解決方法について知りたい方は以下の記事を参考にしてほしい。
そもそもどういう構文？
Pycordのサンプルコードを見てみよう。
一番シンプルなBOTはこれだけで実装可能だ。
import discord
from discord.ext import commands
intents = discord.Intents.default()
intents.members = True
intents.message_content = True
bot = commands.Bot(
command_prefix=commands.when_mentioned_or("!"),
intents=intents,
)
# これだけで hello コマンドが実装できる！ すごい！
@bot.command()
async def hello(ctx: commands.Context):
await ctx.send("hello ！")
bot.run("TOKEN")
面倒くさいAPI接続やらソケット通信の確立なんかを全てすっ飛ばして良い感じにコマンドが定義できる。
また、FastAPIやJavaの有名なSpringBootなんかでも同様にいい感じにエンドポイントを作成するのに利用できる。
しかし、これを実装にするにあたってどのような処理が可能かを理解していないため、1つずつ調べていくことにする。
関数型引数との違いを考える
Pythonの話で、他の言語の話を出すのは申し訳ないが、関数型引数を多様するTypeScriptとの違いを見ていきたいと思う。（ちなみにTypeScriptでもデコレータ構文があるので全部TypeScriptだけでよかったのでは？とちょっと思う）
関数型引数を利用した方法(TypeScript)
function hoge(func: Function){
console.log("start hoge")
fuga()
console.log("end hoge")
}
function test(){
hoge(() => {
console.log("Hello")
})
}
test()
デコレータを利用した方法(Python)
from typing import Callable, Any
def hoge(func: Callable[..., None]):
def wrapper(*args, **kwargs):
print("start hoge")
func(*args, **kwargs)
print("end hoge")
return wrapper
@hoge
def test():
print('Hello')
どちらも出力は以下のようになる。
start hoge
Hello
end hoge
デコレータ入門のコードで出てくるコードはこの辺だろう。
やってることは単純で、どちらの方法でも問題ないように思える。
しかし、デコレータは重ねて使うことも可能なので以下のように書くことができる。
from typing import Callable, Any
def hoge(func: Callable):
def wrapper(*args, **kwargs):
print("start hoge")
func(*args, **kwargs)
print("end hoge")
return wrapper
def fuga(func: Callable):
def wrapper(*args, **kwargs):
print("start fuga")
func(*args, **kwargs)
print("end fuga")
return wrapper
@hoge
@fuga
def test():
print('Hello')
test()
これを関数型引数で実装しようとすると以下のようになる。
function hoge(func: Function){
console.log("start hoge")
func()
console.log("end hoge")
}
function fuga(func: Function){
console.log("start fuga")
func()
console.log("end fuga")
}
function test(){
hoge(() => {
fuga(() => {
console.log("Hello")
})
})
}
test()
関数ネストが増えてしまってイマイチで、このまま続けていくならば波動拳コードが生まれてしまいそうだ。
DB接続などのコネクションの確立として利用する
なんとなく可読性のいいコードが作れそうだなーという感覚は得たが、前処理として利用しコールバックを受け取るような処理などで利用したい。
そこでコンソールだけで遊べるSNSのようなものを作る事にする。
テーブル
ユーザー情報とポストを保存するだけの簡単な構造だ。
最近は生成AIのお陰でデータ構造もカラム名も考える必要がなくて、くそざこえんじにあの私の必要性が脅かされているとても楽である。
create_table.sql
-- Usersテーブルの作成
CREATE TABLE Users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
username TEXT NOT NULL UNIQUE,
password TEXT NOT NULL,
created_at DATETIME DEFAULT CURRENT_TIMESTAMP
);
-- Postsテーブルの作成
CREATE TABLE Posts (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER NOT NULL,
content TEXT NOT NULL,
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (user_id) REFERENCES Users (id)
);
これをPythonで使うためにModelを作成しておきます。
model.py
from datetime import datetime
from dataclasses import dataclass
from typing import Optional
@dataclass
class User:
id: Optional[int] = None
username: Optional[str] = None
password: Optional[str] = None
created_at: Optional[datetime] = None
@dataclass
class Post:
id: Optional[int] = None
user_id: Optional[int] = None
content: Optional[str] = None
created_at: Optional[datetime] = None
user: Optional[User] = None
DB接続を行うデコレータ
データベースに接続し、何かしらの操作を行う際に毎回呪文的にコネクションを生成してSQLを実行し、例外があったらロールバックして接続を終了するなど、めんどくさい処理を行う場合がある。
接続や、例外処理は共通の処理なので前後の処理はあらかじめデコレータにすることでかなりスッキリした書き方が可能になる。
db_connection.py
import sqlite3
from typing import Callable, Any
from model import User, Post
DB_PATH = 'test.db'
def db_connect(auto_commit: bool = False) -> Callable:
# デコレータ
def decorator(func: Callable[[sqlite3.Connection, sqlite3.Cursor], Any]) -> Callable:
# ラッパー
def wrapper(*args, **kwargs):
connection = sqlite3.Connection(DB_PATH)
cursor = connection.cursor()
try:
print('open')
result = func(connection, cursor, *args, **kwargs)
except Exception as e:
connection.rollback()
raise e
else:
# デコレータ生成時のauto_commitによってコミットするか決定する
if auto_commit:
connection.commit()
return result
finally:
print('close')
connection.close()
return wrapper
return decorator
これを利用するには以下のようにデコレータを生成する関数を呼び出しラッピングする。
# デコレータを呼び出す関数を実行 -> decorator関数を呼び出す
@db_connect(auto_commit=True)
# 以下の関数はdecorator関数内のwrapper関数でラッピングされる。
def create_table(connection: sqlite3.Connection, cursor: sqlite3.Cursor):
with open('create_table.sql', 'r') as f:
cursor.execute(f.read())
connectionとcursorはラッパー関数から接続した状態で受け渡されるため、それを利用する。
この関数を呼び出す際はwapper関数を呼び出すイメージとなるので追加の引数としてuserを渡した場合は以下のように利用可能。
@db_connect(auto_commit=True)
def create_user(
connection: sqlite3.Connection,
cursor: sqlite3.Cursor,
# このuserが引数として追加されている
user: User
):
cursor.execute(
'INSERT INTO Users (username, password) VALUES (?, ?)',
(user.username, user.password)
)
# 呼び出す際は引数としてユーザーを与えるだけで実行可能
create_user(User(username="nikawamikan", password="P@SSW0RD"))
以下のようにSQLを実行する関数を実装した。
db.py
import sqlite3
from typing import Callable, Any
from model import User, Post
@db_connect(auto_commit=True)
def create_table(connection: sqlite3.Connection, cursor: sqlite3.Cursor):
with open('create_table.sql', 'r') as f:
cursor.executescript(f.read())
@db_connect(auto_commit=True)
def create_user(connection: sqlite3.Connection, cursor: sqlite3.Cursor, user: User):
cursor.execute(
'INSERT INTO Users (username, password) VALUES (?, ?)',
(user.username, user.password)
)
@db_connect()
def login_user(connection: sqlite3.Connection, cursor: sqlite3.Cursor, user: User) -> User:
cursor.execute(
'SELECT id, username, created_at FROM Users WHERE username = ? AND password = ?',
(user.username, user.password)
)
result = cursor.fetchone()
if user is None:
raise ValueError('ユーザーが存在しません')
return User(id=result[0], username=result[1], created_at=result[2])
@db_connect(auto_commit=True)
def post(connection: sqlite3.Connection, cursor: sqlite3.Cursor, user: User, content: str):
cursor.execute(
'INSERT INTO Posts (user_id, content) VALUES (?, ?)',
(user.id, content)
)
@db_connect()
def get_posts(connection: sqlite3.Connection, cursor: sqlite3.Cursor) -> list[Post]:
cursor.execute(
'''SELECT
Posts.id,
Posts.user_id,
Posts.content,
Posts.created_at,
Users.username
FROM
Posts
JOIN
Users ON Posts.user_id = Users.id''',
)
results = cursor.fetchall()
return [
Post(
id=result[0],
user_id=result[1],
content=result[2],
created_at=result[3],
user=User(username=result[4]),
) for result in results
]
@db_connect()
def get_my_posts(
connection: sqlite3.Connection,
cursor: sqlite3.Cursor,
user: User
) -> list[Post]:
cursor.execute(
'SELECT id, content, created_at FROM Posts WHERE user_id = ?',
(user.id,)
)
results = cursor.fetchall()
return [
Post(
id=result[0],
user_id=user.id,
content=result[1],
created_at=result[2],
) for result in results
]
@db_connect(auto_commit=True)
def delete_post(
connection: sqlite3.Connection,
cursor: sqlite3.Cursor,
posts: tuple[int], user: User
):
sql = f'DELETE FROM Posts WHERE id IN ({",".join(
["?"] * len(posts))}) AND user_id = ?'
posts = posts + (user.id,)
cursor.execute(sql, posts)
コマンド登録を行うデコレータ
コンソールアプリとするためにコマンド登録を行うデコレータを作成していく。
少しややこしいが以下のようなクラスを作成してコマンドを登録することのできるクラスを作成する。
import sqlite3
from typing import Callable, Optional
import inspect
import db
from model import User, Post
class Commands:
class Command:
# Commandクラスのコンストラクタ。関数、引数、説明、複数引数対応フラグを受け取る
def __init__(
self,
func: Callable[..., None],
params: dict[str, inspect.Parameter],
description: Optional[str],
multiple: bool = False
):
self.func = func
self.params = params
self.description = description
self.multiple = multiple
# Commandsクラスのコンストラクタ。コマンドの辞書と名前を初期化
def __init__(self, name: str):
self.commands: dict[str, Commands.Command] = {}
self.name = name
# コマンドを追加するためのデコレータ関数
def add_command(
self,
name: Optional[str] = None,
description: Optional[str] = None,
multiple: bool = False
):
def decorator(func: Callable[..., None]):
# 実行されるコマンド関数を定義し、コマンド辞書に追加
def command(*args, **kwargs):
return func(*args, **kwargs)
self.commands[name or func.__name__] = Commands.Command(
command,
inspect.signature(func).parameters,
description or func.__doc__,
multiple
)
return decorator
# 登録されたコマンドのヘルプを表示する関数
def help(self):
print(
"\n".join([
f"{k}: {v.description}"
for k, v in self.commands.items()
])
)
# コマンドの実行ループ
def run(self):
while True:
try:
# コマンドの入力を受け取る
command_str = input(f'{self.name}> ')
except KeyboardInterrupt:
# キーボード割り込みが発生した場合、プログラムを終了
print('exit')
break
if (command_str == ''):
continue
# 入力をスペースで分割してリスト化
command_list = command_str.split()
command_str = command_list[0] # 最初の要素をコマンドとして取得
if command_str == 'exit':
break # 'exit'コマンドが入力された場合、ループを抜ける
if command_str == 'help':
self.help()
continue # 'help'コマンドが入力された場合、ヘルプを表示して次のループへ
if command_str in self.commands:
command = self.commands[command_str]
# 引数とコマンドの数が一致しているか確認
if not command.multiple and len(command.params) != len(command_list[1:]):
print(command.description or '引数の数が一致しません')
continue
# コマンドを実行
command.func(*command_list[1:])
else:
# コマンドが存在しない場合のエラーメッセージ
print('コマンドが存在しません')
これを利用するにはインスタンス作成を行い、インスタンスの@add_command()デコレータを呼び出して登録すればOK。
# Commandsクラスのインスタンスを作成
test = Commands('test')
# 'echo'コマンドを追加。デコレータを使用して関数を登録
@test.add_command()
def echo(name: str):
"""名前を表示します
name: 名前
"""
print(name)
# コマンドの実行ループを開始
test.run()
実際に実行すると以下のようにechoコマンドが利用できる。
test> test
コマンドが存在しません
test> echo
名前を表示します
name: 名前
test> echo nikawamikan
nikawamikan
あとはコレをいい感じに組み合わせるだけで作成可能だ。
各コマンドの分類ごとにCommandsクラスを生成する。
削除コマンド
ここでは削除するためのコマンドを定義する。
先ほど作成した関数を利用してデータの取得や、削除などを行う。
delete_cmd = Commands('delete')
@delete_cmd.add_command(
name="show",
description="削除対象を選択します。"
)
def show_my_post():
global user
posts = db.get_my_posts(user)
print('自分の投稿一覧')
for post in posts:
print(f'{post.id}: \n content: {post.content}' +
f'\n created_at: {post.created_at}')
@delete_cmd.add_command(
name="delete",
description="削除します。\n 引数にidを指定してください。(複数可)",
multiple=True
)
def delete_post(*ids: str):
if len(ids) == 0:
print('削除する投稿のidを指定してください。')
return
global user
db.delete_post(tuple(map(int, ids)), user)
メニューコマンド
ここではログイン後のメニューを実装した。
先ほどのdelete_cmdインスタンスをdeleteコマンドから呼び出すようにすることで、コマンドを階層化するようにしている。
menu_cmd = Commands('menu')
@menu_cmd.add_command(
name="show",
description="投稿を表示します。"
)
def show_post():
posts = db.get_posts()
for post in posts:
print(f'{post.user.username}: {post.content} ({post.created_at})')
print('')
@menu_cmd.add_command(
name="delete",
description="投稿を削除します。"
)
def delete():
print('削除モードに入ります。')
delete_cmd.run()
@menu_cmd.add_command(
name="post",
description="投稿します。\n 引数に投稿内容を指定してください。"
)
def post(content: str):
global user
db.post(user, content)
メインコマンド
ここではログインとアカウント作成の機能を定義し、ログインを実行することでメニューを呼び出している。
main_cmd = Commands('main')
@main_cmd.add_command(
name="create",
description="ユーザーを作成します。"
)
def create_user():
username = input('ユーザー名: ')
password = input('パスワード: ')
try:
db.create_user(User(username=username, password=password))
except sqlite3.IntegrityError:
print('ユーザーが既に存在します')
@main_cmd.add_command(
name="login",
description="ログインします。"
)
def login():
global user
username = input('ユーザー名: ')
password = input('パスワード: ')
try:
user = db.login_user(User(username=username, password=password))
print(f'{user.username}さん、こんにちは')
menu_cmd.run()
except ValueError:
print("ログインに失敗しました。")
あとはmain_cmd.run()で実行が可能だ。
このようにすることでコマンドの実装が比較的カンタンに行える事がわかった。
まとめ
所感としては以下の様な感じ。
- 関数型引数を普段から利用するのであればデコレータにリファクタリングすることでネストが減ったり、意味もわかりやすくなりやすい
- 何かしらのおまじない的な文法があるコードであればコードの共通化がしやすい
- コールバック的な処理を実装するの際、その関数自体がどのイベントと紐づいてるかわかりやすいし実装も容易
- ただしコールバック的に利用する場合、引数がどんな意味を持つのかわかりにくくなるため、アノテーションをつけたほうが良い
nikawamikan = mikan('nikawa')
@nikawamikan.小学生並みの感想
def say():
return "個人的にはフレームワーク向きでいいなーという感じでした"