Fast APIでWebAPIを作ってみる③ #Docker

FastAPIを利用してWebAPIを作っていくシリーズの2回目です。

前回作成した、Docker上にpostgres環境にFastAPIでアクセスする機能を追加していきます。
Docker上にPostgres環境を作成する記事は以下を参照してください。
Fast APIでWebAPIを作ってみる②

開発環境

	バージョン
Python	3.10.4
FastAPI	0.78.0
Uvicorn	0.18.1
Docker	20.10.16
Postgres	14.4
SQLAlchemy　追加	1.4.39

今回のお題

今回は
　①Postgresqlへのアクセス処理の実装
　②DAO／DTOの実装
　③データ取得部分の実装
　④動作検証

辺りまでを実装していきます。

実装

フォルダ構成は以下の通りです。

フォルダ構成.

┣━backend/
┃　　┣━app
┃　　┃　┣━api
┃　　┃　┃　┗━routes
┃　　┃　┣━core
┃　　┃　┣━db　　　　　　　　　　# 追加
┃　　┃　┗━models　　　　　　　　# 追加
┃　　┃　　┣━DAO　　　　　　　　 #追加
┃　　┃　　┗━DTO　　　　　　　　 #追加
┃　　┣━Dockerfile
┃　　┗━requirements.txt
┣━script
┗━docker-compose.yml

①Postgresqlへのアクセス処理の実装

先ずは、docker上で動作しているpostgresqlへのアクセス処理を実装していきます。
appフォルダ内にある、dbフォルダ内に作成します。

dbaccess.py

from sqlalchemy import create_engine

class DBAccess():

    # DB接続処理
    def connect_database():
        try:
            # DB接続文字列の作成
            # 接続文字列は、postgresql://接続ユーザ名:接続PW@サーバ名:port番号
            engine = create_engine('postgresql://postgres:postgres@host.docker.internal:5436/Incident')

            # DBへの接続
            conn = engine.connect()
            
        except Exception as err:
            # 接続時にエラーとなった場合、エラーを返す。
            print(err)
        finally:
            print("Successful Database Access!")
        
        return conn

    # DB切断処理
    def disconnect_database(conn):
        conn.close()

②DAO／DTOの実装

MA_CODEテーブルに登録されているデータを取得した際、JSON形式で返すためのMODELとなります。

ma_code_model.py

from xmlrpc.client import boolean
from datetime import datetime
from pydantic import BaseModel

class Select_MA_CODE(BaseModel):
    id:int
    code_1:str
    code_2:str
    code_nm_1:str
    code_nm_2:str
    Active_flg:bool
    createdby:str
    created_date:datetime
    updated_by:str
    updated_date:datetime
    update_counter:int

③データ取得部分の実装

今回は、ORMを利用せず、生SQLを実行する形をとりたいと思います。
※ただ、ORMが苦手なだけ。そのうち、勉強して記事にします。

MaCodeRepository.py

from typing import List
from app.models.DTO.ma_code_model import Select_MA_CODE
from app.db.dbaccess import DBAccess
from sqlalchemy.sql import text

class MaCodeRepository():
    # MA_CODEの内容を全件取得する。
    def selectallcode():
        # Databaseへのアクセス
        con = DBAccess.connect_database()

        allcodedata = List[Select_MA_CODE]
        allcodedata = []
        query = text("SELECT * FROM MA_CODE;")

        try:
            rows = con.execute(query)
            for row in rows:
                allcodedata.append(row)
                
        except Exception as err:
            print(err)
        
        finally:
            DBAccess.disconnect_database(con)

        return allcodedata

④動作検証

以下のコマンドを実行し、Dockerを起動させます。

$ docker-compose up

以下の内容が表示されたら、正常にdockerが動作したことになります。

postgres_test | 2022-07-07 22:06:55.963 JST [21] LOG:  database system was shut down at 2022-07-07 22:06:51 JST
postgres_test | 2022-07-07 22:06:55.970 JST [1] LOG:  database system is ready to accept connections
server_1  | INFO:     Will watch for changes in these directories: ['/backend']
server_1  | INFO:     Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)
server_1  | INFO:     Started reloader process [1] using StatReload
server_1  | INFO:     Started server process [8]
server_1  | INFO:     Waiting for application startup.
server_1  | INFO:     Application startup complete.

以下のURLを叩いて以下の画面を表示させます。
http://localhost:8000/docs

「Try it out」ボタンをクリックすると、「execute」ボタンが表示されるので、それをクリックしてください。

画面下に結果が表示されたら、正常にMA_CODEテーブルの値が取得できたことになります。

今回のソースコードは以下のURLに上げています。
https://github.com/NoriStock1983/Incident_src

今回はこれで以上となります。

次回は、ＪＷＴを利用した認証機能の実装を行いたいと思います。

参考にしたサイト

https://nmomos.com/tips/2021/01/23/fastapi-docker-1/
https://fastapi.tiangolo.com/ja/

Fast APIでWebAPIを作ってみる③