概要(成果物)
今回はPythonを使って反義語(反対の意味を持つ言葉)を返すAPIを構築しました。
その流れを書いていきたいと思います。
主な流れ
まずは、API構築に必要なライブラリをインストールする。
DBはPythonの組み込みで使えるsqlliteを使用したので、インストールの必要はありません。
fastapi==0.89.1
uvicorn==0.20.0
次にDBからデータを取ってくる処理を書く。
(DBの構築についてはおまけに記載)
今回はdetaというFastApiのスポンサーになっているデプロイメントサービスを利用しました。
detaのDocsによるとsqlliteの相性が良いと書いてありましたのでsqlliteで構築をおこないました。
わかりやすいように関数にしてファイルを分けました
import sqlite3
def serarchAnt(searchWord):
# 反義語が見つからない時はnoneを返す(何も返さないとエラーになるため)
antWord = "none"
# DBに接続する
dbname = 'words.db'
conn = sqlite3.connect(dbname)
# SQLiteを操作するためのカーソルを作成
cur = conn.cursor()
# プレスホルダーを使ってsql文を実行する
print(searchWord)
searchMethod = f'SELECT * FROM words WHERE word = "{searchWord}"'
cur.execute(
searchMethod
)
for row in cur:
antWord = row[2]
# DBとの接続を閉じる(必須)
conn.close()
# 反義語を返す
return antWord
# テスト用(デプロイ時には不要)
# if __name__ == '__main__':
# serarchAnt("good")
今回のメイン処理であるFastapiを使ったapiの処理を書いていきます。
参考のQiita記事を参考に最小限の処理を書きました。
from fastapi import FastAPI
from pydantic import BaseModel
from search_antword import serarchAnt
app = FastAPI()
class Item(BaseModel):
name: str
price: float
@app.get("/")
async def root():
return {"message":"This is antonym dictionary"}
@app.get("/searchant/")
def read_item(word:str):
ant = serarchAnt(word)
return {"antword":ant}
ローカルでのテスト
ローカルでテストします。
ターミナルで以下を実行
(--reloadオプションを指定するとapiが起動中もプログラムの変更があった際は自動で反映してくれます)
uvicorn main:app --reload
もう一つターミナルを開いてAPIにアクセスしてみます。
curl "localhost:8000/searchant/?word=good"
# データが返ってくる()
{"antword":"bad"}%
detaにデプロイ
FastApiの公式マニュアルに沿ってdetaにデプロイする
事前にdetaの登録は済ませておいてください
# detaをインストール
$curl -fsSL https://get.deta.dev/cli.sh | sh
# detaがきちんとインストールされたか確認
$deta --help
# ログイン
$deta login
# デプロイ
$deta new
# このようなjsonが返ってくる
# endpointが今回のapiのURLになる
{
"name": "fastapideta",
"runtime": "python3.7",
"endpoint": "https://qltnci.deta.dev",
"visor": "enabled",
"http_auth": "enabled"
}
# そのままだと自分のブラウザ上(キャッシュ利用)でしか動かないので
# インターネットに公開する場合は以下を実行
$deta auth disable
# 2回目以降のデプロイは以下のコマンド
$deta deploy
APIのテスト
APIからデータを持って来れるかテストする
※endpointは実際のものに変えてください
import requests
import sys
word = sys.argv[1]
url = f'https://2obtfx.deta.dev/searchant/?word={word}'
response = requests.get(url)
if response.status_code == 200:
data = response.json()
print(data)
else:
print("failed")
テストしてみると以下のように結果が返ってきました。
$python api_test.py good
{"antword":"bad"}%
おまけ(反義語の辞書を作る)
元データは以下のようなエクセルになります。
これを今回はpandasのフレームワークに起こして、リスト化後DBに保存しました。
なぜそんな面倒なことをしたのかというと、何となくです(好奇心がまさった)
・元データ
・エクセルからpandasのデータにするプログラム
(DBのファイルから読み込むために関数にしています)
import pandas as pd
def makeList():
# エクセルファイルを指定
input_file_name = 'antonyum_dic.xlsx'
# エクセルファイルを開く
input_book = pd.ExcelFile(input_file_name)
# ブック内のシート名を取得
input_sheet_name = input_book.sheet_names
# 一つ目のシートのデータをpandasフレームワークに格納
input_sheet_df = input_book.parse(input_sheet_name[0])
# データフレームからリストを作成する
list_data = input_sheet_df.to_numpy().tolist()
return list_data
import sqlite3
from make_excel_list import makeList
dbname = 'words.db'
# DBを作成する(既に作成されていたらこのDBに接続する)
conn = sqlite3.connect(dbname)
# SQLiteを操作するためのカーソルを作成
cur = conn.cursor()
# テーブルの作成(作成済みの場合は何もしない)
try:
cur.execute(
'CREATE TABLE words(id INTEGER PRIMARY KEY AUTOINCREMENT,word STRING,ant_word STRING)'
)
except:
pass
# テーブル削除
# cur.execute(
# 'DROP TABLE items;'
# )
t = cur.execute(
'SELECT NAME FROM sqlite_master WHERE TYPE="table";'
)
for x in t.fetchall():
print(x)
list = makeList()
cur.executemany('INSERT INTO words values(?,?,?);',list)
# cur.execute(
# 'SELECT * FROM words'
# )
# for row in cur:
# print(row)
# データベースへ
conn.commit()
# DBとの接続を閉じる(必須)
conn.close()
参考
今回のAPIを作り当たり参考にした記事を載せておきます。
(ありがとうございました)
FastApiについて
detaでFastApiをデプロイする(マニュアル)
DBとしてsqliteを使う
sqliteのコマンド