ハンズオンで学ぶ、初心者向けのFastAPI~その2~
はじめに
前回に続き第二回目のfastapi
に関する記事です。前回の記事を未読の方は、前回の記事を読んで頂ければと思います。
今回は、DB との接続を行いデータの処理をやってみようと思います。まず今回、新しく使用するフレームワークを紹介しようと思います。
今回新しく使用するフレームワーク
今回、新しく使用するフレームワークは前回の記事でもチラッと出しましたがsqlalchemy
というフレームワークを使用します。(下記は、sqlalchemy
の公式ドキュメントです。)
こちらのフレームワークは、ORM(Object Relational Mapping)と言われるものの一つです。これのメリットは、sql
を書くことなくpython
コードを書くだけでsql
が書けちゃうことです。python
でsql
を書くだけなら普通のsql
書くのと変わらないのではと思った方、python
コードで書く普通のsql
を書くよりメリットがちゃんとあります。例えば、postgres に対するsql
を普通に書くとします。もし、これが SQLServer に DB が変わったとしましょう。postgres と SQLServer とでは、sql の書き方が異なる部分があるため、再度書き直す必要があります。しかし、python
コードで書いてしまえば勝手に補完をしてくれるのです。ただし、デメリットとして内部で実際にどのようなことをしているかが不明瞭になる、細かい処理を行うことが難しいといった部分もあります。もちろん、sqlalchemy
には、python
コードで書く方法と生 SQL を書く方法がどちらもあるので、そこは上手くやりましょう。今回は、こちらの都合で生 SQL でやる方法を先に紹介します。次回、python
コードを使った場合を紹介させて頂きます。
環境
- vscode(1.74.0)
- python(3.10.8)
- fastapi(0.87.0)
- sqlalchemy(1.4.44)
- postgres (15.1)
いざ実践
今回は、以下のテーブルを作成します。
departments
id | name |
---|---|
integer | varchar(10) |
users
id | user_name | departments_id |
---|---|---|
integer | varchar(10) | integer |
SQL 書くのが面倒な方用に、今回テーブル作成に使用した SQL を以下に貼っておきます。
テーブル作成に使用した SQL
create table departments ( id serial primary key, name varchar(10) );
create table users ( id serial, user_name varchar(10), departments_id int references departments(id) );
また、今回のファイル構成は以下のようになります。
fastapi-project
├─ schema
| ├─__init__.py
| ├─departments.py
| └─users.py
├─ DB
| ├─__init__.py
| └─detabase.py
├─ crud
| ├─__init__.py
| ├─crud_departments.py
| └─crud_users.py
├─ main.py
└─ __init__.py
前回と比較し、DB
とcrud
というフォルダが増えていますね。これらのフォルダについて簡単に説明を以下に載せておきます。
-
DB
:データベースとの接続を行う -
crud
:データベースとのやりとりを記述
まずは、データベースとの接続を行いましょう。(基本的に、書き方のみを説明させて頂きます。)
データベース接続
早速、データベース接続を行います。database.py
に移りましょう。
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
# postgresでの接続方法
SQLALCHEMY_DATABASE_URL = "postgresql://{ユーザー名}:{パスワード}@{ホスト名}:{ポート番号}/{データベース名}"
engine = create_engine(SQLALCHEMY_DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
SQLALCHEMY_DATABASE_URL
の部分だけ、今回自分が使用するデータベース情報に変更して頂ければ、データベースの記述は終了です。
何をしているかは、公式チュートリアルに詳しくあるため、省こうと思います。
次にmain.py
で、以下を記述しましょう。
from fastapi import FastAPI
from db.database import SessionLocal
app = FastAPI()
# get
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
以上で、データベース接続は完了になります。正直データベース接続さえできてしまえば、それ以降は簡単です。
次に crud を記述をしていきます。
crud 処理
crud 処理を記述していこうと思います。今回はget
とpost
のみにしておこうと思います。では、早速行っていきます。まずは、departments
に対する処理をしていきます。
get の処理
departments
のget
処理は以下のようになります。
from sqlalchemy.orm import Session
from sqlalchemy.sql import text
# 全データを取得する
def get_departments(db: Session):
sql = text("select * from departments;")
return db.execute(sql).all()
# 一部データのみ取得する
def get_department(id:int, db:Session):
sql = text(f"select * from departments where id = {id}")
return db.execute(sql).one()
get_departments
を解説します。get_departments
は引数にsqlalchemy.orm
からインポートしたSession
を型とするdb
を受け取ります。このdb
が、データベース情報になるものです。そして、2 行目で生SQL
を実行するための変数sql
を用意して、sqlalchemy.sql
からインポートしたtext
を使用し、text
の引数に今回行いたいSQL
の情報を与えています。あとは、SQL
を実行してデータを返せばいいだけです。データベースに対して、今回用意したSQL
を実行するには、最初に引数で受け取ったdb
を使います。db.execute(sql)
で、sql
を実行します。しかし、これだけではまだダメです。このexecute()
はResult
というオブジェクトをデータとして返します。つまり、結果として返すデータの形式を選択する必要があります。今回は、テーブル内にある全データを取得したいので、全データをList
形式で返すall()
を選択しました。get_departments
が理解できれば、二つ目の関数get_department
も理解できるかと思います。
Reslut
オブジェクトのメソッドについてのドキュメントを以下に載せておきます。
post の処理
次にpost
ですね。やり方は、前節のget の処理が理解できれば、難しくないです。post
のコードも引き続きcrud_departments.py
に記述します。以下が、post
に関するコードです。
def insert_department(name:str,db:Session):
sql = text(f"insert into departments (name) values('{name}');")
db.execute(sql)
db.commit()
sql2 = text("select * from departments order by id desc")
return db.execute(sql2).first()
では、解説です。前半は先ほどのget
の処理とほぼ変更はありません。SQL
の内容がinsert
処理になったぐらいですね。異なるのは、3 行目以降ですね。3 行目では、データベースに対してSQL
を実行しています。その後、データベースに対して情報を反映する為に、db.commit()
をする必要があります。これはpost
だけでなく、delete
、patch
に関しても同じです。必ずデータベースの情報が変化する際は、最後にコミットしましょう。データベースにデータを挿入するだけであれば、これで終わって頂いて構いません。しかし、データを挿入したなら、データが挿入されていることを確認したいので、最後にselect
で今回投げたデータを確認しましょう。(最後の部分は、独学でやってるのであってるか分かりません。。。post
なのに、select
しているから気持ち悪い。。。)
これで、post
の処理は終了です。これで、get
とpost
の関数がそれぞれ完成しました。次に、schema
フォルダにデータの返り値の型を定義しようと思います。
schema による型定義
ここでは、データベースに対してSQL
を投げた際に返ってきたデータの型定義を行います。これを行うことで、バグやエラーを極力減らすことが可能です。以下が、schema/departments.py
のコードです。
from pydantic import Field,BaseModel
class Departments(BaseModel):
id:int
name: str = Field(min_length=1,max_length=10)
ここに関しては、前回の記事を参考にして頂ければ理解できるかと思うので、解説は省かせていただきます。作成したテーブルのテーブル定義を参考にしながらコーディングすることがポイントです。
これで、下準備は完了です。main.py
に移りましょう。
main の処理
ここでは、main.py
の処理を記述していきます。以下が、main.py
のコードです。
app = FastAPI()
# get
from fastapi import FastAPI,Depends,HTTPException
from db.database import SessionLocal
from crud import crud_departments
from sqlalchemy.orm import Session
from schema import departments
from typing import List
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
# get
@app.get("/departments",response_model=List[departments.Departments])
async def get_departments(db:Session=Depends(get_db)):
return crud_departments.get_departments(db=db)
@app.get("/departments/{id}",response_model=departments.Departments)
async def get_department(id:int,db:Session=Depends(get_db)):
try:
result = crud_departments.get_department(id=id,db=db)
if result == None:
raise HTTPException(status_code=404, detail=f"{id}:Negative values are not accepted")
return result
except:
raise HTTPException(status_code=404,detail= f'{id} is 404 not found')
# insert
@app.post("/departments",response_model=departments.Departments)
async def insert_department(name:str,db:Session=Depends(get_db)):
return crud_departments.insert_department(name=name,db=db)
前回の記事を参考にして頂ければ、ここの処理も理解ができるかと思います。前回と違うところは、一部のget
の処理には例外処理を組み込んでいることです。try
やexcept
等のpython
の例外処理を行うためのコードが分からない方は、調べましょう。例外処理を記述する際は、fastapi
からインポートしているHTTPException
を利用しましょう。第一引数にはステータスコード、第二引数にはエラーメッセージを記述するだけです。これで、エラー原因を特定することができます。これがないとエラー原因が分かりにくくなる為、エラー原因を特定する為にも記述はしておきましょう。今回は、あくまで紹介の為このぐらいにしています。以上で、テーブルdepartments
に対する処理は終了です。実際に実行してみます。
実行してみる
ここでは、実行結果の確認を行いたいと思います。現在、テーブルにはデータがないためデータを 2 つ挿入したいと思います。試しに、sales
とmaker
という二つのデータを挿入してみます。では、サーバーを立ち上げます。コマンドラインでmain.py
があるディレクトリに、移りましょう。そして以下のコマンドを叩いてください。
uvicorn main:app --reload
これで、サーバーが起動します。サーバー起動後、(http://127.0.0.1:8000/docs)にアクセスしてください。以下の画面が表示されます。
では、post
メソッドを使用してsales
とmaker
というデータを挿入してみましょう。
私の方では、以下のレスポンスが返ってきてデータが挿入されていることを確認できました。
では、次にget
メソッドを使用してデータの取得を行います。
問題なくデータを取得できることを確認できました。また、id
を指定してデータの取得も問題なくできることを確認してください。
問題なく、データを取得できることを確認できました。
以上で、departments
に対する処理が完了しました。users
に対する処理は、ご自身で一度挑戦してみましょう。以下に答えのコードを全て貼っておきます。
答えのコード
from fastapi import FastAPI,Depends,HTTPException
from db.database import SessionLocal
from crud import crud_users,crud_departments
from sqlalchemy.orm import Session
from schema import users,departments
from typing import List
app = FastAPI()
# get
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
@app.get("/users", response_model=List[users.User])
async def get_users(db:Session=Depends(get_db)):
return crud_users.get_users(db=db)
@app.get("/users/{id}", response_model=users.User)
async def get_user(id:int,db:Session=Depends(get_db)):
try:
result = crud_users.get_user(id=id,db=db)
if result == None:
raise HTTPException(status_code=404, detail=f"{id}:Negative values are not accepted")
return result
except:
raise HTTPException(status_code=404,detail= f'{id} is 404 not found')
@app.get("/departments",response_model=List[departments.Departments])
async def get_departments(db:Session=Depends(get_db)):
return crud_departments.get_departments(db=db)
@app.get("/departments/{id}",response_model=departments.Departments)
async def get_department(id:int,db:Session=Depends(get_db)):
try:
result = crud_departments.get_department(id=id,db=db)
if result == None:
raise HTTPException(status_code=404, detail=f"{id}:Negative values are not accepted")
return result
except:
raise HTTPException(status_code=404,detail= f'{id} is 404 not found')
# insert
@app.post("/departments",response_model=departments.Departments)
async def insert_department(name:str,db:Session=Depends(get_db)):
return crud_departments.insert_department(name=name,db=db)
@app.post("/users",response_model=users.User)
async def insert_user(user_name: str, departments_id: int, db: Session=Depends(get_db)):
try:
result = crud_users.insert_user(user_name=user_name,departments_id=departments_id,db=db)
except:
raise HTTPException(status_code=404,detail= f'{departments_id} is 404 not found')
return result
from sqlalchemy.orm import Session
from sqlalchemy.sql import text
def get_users(db: Session):
sql = text("select * from users;")
return db.execute(sql).all()
def get_user(id:int,db: Session):
sql = text(f"select * from users where id = {id};")
return db.execute(sql).one()
def insert_user(user_name:str,departments_id:int,db:Session):
sql = text(f"insert into users (user_name,departments_id) values('{user_name}',{departments_id});")
db.execute(sql)
db.commit()
sql2 = text("select * from users order by id desc")
return db.execute(sql2).first()
from pydantic import BaseModel, Field
class User(BaseModel):
id: int
user_name: str = Field(min_length=1,max_length=10)
departments_id: int
最後に
前回に引き続き、fastapi
の記事でした。今回は、不規則でORM
を使用しない生SQL
を実行する処理について記事にしてみました。次回は、ORM
機能を使用して、データのやりとりをやってみようと思います。とりあえず、年末ギリギリに書き終えてよかった。。。(笑)
Discussion