PostgreSQLとFastAPIとStreamlitを使ってTodoリストを作ろう（前編）
本記事の目的
最近、Todoリストの作り方を学んだので、知識を深化させるために言語化する。プログラミング初心者の筆者でも作れたので、プログラミング初心者のモチベーションを上げるためにも、できるだけ分かりやすく書いていく。
PostgreSQLとは
リレーショナルデータベース管理システムの一つ。正直、MySQLやOracle Databaseでも良いが、初心者の自分にとっては、PostgreSQL用のGUI管理ツールのpgAdmin4が使いやすいので、こちらを採用。インストール方法や操作方法については下記の記事を参考にしてほしい。
FastAPIとは
Web APIのフレームワークの一つ。軽量かつ高速が特徴。公式も必要な部分は日本語化されており、困ったら参考にしてもらいたい。 また、こちらのFastAPI入門書も読みやすいのでオススメ。
手順0：環境・ディレクトリ構成
初めに環境とディレクトリ構成だけ示しておく。今回はバックエンドだけの記事だが、後編ではフロントエンドについて書いていくので、フロントエンドの構成図も記入しておく。
Python：3.9.13
pgAdmin4：3.6.19
VSCode：1.76.2
TODOLIST
├ BACKEND/
│ ├ CRUD/
│ │ ├ delete_todo.py
│ │ ├ get_todo.py
│ │ ├ insert_todo.py
│ │ └ update_todo.py
│ ├ DB/
│ │ └ db.py
│ ├ ROUTER/
│ │ ├ delete_todo.py
│ │ ├ get_todo.py
│ │ ├ insert_todo.py
│ │ └ update_todo.py
│ ├ SCHEMA/
│ │ └ schema.py
│ ├ main.py
│ ├ .env
│ └ .gitignore
└ FRONTEND/
└ front.py
手順1：データベースの作成
まずは、Todoリストを保管するためのデータベースを作る。PostgreSQLのインストール方法については上記の記事を参考にしていただきたい。まず、PostgreSQL内のデータベースを右クリックする。
次に作成→データベースをクリックし、データベース名を登録する。（今回はtestdb）
すると、データベース直下にtestdbというデータベースができるので、これを右クリック→クエリツールをクリック。
ここでお気づきの方はいると思うが、今回はUIベースではなく、コードベースでテーブルを作成する。pgAdmin4はUIベースで直感的にテーブルを作成できることが良いところだが、SQLの知識を蓄えたいので、あえてコードベースで作成する。クエリ上には下記のコードを入力。
CREATE TABLE todolist
(
deadline date,
todo character varying,
priority integer,
genre character varying,
completed_flg character varying,
nowdate date
);
すると、テーブル直下にtestdbというテーブルが生成される。
実際に下記のコードを入力して、テーブルを確認してみよう。
select * from todolist
これで、Todoリストを保存するテーブルは準備できた。次に、このデータベースにアクセスするコードをVSCode上に書いていく。
手順2-1：事前準備
ここからは必要なライブラリがインストールしてある前提で話す。ライブラリについては「○○ インストール python」と検索すれば大体出てくるので、そちらを参考にしてほしい。
まずは、PostgreSQLのデータベースに接続する。接続するために必要な情報は、データベースがあるサーバー（ホスト）名、データベースにアクセスする際のユーザー名とパスワード、そして使いたいデータが入っているデータベース名である。これらに対応する変数をそれぞれ、DB_HOST、DB_USER、DB_PASS、DB_NAMEとする。これらの情報がどこに記載されているかについて次に説明する。まずは、pgAdmin4を開く。ここでパスワードが求められると思う。このパスワードがDB_PASSである。次にServers直下のPostgreSQLを右クリックする。
そして、プロパティ→接続の順にクリックする。ここで表示されるホスト名/アドレスがDB_HOST、ユーザー名がDB_USERである。最後にデータベース名だが、こちらは手順1で設定したデータベース名（testdb）がDB_NAMEに該当する。
安全性を高めるため、外に漏れるとまずいデータは.envファイルに保管する。下記のコードは一部マスキングしてあるので、各自のデータを入力してほしい。
DB_USER="****"
DB_PASS="****"
DB_HOST="****"
DB_NAME="testdb"
筆者はgitlab上で開発を行うので、.gitignoreファイルに.envを書き、gitlab上に表示されないようにする。
.env
.gitignoreに入れられたファイルはVSCode上のファイル名が薄くなっているので、これが確認出来たら隠しファイルになっている。これで事前準備が終わった。
手順2-2：データベースとのリンク
まずコードを記す。その後のコードの説明をする。
import os
from posixpath import dirname
from dotenv import load_dotenv
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from os.path import join,dirname
dotenv_path=join(dirname(__file__),'../.env')
load_dotenv(dotenv_path)
DB_USER=os.environ.get("DB_USER")
DB_PASS=os.environ.get("DB_PASS")
DB_HOST=os.environ.get("DB_HOST")
DB_NAME=os.environ.get("DB_NAME")
SQLALCHEMY_DATABASE_URL='postgresql+psycopg2://{}:{}@{}/{}'.format(
DB_USER,
DB_PASS,
DB_HOST,
DB_NAME
)
engine=create_engine(SQLALCHEMY_DATABASE_URL)
Sessionlocal = sessionmaker(autocommit=False,autoflush=True,bind=engine)
Base=declarative_base()
def get_db():
db=Sessionlocal()
try:
yield db
finally:
db.close()
-
join(dirname(file),'../.env')で何をやっているか
dirname(__file__)は現在のディレクトリ名を返す。（今回だと~TODOLIST/BACKEND/DB）
ということでjoin(dirname(__file__),'../.env')は~TODOLIST/BACKEND/DB/../.envを表す。..はひとつ前のフォルダに移動なので~TODOLIST/BACKEND/.envといったパスをdotenv_pathに代入している。
-
load_dotenv(dotenv_path)で何をやっているか
load_dotenv()は.envファイルが見つかるまで、親ディレクトリを遡り、.envファイルを実行する。なので今回の様に.envファイルの正確なパスを指定しなくてもよい。正直、load_dotenv()でもルートディレクトリ（起点となるディレクトリ。今回の場合はTODOLISTファルダ。）内に.envファイルがあれば読み取ってくれるが、.envのパスを理解しておくため、ガチガチに記入した。また、個人的にはload_dotenv(dotenv_path)を記入する場所も大事に感じた。次に書くos.environ.get()で環境変数にアクセスするが、別ファイルの同じ変数名のものが存在するとき、最後に読み取られた変数のものを参照する。なので、os.environ.get()でアクセスする環境変数は、可能ならばos.environ.get()の直前に定義しておくことで、間違いなくアクセスできる。
-
os.environ.get("DB_????")で何をやっているか
os.environ.get()は()内のキー（環境変数）をルートディレクトリ上で調べ、値を文字列で返す。例えばos.environ.get("DB_NAME")は.env内のDB_NAMEを参照し、代入されている"testdb"を返す。
-
'postgresql+psycopg2://{}:{}@{}/{}'で何をやっているか
PostgreSQLに接続するためのおまじないだと思ってほしい。{}内には下記の順で情報を入れる。今回はformat関数を使って、後ろに書いている。
postgresql+psycopg2://{ユーザー名}:{パスワード}@{ホスト名}:{ポート番号}/{データベース名}
-
sessionmaker(autocommit=False,autoflush=TRUE,bind=engine)で何をやっているか
autoflushをTRUEにすることで、データベースへの変更処理が次回データベースを呼んだタイミングで変更内容が反映される。変更処理後にdb.commit()などで呼び出さないと、Swagger UI上では変更が確認されるのに、大元のdb上では変わっていないということになってしまう。
-
def get_db()で何をやっているか
Sessionlocal()で前の文で定義したデータべース情報をdbに代入。try+yield/finally文で書くことで、データベースが開く・開かない関係なく、最後はデータベースをクローズしてくれる。これは後々使うので関数化しておく。
手順3：データの呼び起こし
手順2でデータベースに接続することができたので、そこに保存されているテーブル（データ）を呼んでみましょう。まずは、コード。
from sqlalchemy.orm import Session
from sqlalchemy.sql import text
def get_todolist(db:Session):
sql=text('select * from todolist order by "deadline" asc;')
result=db.execute(sql).all()
return result
-
db.execute(sql).all()で何をやっているか
手順2で指定したデータベース(db)にSQL文を実行(execute(sql))し、全データを返す(all())ことを行う。SQL文はtext文の（）内を指す。
手順4：出力データの型定義
Pythonの良いところでもあり、悪いところでもあるのが、「動的に変数の型を変える」ことです。VBAを使ったことがある方は経験していると思うが、通常、変数を用いる際は先にその変数の型を定義する必要がある。
Dim i as integer
i=2
Pythonだと型を定義せず使える。しかも、次に違う型のデータが入ったとしても、その型に動的に変わる。
i=2
print(type(i))
#<class 'int'>
i="test"
print(type(i))
#<class 'str'>
しかし、これだと気づかないうちに変数の型が変わってしまい、後のコードで引っかかる＋どこから間違えているか分からなくなる。なので、何のデータであれ、型定義は大事。ということで、定義していく。
from pydantic import BaseModel
import datetime
class class_schema(BaseModel):
#期日
deadline:datetime.date
#やること
todo:str
#優先度
priority:int
#カテゴリ
genre:str
#完了フラグ
completed_flag:bool
#作成日
nowdate:datetime.date
class Config:
orm_mode = True
-
class Config~で何をやっているか
ここで10時間は取られた。別パソコンで実装しているときは、この文が無くても動いたので油断していた。FastAPIに渡すデータは本来、dict形式でないといけないが、SQLAlchemyで呼ぶデータはORM形式（データベースとオブジェクト指向プログラミング言語を変換する型）なので、orm_modeをTrueにしないと下記の様なエラーが出て、止まる。逆に、何故この文が無くても、別のパソコンでは動いたのか。ここはもう少し検証する。
value is not a valid dict (type=type_error.dict)
手順5：パスオペレーションの定義
FastAPIではREST API形式でデータのやり取りができる。REST API形式とは、ざっくり言えば、「パス（/以降の文）でデータをやり取りする方法」です。HTTP通信みたいにデータをやり取りしましょうねということ。手順3ではデータのやり取り部分を作成したので、これをパスでやり取りできるようにする。まずはコード。
from fastapi import APIRouter,Depends
from sqlalchemy.orm import Session
from DB import db
from CRUD import get_todo
from SCHEMA.schema import class_schema
router=APIRouter()
@router.get("/router",response_model=list[class_schema],tags=['todo'],description='get_todo')
def get(db:Session=Depends(db.get_db)):
result=get_todo.get_todolist(db=db)
return result
-
@router.get~で何をやっているか
@~はデコレータ文といい、第一引数にパスを指定することで、そのパスが呼ばれたら、def以降の関数を行うようになる。以上が、パスパラメータ関数のプログラミングの一通りの流れである。ほとんどコピペで済むので説明は省くが、残りのCRUD処理のコードだけ記す。
from sqlalchemy.orm import Session
from sqlalchemy.sql import text
import datetime
def delete_todolist(deadline:datetime.date,todo:str,priority:int,genre:str,completed_flag:bool,nowdate:datetime.date,db:Session):
sql=text('delete from todolist where "deadline"=%s and "todo"=%s and \n\
"priority"=%s and "genre"=%s and "completed_flag"=%s and "nowdate"=%s;'\
%(repr(deadline),repr(todo),repr(priority),repr(genre),repr(completed_flag),repr(nowdate)))
db.execute(sql)
db.commit()
sql2=text('select * from todolist order by "deadline" asc')
result=db.execute(sql2).all()
return result
from sqlalchemy.orm import Session
from sqlalchemy.sql import text
import datetime
def insert_todolist(deadline:datetime.date,todo:str,priority:str,genre:str,db:Session):
completed_flag=False
nowdate=datetime.datetime.today().strftime("%Y-%m-%d")
sql=text('insert into todolist values (%s,%s,%s,%s,%s,%s);'\
%(repr(deadline),repr(todo),repr(priority),repr(genre),repr(completed_flag),repr(nowdate)))
db.execute(sql)
db.commit()
sql2=text('select * from todolist order by "deadline" asc;')
result=db.execute(sql2).all()
return result
from sqlalchemy.orm import Session
from sqlalchemy.sql import text
import datetime
def update_todolist(deadline:datetime.date,todo:str,priority:int,genre:str,nowdate:datetime.date,db:Session):
sql=text('update todolist set "completed_flag"=True where "deadline"=%s and "todo"=%s and \n\
"priority"=%s and "genre"=%s and "nowdate"=%s;'\
%(repr(deadline),repr(todo),repr(priority),repr(genre),repr(nowdate)))
db.execute(sql)
db.commit()
sql2=text('select * from todolist order by "deadline" asc')
result=db.execute(sql2).all()
return result
from fastapi import APIRouter,Depends
from sqlalchemy.orm import Session
from DB import db
from CRUD import delete_todo
from SCHEMA.schema import class_schema
router=APIRouter()
@router.delete("/router",response_model=list[class_schema],tags=['todo'],description='delete_todo')
def delete(deadline:str,todo:str,priority:int,genre:str,completed_flag:bool,nowdate:str,db:Session=Depends(db.get_db)):
result=delete_todo.delete_todolist(deadline=deadline,todo=todo,priority=priority,genre=genre,completed_flag=completed_flag,nowdate=nowdate,db=db)
return result
from fastapi import APIRouter,Depends
from sqlalchemy.orm import Session
from DB import db
from CRUD import insert_todo
from SCHEMA.schema import class_schema
router=APIRouter()
@router.post("/router",response_model=list[class_schema],tags=['todo'],description='insert_todo')
def insert(deadline:str,todo:str,priority:str,genre:str,db:Session=Depends(db.get_db)):
result=insert_todo.insert_todolist(deadline=deadline,todo=todo,priority=priority,genre=genre,db=db)
return result
from fastapi import APIRouter,Depends
from sqlalchemy.orm import Session
from DB import db
from CRUD import update_todo
from SCHEMA.schema import class_schema
router=APIRouter()
@router.patch("/router",response_model=list[class_schema],tags=['todo'],description='update_todo')
def update(deadline:str,todo:str,priority:int,genre:str,nowdate:str,db:Session=Depends(db.get_db)):
result=update_todo.update_todolist(deadline=deadline,todo=todo,priority=priority,genre=genre,nowdate=nowdate,db=db)
return result
最後にFastAPIのUIであるSwagger UIにこれらの機能を実装する。
手順6：Swagger UIへの実装
DB→CRUD→SCHEMA→ROUTERと9割は完了しており、mainには実装コードだけを入力し、最後にコマンドでFastAPIを起動する。
from fastapi import FastAPI
from ROUTER import get_todo,insert_todo,delete_todo,update_todo
app=FastAPI()
app.include_router(router=get_todo.router)
app.include_router(router=insert_todo.router)
app.include_router(router=delete_todo.router)
app.include_router(router=update_todo.router)
uvicorn main:app --reload
-
app.include_router~で何をやっているか
手順5で作成したrouterをSwagger UI上に表示させるために、FastAPIオブジェクトに追記している。
-
uvicorn main:app --reloadで何をやっているか
uvicorn（ユーブイアイコーン）はWebサーバとWebアプリケーションが通信するためのインターフェースである。このコードでは、main.py内のappを呼んでいることを示す。--reloadを最後に付けることで、main.pyがあるディレクトリで変更が生じた際に、再度appを呼んでくれる。このコードを実行することで、http://127.0.0.1:8000 が生成されると思う。これに/docsを付けて（http://127.0.0.1:8000/docs） にアクセスすると、Swagger UIにアクセスできる。
これでバックエンドの作成は終了。Swagger UIを用いれば、フロントエンドを作らなくても、最低限のTodoリスト操作は可能だと思う。余力がある方は、後編のstreamlitを用いたフロントエンド編もお付き合いください。
Discussion