🚌

ハンズオンで学ぶ、初心者向けのFastAPI～その2～

2022/12/31に公開

Python

FastAPI

tech

はじめに

前回に続き第二回目のfastapiに関する記事です。前回の記事を未読の方は、前回の記事を読んで頂ければと思います。

今回は、DB との接続を行いデータの処理をやってみようと思います。まず今回、新しく使用するフレームワークを紹介しようと思います。

今回新しく使用するフレームワーク

今回、新しく使用するフレームワークは前回の記事でもチラッと出しましたがsqlalchemyというフレームワークを使用します。（下記は、sqlalchemyの公式ドキュメントです。）
https://docs.sqlalchemy.org/en/14/

こちらのフレームワークは、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に移りましょう。

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で、以下を記述しましょう。

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処理は以下のようになります。

crud_departments.py

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に関するコードです。

crud_departments.py

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のコードです。

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のコードです。

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に対する処理は、ご自身で一度挑戦してみましょう。以下に答えのコードを全て貼っておきます。

答えのコード

main.py

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

crud/crud_users.py

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()

schema/users.py

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機能を使用して、データのやりとりをやってみようと思います。とりあえず、年末ギリギリに書き終えてよかった。。。(笑)

はじめに

今回新しく使用するフレームワーク

環境

いざ実践

データベース接続

crud 処理

get の処理

post の処理

schema による型定義

main の処理

実行してみる

最後に

Discussion