見出し画像

Pythonでやってみた１０：DBデータ管理＿担当者情報の一覧表示 Streamlit×FastAPI×SQLAlchemy

KIYO

概要

　業務でのデータ扱いに関して”あるある”を改善できるか模索中です。

【業務のデータ管理に関してのあるある？】
●業務に携わったであろう人から引継ぎがなく整理されていないサーバーを探させられる（口頭レベルの説明もしない）。
●退職者が多く、かつ引継ぎが適当なため後々欲しいデータがでてこない
●データの取り扱いが適当（サーバーにまともな名前も付けずに保管）なため必要なファイルが見当たらない（実は存在しないことも多々ある）
●各人でデータの取りまとめ方が異なるため①データの結合がしにくい、②そもそも必要な項目が抜けてる、③実験データか手計算かもわからない　など部内での共有化が難しい。
●ずっと古い知識・データを押し付けるやつへの証明データを出すのも手間
●非構造化データばかりであり将来的にAI・機械学習をするうえでデータ整理に死ぬほど時間がかかる（最悪できない）。

　SQLを使用したDBによる共有化・高速化×サーバー連動のためのAPI×可視化・BIツールのWebアプリ開発で何かできるか検討しました。
　第１弾として、発注したい業者の担当者を共有化するためのシステム開発にトライアルしました。

１．DBおよびWebアプリ設計

　複数の機能を実装していくため各工程で設計を実施します。

１－１．DB設計

　DB設計の思想は下記の通りにしました。

【DB設計の思想】
１．どのプロジェクトで何の製品を依頼したかをデータで管理する
２．備考以外の項目はNullは不可とする。備考もデータそのものはNullではなく、”無し”などに変換する。
３．業種と発注形態はマスタテーブルとする。

１－２．UI設計（フロントエンド）

　Webアプリによるフロントエンド（UI設計）の思想は下記の通りです。

【Webアプリ開発の設計思想】
１．画面は①データ登録用、②データ削除用、③データ検索用の3つ
（更新はなし：一回削除して新規で登録してもらう）
２．SQLでソートができるように変更ない部分（マスタデータ）はセレクトボックスで選択できるようにする。
３．製品品目に関しては過去のデータを参考にできるようにしたい。（セレクトボックスから取得してテキストボックスに入力）
４．ボタンの種類は「登録（CREATE）」、「削除（DELETE）」、「読み込み（READ）」とする。更新は今回は必須ではないのでスルーする。
５．出力をソートできるようにする。SQL側でやる方が処理速度が速いはずなのでできる限りソートはSQL側に任せる。

１－３．全体イメージ図

　今回はすべてローカルで実行しますがDBサーバーを別で用意することを想定してDB操作はFastAPIを用いたAPI実装したいと思います。
　よってDB接続はAPI操作のみで実行でき”from sql_app import xx”での操作(同じPC内での操作)はできないものと想定して実装します。

２．作成要領

　ほぼ自分の備忘録に近い内容です。（本職の方がいれば指導してほしい）

【１．設計】
基本的なDB、UI、Webアプリ構成の設計をする（１章参照）。　
【2.WebアプリのUI実装-Poc含む】
StreamlitでおおよそのUIをWebアプリ作成する->外観チェックのみでありボタン動作の実装は追って　
【3.バックエンド：DB/API】
●SQL Alchemyで作成するDB構成を作成する　
●FastAPIを用いてDBのデータを受け入れることができるGET/POST作成　
【4.Webアプリの機能実装（ボタンにHTTPクライアント追加）】
WebアプリのボタンにHTTPクライアント機能(requestsライブラリ)を追加　
【5. 調整】
●システムを起動　
●マスタテーブルの追加　
●動作の調整：SQLのSELECT文でデータ取得・表示を調整
【6.機能追加】
●パスワード機能追加
●データが0（初期状態）と登録後の動作に対する検証・対応

３．Webアプリ画面の作成(Streamlit)

　ブラウザベースでデータを見れるようにフロントエンドのWebアプリ開発としてStreamlitを使用します。

３－１．ページ選択用アプリのパッケージ作成

　今回は検索・登録・削除の３つの画面を作成していくため、各ページのモジュールをまとめてパッケージにしました。

【メインWebアプリ：Streamlit起動はこのモジュールを起動】
●webapp.py：サイドバーおよび下記パッケージ内のアプリを選択する

【サブアプリ：各ページの機能を全て集約したパッケージ】
●__init__.py：初期化モジュール（中身は空）
●utils.py：全モジュールで共通して使用する変数・関数を集めたモジュール
●app_show.py：DBから取得したデータを表示させる画面
●app_create.py：FastAPIへCREATE操作をPOSTするための画面
●app_delete.py：：FastAPIへDELETE操作をPOSTするための画面

　現段階では動作はしませんが”webapp.py”は下記の通り記載することでサイドバーから"subwebapp"パッケージ内の好きなページを選択できます。

[webapp.py]
import streamlit as st
import sys, os
from subwebapp import app_create, app_delete, app_show

#サイドバーで選択するページの設定
PAGES ={
    "データ検索(READ)": app_show,
    "データ登録(CREATE)": app_create,
    "データ削除(DELETE)": app_delete
}

st.sidebar.title("ページ設定")
selection = st.sidebar.radio("表示するページを設定してください。", list(PAGES.keys()))
page = PAGES[selection]
page.app()
[subwebapp/__init__.py] ※空ファイル
[subwebapp/utils.py]
ROOT_API = "http://127.0.0.1:8000" #APIのエンドポイント

３－２．データ入力フォームの作成：st.form()

　DBに登録したいデータを入力する場合はブラウザ操作中に毎回更新待ち時間が発生するよりデータ入力後にまとめて更新する方が便利です。Streamlitでは「st.form()」を用いて機能を実装します。

３－３．データテーブル表示：st.table(), st.dataframe()

　DBデータは構造化データのためテーブル形式Excelのような形で取得できます。テーブルをWebアプリに実装するにはst.tableやst.dataframe()を使用します。（公式Doc）

３－４．アプリのイメージ図

　検索、作成、削除画面のイメージはそれぞれ下記としてコードを実装していきます。この時点ではボタン機能の実装はないため、完成コードを最終章にまとめて記載しました。

４．DBおよびAPI作成(SQL Alchemy・FastAPI)

　DBおよびAPIを操作するパッケージを作成していきます。内容は基本的にはFastAPIの公式Docsを参照しております。

４－１．DB/APIパッケージ作成

　DB・API操作を実施するパッケージの作成をします。ファイル構成はFastAPI記事の「SQL (Relational) Databases」を参考にしたいと思います。

　一部は個人的に理解しにくいので勝手に名前を変えておりますが、基本的には記事と名前をそろえて作成する方が他の人が見やすいと思います。

【sql_appライブラリのファイル（モジュール）構成】
●__init__.py：パッケージを初期化する（※空ファイルでよい）
●crud_DB.py：SQL Alchemy（Session）によるCRUD操作の関数を記載
●database_DB.py：SQL AlchemyのSession管理およびテーブル作成のためのBaseクラスの作成
●main_API.py：FastAPIでのHTTPクライアント(GET/POST)のAPI作成
●tables_DB.py：SQL Alchemyテーブル構造のクラス定義(models.pyに相当)
●schemas_API.py：FastAPIのPOST処理時のデータの型ヒントを定義

４－２．SQL AlchemyのSesssion管理：database_DB.py

　SQL AlchemyのSession管理およびテーブル作成のためのBaseクラスの作成用モジュールを作成していきます。

[sql_app/database_DB.py]
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

#SQLAlchemyのDB作成・接続のためのURL作成
SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db" #同じディレクトリにsql_app.dbを作成
# SQLALCHEMY_DATABASE_URL = "postgresql://user:password@postgresserver/db" #PostgreSQLを使用したい人向け

engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False} #SQLiteを使用するため引数：connect_args={"check_same_thread": False}を設定
)

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base() #DBのテーブル作成時に継承させるためのクラス

４－３．SQL AlchemyのDBテーブル作成：tables_DB.py

　次にSQL AlchemyのDB用テーブルのクラスをtables_DBに記載します。

[sql_app/tables_DB.py]
from sqlalchemy import Boolean, Column, ForeignKey, Integer, String
from sqlalchemy.orm import relationship

from .database_DB import Base #同じディレクトリのdatabase_DB.pyのBaseクラスをインポート


class Industry(Base):
    __tablename__ = "industry"

    id = Column(Integer, primary_key=True, index=True)
    industry = Column(String, unique=True, index=True) #重複無し
    
    ordercompany = relationship("Ordercompany", back_populates="industry") #親テーブル側の外部キーを設定

class Ordertype(Base):
    __tablename__ = "ordertype"

    id = Column(Integer, primary_key=True, index=True)
    ordertype = Column(String, unique=True, index=True) #重複無し
    
    ordercompany = relationship("Ordercompany", back_populates="ordertype") #親テーブル側の外部キーを設定
    
class Ordercompany(Base):
    __tablename__ = "ordercompany"

    id = Column(Integer, primary_key=True, index=True)
    id_industry = Column(Integer, ForeignKey("industry.id")) #外部キーでID管理
    id_ordertype = Column(Integer, ForeignKey("ordertype.id")) #外部キーでID管理
    companyname = Column(String, index=True) 
    personname = Column(String) 
    phonenum = Column(String) 
    email = Column(String) 
    producttype = Column(String) 
    project = Column(String) 
    remark = Column(String) 
    
    industry = relationship("Industry", back_populates="ordercompany") #relationshipsにより、industryテーブルと関連付け
    ordertype = relationship("Ordertype", back_populates="ordercompany") #relationshipsにより、ordertypeテーブルと関連付け

【参考説明】
「from .database_DB import Base」は先ほど作成したモジュール内のBaseをimportしています。

４－４．FastAPIの型ヒントを定義：schemas_API.py

　FastAPIでPOST処理を受け付ける時のデータの型ヒントを定義します。tables_DBにおけるSQL Alchemyのデータ型はBaseを継承しましたが、FastAPIではPydanticを継承して使用します。

【モジュール作成時の注意事項】
１．SQL AlchemyでDB登録(CREATE)するときは”autoincrement”機能を使用するため"id"の値は不要である。
　しかしデータ取得(READ)の時は"id"を引数で指定する必要がある。
ー＞CREATE用とREAD用でデータ型を分けて記載する必要がある。
ー＞公式を参考にして①継承用のBase、②CREATE用、③FastAPIのPOST用（READ用)の３つのクラスを作成します。
２．POSTではなくクエリパラメータで渡す変数はBaseModel継承クラスには記載しなくてよい（Ordercompany参照）。
ー＞クエリパラメータはURLに記載されるためpasswordを変数に渡すような処理には使用してはいけない（対象：DeleteOrderCompany）。
３．FastAPIでクエリパラメータを使用時は公式参照

[sql_app/schemas_API.py]
from typing import List, Optional  #python 3.9以上ではListは省略可, 10以上ではこの行は省略可※その代わり下記記載が変わる
from pydantic import BaseModel

#Industry(業種)クラス
class IndustryBase(BaseModel):
    industry: str
    
class IndustryCreate(IndustryBase):
    pass

class Industry(IndustryBase):
    id: int

    class Config:
        orm_mode = True #SQLAlchemyのORMを受け入れる

#Ordertype(発注形態)クラス
class OrdertypeBase(BaseModel):
    ordertype: str

class OrdertypeCreate(OrdertypeBase):
    pass

class Ordertype(OrdertypeBase):
    id: int

    class Config:
        orm_mode = True


#Ordercompany(発注業者管理)（メイン)
class OrdercompanyBase(BaseModel):
    companyname: str
    personname: str
    phonenum: str
    email: str
    producttype: str
    project: str
    remark: Optional[str] = None
    # remark: str | None = None #Python3.10以上の記載方法

class OrdercompanyCreate(OrdercompanyBase):
    pass

class Ordercompany(OrdercompanyBase):
    id: int
    id_industry: int #Industryクラスの外部キー
    id_ordertype: int #Ordertypeクラスの外部キー

    class Config:
        orm_mode = True

#DELETE操作専用
class DeleteOrderCompany(BaseModel):
    id: int
    password: str

４－５．DBのCRUD操作：crud_DB.py

　crud_DB.pyにはDB操作を記載します。実装内容は下記の通りです。

【CREATE操作】
●発注業者のデータ登録
【READ操作】
●マスタデータ取得
●登録データ（発注業者）の取得
【UPDATE操作】
●なし
【DELETE操作】
●登録したデータの削除（※DELTEのみパスワードが必要に設計）

　DB操作ではDBテーブル用クラス情報がある「tables_DB.py」とFastAPIでの型ヒントを記載した「schemas_API.py」をインポートして使用します。

[sql_app/crud_DB.py]
from sqlalchemy.orm import Session #型ヒント用にSQLAlchemyのSessionクラス
from . import tables_DB #SQLAlchemyのテーブル
from . import schemas_API #APIの型ヒント

#READ操作
def get_industry(db: Session, skip: int=0, limit: int=100):
    return db.query(tables_DB.Industry).offset(skip).limit(limit).all()

def get_ordertype(db: Session, skip: int=0, limit: int=100):
    return db.query(tables_DB.Ordertype).offset(skip).limit(limit).all()

def get_ordercompanies(db: Session, skip: int =0, limit: int = 100):
    return db.query(tables_DB.Ordercompany).offset(skip).limit(limit).all()

#CREATE操作
def create_industry(db: Session, industry: schemas_API.IndustryCreate):
    db_industry = tables_DB.Industry(
        industry = industry.industry,
    )
    db.add(db_industry)
    db.commit()
    db.refresh(db_industry)
    return db_industry

def create_ordertype(db: Session, ordertype: schemas_API.OrdertypeCreate):
    db_ordertype = tables_DB.Ordertype(**ordertype.dict()) #省略型の記法
    db.add(db_ordertype)
    db.commit()
    db.refresh(db_ordertype)
    return db_ordertype

def create_ordercompany(db: Session, ordercompany: schemas_API.OrdercompanyCreate, id_industry: int, id_ordertype: int):
    #DB登録用のインスタンスを生成
    db_ordercompany = tables_DB.Ordercompany(
        **ordercompany.dict(),
        id_industry = id_industry, #Pydanticで定義されていないキー
        id_ordertype = id_ordertype #Pydanticで定義されていないキー
    )
    
    # db_ordercompany = tables_DB.Ordercompany(**ordercompany.dict()) #省略型の記法
        
    db.add(db_ordercompany) #DB(session)に追加
    db.commit() #DBに反映
    db.refresh #インスタンスを更新->DBから新しい値を取得可能
    return db_ordercompany

#DELETE操作
def delete_ordercompany(db: Session, datas: schemas_API.DeleteOrderCompany):
    if datas.password == 'KIYO':
        db_ordercompany = db.query(tables_DB.Ordercompany).filter(tables_DB.Ordercompany.id == datas.id).first()
        db.delete(db_ordercompany)
        db.commit()
        return db_ordercompany
    else:
        return 'パスワードが間違っています。'

【備忘録】
　「SQLAlchemyでmigrationにAlembicを使用する」と記載ある。今回は使う必要性はなさそうだけど今後学習が必要かもしれないので参考を記載。

４－６．API開発ーGET/POSTメソッド：main_API.py

　最後にFastAPIを用いてAPI実装をします。

【API実装の思想】
●GETメソッドはDBのREAD操作のみ->基本的な内容で実装可能
●POSTメソッドに関してクライアント側システム使用者は発注業者登録しかしないが、マスタデータ(業種、発注形態）を登録できるようにDBテーブル数だけCREATE操作のPOSTメソッドを作成する。
●DELETE操作に関してはpasswordを引数に入れるようにした。HTTPメソッド時にpasswordがURLに見えないようにPOSTメソッドで実行。

[sql_app/main_API.py]
import string
from typing import List #Python3.6 ver.
from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session #SQL AlchemyのDB型

from . import crud_DB, tables_DB, schemas_API
from .database_DB import SessionLocal, engine

tables_DB.Base.metadata.create_all(bind=engine) #テーブル作成

app = FastAPI() #FastAPIのapp作成

# Dependency
def get_db():
    db = SessionLocal() #DB接続
    try: 
        yield db #DBリクエスト
    finally:
        db.close() #エラーが出ても必ずDB切断


#GETメソッド->特定のパラメータが不要なためDBのREAD操作を実行
@app.get("/industry/", response_model=List[schemas_API.Industry]) #response_modelで戻り値の型を指定
def read_industry(skip: int =0, limit: int =100, db: Session = Depends(get_db)):
    return crud_DB.get_industry(db, skip=skip, limit=limit)

@app.get("/ordertype/", response_model=List[schemas_API.Ordertype])
def read_ordertype(skip: int =0, limit: int =100, db: Session = Depends(get_db)):
    return crud_DB.get_ordertype(db, skip=skip, limit=limit)

@app.get("/ordercompany/", response_model=List[schemas_API.Ordercompany])
def read_ordercompany(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    ordercompany = crud_DB.get_ordercompanies(db, skip=skip, limit=limit)
    return ordercompany


#POSTメソッド
@app.post("/industry/", response_model=schemas_API.Industry)
def create_industry(industry: schemas_API.IndustryCreate, db: Session = Depends(get_db)):
    return crud_DB.create_industry(db, industry)

@app.post("/ordertype/", response_model=schemas_API.Ordertype)
def create_ordertype(ordertype: schemas_API.OrdertypeCreate, db: Session = Depends(get_db)):
    return crud_DB.create_ordertype(db, ordertype)

#POST-CREATE操作
@app.post("/ordercompany/", response_model=schemas_API.Ordercompany)
def create_ordercompany(id_industry: int, id_ordertype:int, ordercompany: schemas_API.OrdercompanyCreate, db: Session = Depends(get_db)):
    return crud_DB.create_ordercompany(db=db, ordercompany=ordercompany, id_industry=id_industry, id_ordertype=id_ordertype)


#POST-DELETE操作
@app.post("/ordercompany/delete/")
def delete_ordercompany(datas: schemas_API.DeleteOrderCompany , db: Session = Depends(get_db)):
    return crud_DB.delete_ordercompany(db, datas)

【def vs async defの選択】
async defは非同期処理を実現しますが、公式Docsより「SQL Alchemyは直接”await”使用に対して互換性がない」との記載があるためdefで作成しました（十分に意味が理解できていないため紹介のみ）。

５．DB登録/Webアプリの機能実装

５－１．システム起動(Webアプリ/API)

　ターミナルを２つ用意してStreamlit/FastAPIをそれぞれ起動します。

[Terminal_1] webappモジュールを起動
streamlit run webapp.py

[Terminal_2] sql_appパッケージ内のmain_APIモジュールにあるappを起動
uvicorn sql_app.main_API:app --reload

　main_APIを起動すると「tables_DB.Base.metadata.create_all(bind=engine) 」も合わせて実行されるためDB作成・接続が実行されます。

５－２．マスタテーブルの登録

　WebアプリとDB/APIライブラリは同じディレクトリにありますが、あくまで別サーバーを想定します。よってDB操作はすべてAPIで実施します。

　今回は使用者がマスタデータを追加できない（ブラウザ上で操作できない）ようにするため、個別でマスターデータを追加します。今回は適当にJupyter(下図のipynbファイル)で処理しました。

　まずrequestライブラリを用いてGETメソッドによるDBデータ取得(READ)を実施して①データが空であること、②通信できていることを確認します。

[IN]
import requests 
import json

root_api = "http://127.0.0.1:8000" 
res = requests.get(root_api + "/industry/")
print(res.status_code) #200なら通信成功
print(res.json())

[Out]
200
[]

　次にPOSTメソッドでDBのCREATE操作をするAPIを叩くことでデータ登録・確認をします。※for文で回すために冗長なコードになってますが{"industry": data}を1回ずつ処理しても登録できます。

[In]
import requests 
import json

industry = {1:"メーカー",
            2:"商社"}
datas = [{"industry": v} for v in industry.values()] #[{'industry': 'メーカー'}, {'industry': '商社'}]

for data in datas:
    res = requests.post(root_api + "/industry/", data=json.dumps(data))
    # print(res.status_code) #通信確認用

root_api = "http://127.0.0.1:8000" 
res = requests.get(root_api + "/industry/")
# print(res.status_code) #200なら通信成功
print(res.json())

[Out]
#response_model=List[schemas_API.Industry]で出力される
[{'industry': 'メーカー', 'id': 1}, {'industry': '商社', 'id': 2}]

　上を参考にして”Ordertype”のマスタデータも登録します。

[In]
import requests 
import json

ordertype={1:"直接発注",
           2:"商社経由"}

datas = [{"ordertype": v} for v in ordertype.values()] #[{'ordertype': '直接発注'}, {'ordertype': '商社経由'}]

for data in datas:
    res = requests.post(root_api + "/ordertype/", data=json.dumps(data))
    
root_api = "http://127.0.0.1:8000" 
res = requests.get(root_api + "/ordertype/")
print(res.status_code) #200なら通信成功
print(res.json())

datas_ordertype = {i['id']: i['ordertype'] for i in res.json()}
datas_ordertype

[Out]
200
[{'ordertype': '直接発注', 'id': 1}, {'ordertype': '商社経由', 'id': 2}] #response_model=List[schemas_API.Ordertype] 
{1: '直接発注', 2: '商社経由'}

【備忘録：処理動作の確認用※実装はWebアプリにする】
　APIの戻り値はListであるため{id:data}の辞書形式に変換します。またデータをidで登録するためにkey:valueを逆にした辞書も作成します。

[In]
datas_industry = {i['id']: i['industry'] for i in res.json()}
industry2id = {v: k for k, v in datas_industry.items()}

datas_industry , industry2id

[Out]
{1: 'メーカー', 2: '商社'}
{'メーカー': 1, '商社': 2}

５－３．Webアプリの機能実装

　３章ではUIのみ実装(タイトル／見出し、フォームの入力画面、ボタン）しましたが、最後に機能を実装していきます。

【機能実装一覧】
●HTTPクライアントでAPIを叩く->データの読み込み・登録・削除
●読み込んだデータをきれいに表示(st.table, st.dataframe)
●データがない時に出てくるエラーの対処
●DELETEのpassword機能の実装

　上記をまとめると結果として下記コードになりました。

[subwebapp/app_show]
def app():
    import streamlit as st
    import pandas as pd
    import requests
    import json
    from .utils import ROOT_API #app()の中に入れないとエラー
    
    
    st.title('メーカー・担当者管理システム')
    st.markdown('## データ検索')
    
    #IndustryとOrdertypeのマスタデータを辞書型で取得-API使用
    res = requests.get(ROOT_API + "/industry/") 
    industry = {i['id']: i['industry'] for i in res.json()} #{1:"メーカー", 2:"商社"}
    
    res = requests.get(ROOT_API + "/ordertype/")
    ordertype = {i['id']: i['ordertype'] for i in res.json()} #{1:"直接発注", 2:"商社経由"}
    
    #Ordercompanyデータの表示
    ROOT_API = "http://127.0.0.1:8000" 
    res = requests.get(ROOT_API + "/ordercompany/") 
    
    #データなし(空リスト出力時はエラーになるためif文)
    if not res.json():
        st.write('DBに登録されているデータは存在しません。')
    else:
        #ID(整数)が入っているデータを値に変換
        df = pd.DataFrame(res.json())
        df['id_industry'] = df['id_industry'].map(industry)
        df['id_ordertype'] = df['id_ordertype'].map(ordertype)
        
        #dataframeのカラム名を変更
        renames = {
        'id': 'ID',
        'id_industry': '業種',
        'id_ordertype': '発注形態',
        'companyname': '会社名',
        'personname': '担当社',
        'phonenum': '連絡先(E-mail)',
        'email': '連絡先(TEL)',
        'producttype': '取扱い製品',
        'project': 'プロジェクト名',
        'remark': '備考'
        }
        df.rename(columns=renames, inplace=True) #カラム名変更, inplaceでdfを上書き
        
        #カラムの列順を変更
        colnums = list(range(len(df.columns)))
        colnums = colnums[-3:] + colnums[:-3] #output->[7, 8, 9, 0, 1, 2, 3, 4, 5, 6]
        df = df.iloc[:, colnums] #後ろ3の列を前に移動
        
        #データ表示
        st.table(df.set_index('ID'))
[subwebapp/app_create]
import streamlit as st
import pandas as pd
import requests
import json
from .utils import ROOT_API

def app():
    st.title('メーカー・担当者管理システム')
    st.markdown('## データ入力画面')
    
    #IndustryとOrdertypeのマスタデータを辞書型で取得-API使用
    res = requests.get(ROOT_API + "/industry/") 
    industry = {i['id']: i['industry'] for i in res.json()} #{1:"メーカー", 2:"商社"}
    industry2id = {v: k for k, v in industry.items()} #key:valueを反転->値からidを取得
    
    res = requests.get(ROOT_API + "/ordertype/")
    ordertype = {i['id']: i['ordertype'] for i in res.json()} #{1:"直接発注", 2:"商社経由"}
    ordertype2id = {v: k for k, v in ordertype.items()} #key:valueを反転->値からidを取得
    
    
    #データフォームUIの作成
    with st.form("input_data"):
        in_industry : str = st.selectbox("業種", list(industry.values())) #業種選択
        in_ordertype : str = st.selectbox("発注形態", list(ordertype.values())) #業種選択
        in_companyname : str = st.text_input(label="会社名", value="株式会社")
        in_personname : str = st.text_input(label="担当者")
        in_TEL : str = st.text_input(label="TEL(個人がない場合は代表)")
        in_Email : str = st.text_input(label="E-mail(個人がない場合は代表)", value="@.co.jp")
        in_productname : str = st.text_input(label="取扱い製品")
        in_projectname : str = st.text_input(label="関連PJ(依頼時のPJ)")
        in_remark : str = st.text_area(label="備考(未入力でもよい)", value="無し") #初期値は"無し"という文字列
        
        data = {
            'companyname' : in_companyname,
            'personname' : in_personname,
            'phonenum' : in_TEL,
            'email' : in_Email,
            'producttype' : in_productname,
            'project' : in_projectname,
            'remark' : in_remark
        }

        # Every form must have a submit button.
        submitted = st.form_submit_button("データ登録ボタン")
        if submitted:
            params = { 
            'id_industry' : industry2id[in_industry],
            'id_ordertype' : ordertype2id[in_ordertype]}
            
            res = requests.post(ROOT_API + "/ordercompany/",
                                data=json.dumps(data),
                                params=params) 
            
            if res.status_code == 200:
                st.write('データを登録しました。')
                st.json(res.json())
[subwebapp/app_delete]
from email.policy import default
import streamlit as st
import streamlit as st
import pandas as pd
import requests
import json

def app():
    st.title('メーカー・担当者管理システム')
    st.markdown('## データ削除画面')
    
    #Ordercompanyデータの表示
    ROOT_API = "http://127.0.0.1:8000" 
    res = requests.get(ROOT_API + "/ordercompany/") 
    
        #データなし(空リスト出力時はエラーになるためif文)
    if not res.json():
        st.write('DBに登録されているデータは存在しません。')
    else:
        df = pd.DataFrame(res.json())
        
        #カラムの列順を変更
        colnums = list(range(len(df.columns)))
        colnums = colnums[-3:] + colnums[:-3] #output->[7, 8, 9, 0, 1, 2, 3, 4, 5, 6]
        df = df.iloc[:, colnums] #後ろ3の列を前に移動

        st.markdown('### データ選択')
    
        with st.form("show_data"):
            id_select = st.selectbox("削除するデータを選択", list(df['id'])) #業種選択
            
            submitted = st.form_submit_button("データ選択")
            if submitted:
                df = df[df['id']==id_select] #選択したID
                st.dataframe(df)
        
        with st.form("delete_data"):
            st.write(f'削除するデータid：{id_select}')
            text_pass = st.text_input("パスワード入力", type="password") #パスワード入力
            # Every form must have a submit button.
            submitted = st.form_submit_button("データ削除")
            if submitted:
                password = 'KIYO'
                if text_pass == password:
                    params = {'id': id_select,
                            'password': password}
                    res = requests.post(ROOT_API + "/ordercompany/delete/",
                        data=json.dumps(params)) 
                    
                    if res.status_code==200:
                        st.write(f"データid:{id_select}を削除しました。")
                        st.write(res.json())
                else:
                    st.write('パスワードが間違っています。')

６．動作確認

 最後にシステムを起動させた状態で動作チェックします。

７．全コード

７－１．フロントエンド：Webアプリ側

[webapp.py]
import streamlit as st
import sys, os
from subwebapp import app_create, app_delete, app_show

#サイドバーで選択するページの設定
PAGES ={
    "データ検索(READ)": app_show,
    "データ登録(CREATE)": app_create,
    "データ削除(DELETE)": app_delete
}

st.sidebar.title("ページ設定")
selection = st.sidebar.radio("表示するページを設定してください。", list(PAGES.keys()))
page = PAGES[selection]
page.app()
[subwebapp/__init__.py] ※空ファイル
[subwebapp/utils.py]
ROOT_API = "http://127.0.0.1:8000" #APIのエンドポイント
[subwebapp/app_create.py]
import streamlit as st
import pandas as pd
import requests
import json
from .utils import ROOT_API

def app():
    st.title('メーカー・担当者管理システム')
    st.markdown('## データ入力画面')
    
    #IndustryとOrdertypeのマスタデータを辞書型で取得-API使用
    res = requests.get(ROOT_API + "/industry/") 
    industry = {i['id']: i['industry'] for i in res.json()} #{1:"メーカー", 2:"商社"}
    industry2id = {v: k for k, v in industry.items()} #key:valueを反転->値からidを取得
    
    res = requests.get(ROOT_API + "/ordertype/")
    ordertype = {i['id']: i['ordertype'] for i in res.json()} #{1:"直接発注", 2:"商社経由"}
    ordertype2id = {v: k for k, v in ordertype.items()} #key:valueを反転->値からidを取得
    
    
    #データフォームUIの作成
    with st.form("input_data"):
        in_industry : str = st.selectbox("業種", list(industry.values())) #業種選択
        in_ordertype : str = st.selectbox("発注形態", list(ordertype.values())) #業種選択
        in_companyname : str = st.text_input(label="会社名", value="株式会社")
        in_personname : str = st.text_input(label="担当者")
        in_TEL : str = st.text_input(label="TEL(個人がない場合は代表)")
        in_Email : str = st.text_input(label="E-mail(個人がない場合は代表)", value="@.co.jp")
        in_productname : str = st.text_input(label="取扱い製品")
        in_projectname : str = st.text_input(label="関連PJ(依頼時のPJ)")
        in_remark : str = st.text_area(label="備考(未入力でもよい)", value="無し") #初期値は"無し"という文字列
        
        data = {
            'companyname' : in_companyname,
            'personname' : in_personname,
            'phonenum' : in_TEL,
            'email' : in_Email,
            'producttype' : in_productname,
            'project' : in_projectname,
            'remark' : in_remark
        }

        # Every form must have a submit button.
        submitted = st.form_submit_button("データ登録ボタン")
        if submitted:
            params = { 
            'id_industry' : industry2id[in_industry],
            'id_ordertype' : ordertype2id[in_ordertype]}
            
            res = requests.post(ROOT_API + "/ordercompany/",
                                data=json.dumps(data),
                                params=params) 
            
            if res.status_code == 200:
                st.write('データを登録しました。')
                st.json(res.json())
[subwebapp/app_delete.py]
from email.policy import default
import streamlit as st
import streamlit as st
import pandas as pd
import requests
import json

def app():
    st.title('メーカー・担当者管理システム')
    st.markdown('## データ削除画面')
    
    #Ordercompanyデータの表示
    ROOT_API = "http://127.0.0.1:8000" 
    res = requests.get(ROOT_API + "/ordercompany/") 
    
        #データなし(空リスト出力時はエラーになるためif文)
    if not res.json():
        st.write('DBに登録されているデータは存在しません。')
    else:
        df = pd.DataFrame(res.json())
        
        #カラムの列順を変更
        colnums = list(range(len(df.columns)))
        colnums = colnums[-3:] + colnums[:-3] #output->[7, 8, 9, 0, 1, 2, 3, 4, 5, 6]
        df = df.iloc[:, colnums] #後ろ3の列を前に移動

        st.markdown('### データ選択')
    
        with st.form("show_data"):
            id_select = st.selectbox("削除するデータを選択", list(df['id'])) #業種選択
            
            submitted = st.form_submit_button("データ選択")
            if submitted:
                df = df[df['id']==id_select] #選択したID
                st.dataframe(df)
        
        with st.form("delete_data"):
            st.write(f'削除するデータid：{id_select}')
            text_pass = st.text_input("パスワード入力", type="password") #パスワード入力
            # Every form must have a submit button.
            submitted = st.form_submit_button("データ削除")
            if submitted:
                password = 'KIYO'
                if text_pass == password:
                    params = {'id': id_select,
                            'password': password}
                    res = requests.post(ROOT_API + "/ordercompany/delete/",
                        data=json.dumps(params)) 
                    
                    if res.status_code==200:
                        st.write(f"データid:{id_select}を削除しました。")
                        st.write(res.json())
                else:
                    st.write('パスワードが間違っています。')
[subwebapp/app_show.py]
def app():
    import streamlit as st
    import pandas as pd
    import requests
    import json
    from .utils import ROOT_API #app()の中に入れないとエラー
    
    
    st.title('メーカー・担当者管理システム')
    st.markdown('## データ検索')
    
    #IndustryとOrdertypeのマスタデータを辞書型で取得-API使用
    res = requests.get(ROOT_API + "/industry/") 
    industry = {i['id']: i['industry'] for i in res.json()} #{1:"メーカー", 2:"商社"}
    
    res = requests.get(ROOT_API + "/ordertype/")
    ordertype = {i['id']: i['ordertype'] for i in res.json()} #{1:"直接発注", 2:"商社経由"}
    
    #Ordercompanyデータの表示
    ROOT_API = "http://127.0.0.1:8000" 
    res = requests.get(ROOT_API + "/ordercompany/") 
    
    #データなし(空リスト出力時はエラーになるためif文)
    if not res.json():
        st.write('DBに登録されているデータは存在しません。')
    else:
        #ID(整数)が入っているデータを値に変換
        df = pd.DataFrame(res.json())
        df['id_industry'] = df['id_industry'].map(industry)
        df['id_ordertype'] = df['id_ordertype'].map(ordertype)
        
        #dataframeのカラム名を変更
        renames = {
        'id': 'ID',
        'id_industry': '業種',
        'id_ordertype': '発注形態',
        'companyname': '会社名',
        'personname': '担当社',
        'phonenum': '連絡先(E-mail)',
        'email': '連絡先(TEL)',
        'producttype': '取扱い製品',
        'project': 'プロジェクト名',
        'remark': '備考'
        }
        df.rename(columns=renames, inplace=True) #カラム名変更, inplaceでdfを上書き
        
        #カラムの列順を変更
        colnums = list(range(len(df.columns)))
        colnums = colnums[-3:] + colnums[:-3] #output->[7, 8, 9, 0, 1, 2, 3, 4, 5, 6]
        df = df.iloc[:, colnums] #後ろ3の列を前に移動
        
        #データ表示
        st.table(df.set_index('ID'))

７－２．バックエンド：DB・API

[sql_app/__init__.py] ※空ファイル
[sql_app/database_DB.py]
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

#SQLAlchemyのDB作成・接続のためのURL作成
SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db" #同じディレクトリにsql_app.dbを作成
# SQLALCHEMY_DATABASE_URL = "postgresql://user:password@postgresserver/db" #PostgreSQLを使用したい人向け

engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False} #SQLiteを使用するため引数：connect_args={"check_same_thread": False}を設定
)

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base() #DBのテーブル作成時に継承させるためのクラス
[sql_app/tables_DB.py]
from sqlalchemy import Boolean, Column, ForeignKey, Integer, String
from sqlalchemy.orm import relationship

from .database_DB import Base #同じディレクトリのdatabase_DB.pyのBaseクラスをインポート


class Industry(Base):
    __tablename__ = "industry"

    id = Column(Integer, primary_key=True, index=True)
    industry = Column(String, unique=True, index=True) #重複無し
    
    ordercompany = relationship("Ordercompany", back_populates="industry") #親テーブル側の外部キーを設定

class Ordertype(Base):
    __tablename__ = "ordertype"

    id = Column(Integer, primary_key=True, index=True)
    ordertype = Column(String, unique=True, index=True) #重複無し
    
    ordercompany = relationship("Ordercompany", back_populates="ordertype") #親テーブル側の外部キーを設定
    
class Ordercompany(Base):
    __tablename__ = "ordercompany"

    id = Column(Integer, primary_key=True, index=True)
    id_industry = Column(Integer, ForeignKey("industry.id")) #外部キーでID管理
    id_ordertype = Column(Integer, ForeignKey("ordertype.id")) #外部キーでID管理
    companyname = Column(String, index=True) 
    personname = Column(String) 
    phonenum = Column(String) 
    email = Column(String) 
    producttype = Column(String) 
    project = Column(String) 
    remark = Column(String) 
    
    industry = relationship("Industry", back_populates="ordercompany") #relationshipsにより、industryテーブルと関連付け
    ordertype = relationship("Ordertype", back_populates="ordercompany") #relationshipsにより、ordertypeテーブルと関連付け
[sql_app/schemas_API.py]
from typing import List, Optional  #python 3.9以上ではListは省略可, 10以上ではこの行は省略可※その代わり下記記載が変わる
from pydantic import BaseModel

#Industry(業種)クラス
class IndustryBase(BaseModel):
    industry: str
    
class IndustryCreate(IndustryBase):
    pass

class Industry(IndustryBase):
    id: int

    class Config:
        orm_mode = True #SQLAlchemyのORMを受け入れる

#Ordertype(発注形態)クラス
class OrdertypeBase(BaseModel):
    ordertype: str

class OrdertypeCreate(OrdertypeBase):
    pass

class Ordertype(OrdertypeBase):
    id: int

    class Config:
        orm_mode = True


#Ordercompany(発注業者管理)（メイン)
class OrdercompanyBase(BaseModel):
    companyname: str
    personname: str
    phonenum: str
    email: str
    producttype: str
    project: str
    remark: Optional[str] = None
    # remark: str | None = None #Python3.10以上の記載方法

class OrdercompanyCreate(OrdercompanyBase):
    pass

class Ordercompany(OrdercompanyBase):
    id: int
    id_industry: int #Industryクラスの外部キー
    id_ordertype: int #Ordertypeクラスの外部キー

    class Config:
        orm_mode = True

#DELETE操作専用
class DeleteOrderCompany(BaseModel):
    id: int
    password: str
[sql_app/crud_DB.py]
from sqlalchemy.orm import Session #型ヒント用にSQLAlchemyのSessionクラス
from . import tables_DB #SQLAlchemyのテーブル
from . import schemas_API #APIの型ヒント

#READ操作
def get_industry(db: Session, skip: int=0, limit: int=100):
    return db.query(tables_DB.Industry).offset(skip).limit(limit).all()

def get_ordertype(db: Session, skip: int=0, limit: int=100):
    return db.query(tables_DB.Ordertype).offset(skip).limit(limit).all()

def get_ordercompanies(db: Session, skip: int =0, limit: int = 100):
    return db.query(tables_DB.Ordercompany).offset(skip).limit(limit).all()

#CREATE操作
def create_industry(db: Session, industry: schemas_API.IndustryCreate):
    db_industry = tables_DB.Industry(
        industry = industry.industry,
    )
    db.add(db_industry)
    db.commit()
    db.refresh(db_industry)
    return db_industry

def create_ordertype(db: Session, ordertype: schemas_API.OrdertypeCreate):
    db_ordertype = tables_DB.Ordertype(**ordertype.dict()) #省略型の記法
    db.add(db_ordertype)
    db.commit()
    db.refresh(db_ordertype)
    return db_ordertype

def create_ordercompany(db: Session, ordercompany: schemas_API.OrdercompanyCreate, id_industry: int, id_ordertype: int):
    #DB登録用のインスタンスを生成
    db_ordercompany = tables_DB.Ordercompany(
        **ordercompany.dict(),
        id_industry = id_industry, #Pydanticで定義されていないキー
        id_ordertype = id_ordertype #Pydanticで定義されていないキー
    )
    
    # db_ordercompany = tables_DB.Ordercompany(**ordercompany.dict()) #省略型の記法
        
    db.add(db_ordercompany) #DB(session)に追加
    db.commit() #DBに反映
    db.refresh #インスタンスを更新->DBから新しい値を取得可能
    return db_ordercompany

#DELETE操作
def delete_ordercompany(db: Session, datas: schemas_API.DeleteOrderCompany):
    if datas.password == 'KIYO':
        db_ordercompany = db.query(tables_DB.Ordercompany).filter(tables_DB.Ordercompany.id == datas.id).first()
        db.delete(db_ordercompany)
        db.commit()
        return db_ordercompany
    else:
        return 'パスワードが間違っています。'
[sql_app/main_API.py]
import string
from typing import List #Python3.6 ver.
from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session #SQL AlchemyのDB型

from . import crud_DB, tables_DB, schemas_API
from .database_DB import SessionLocal, engine

tables_DB.Base.metadata.create_all(bind=engine) #テーブル作成

app = FastAPI() #FastAPIのapp作成

# Dependency
def get_db():
    db = SessionLocal() #DB接続
    try: 
        yield db #DBリクエスト
    finally:
        db.close() #エラーが出ても必ずDB切断


#GETメソッド->特定のパラメータが不要なためDBのREAD操作を実行
@app.get("/industry/", response_model=List[schemas_API.Industry]) #response_modelで戻り値の型を指定
def read_industry(skip: int =0, limit: int =100, db: Session = Depends(get_db)):
    return crud_DB.get_industry(db, skip=skip, limit=limit)

@app.get("/ordertype/", response_model=List[schemas_API.Ordertype])
def read_ordertype(skip: int =0, limit: int =100, db: Session = Depends(get_db)):
    return crud_DB.get_ordertype(db, skip=skip, limit=limit)

@app.get("/ordercompany/", response_model=List[schemas_API.Ordercompany])
def read_ordercompany(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    ordercompany = crud_DB.get_ordercompanies(db, skip=skip, limit=limit)
    return ordercompany


#POSTメソッド
@app.post("/industry/", response_model=schemas_API.Industry)
def create_industry(industry: schemas_API.IndustryCreate, db: Session = Depends(get_db)):
    return crud_DB.create_industry(db, industry)

@app.post("/ordertype/", response_model=schemas_API.Ordertype)
def create_ordertype(ordertype: schemas_API.OrdertypeCreate, db: Session = Depends(get_db)):
    return crud_DB.create_ordertype(db, ordertype)

#POST-CREATE操作
@app.post("/ordercompany/", response_model=schemas_API.Ordercompany)
def create_ordercompany(id_industry: int, id_ordertype:int, ordercompany: schemas_API.OrdercompanyCreate, db: Session = Depends(get_db)):
    return crud_DB.create_ordercompany(db=db, ordercompany=ordercompany, id_industry=id_industry, id_ordertype=id_ordertype)


#POST-DELETE操作
@app.post("/ordercompany/delete/")
def delete_ordercompany(datas: schemas_API.DeleteOrderCompany , db: Session = Depends(get_db)):
    return crud_DB.delete_ordercompany(db, datas)

次回からの注意点

　エラーで死にそうだったので自分用の備忘録

【コード作成の注意点】
●基礎コード記法の勉強しなおせ（特に命名規則）！まずはPEP 8に従おう。（[Pythonコーディング規約]PEP8を読み解く）
●一般的なクラス名はパスカルケース※(誤：Ordertype、正：OrderType)
●変数名をもっとわかりやすいものにする。(業種の選択肢はindustryよりnameの方がよいないか？）
●最初は会社名にUniqueかけてたけど、同じ会社で別担当もあるためtables_DBのUniqueを外したけど一度DBファイルを作ると途中では変更できない？と思うのでもっと設計を念入りにしておく。
●データが無い場合と1個以上ある場合で処理が変わるため注意。（APIから取得したデータを使用するとデータが存在しない(初期)はエラー）
●FastAPIのAPIドキュメントは死ぬほど便利なので活用する。特にdocsは直接APIたたけるから便利だけどPOSTしたときにDB登録処理も実行される。

参考資料

あとがき

　とりあえずDB操作はシリーズとして何個か作る予定です。

●筋トレ管理：LINE通知＋グラフ化（可視化）
●（取得ではなく）提出した見積もりの価格管理
●メーカーからの見積もり管理（上に乗せられるかな？）
●QAリスト管理表：質問に対する回答をメールベースでなくシステムベースへ変更して、かつデータベース化


　作りたいものが増えるほど必要なスキルのハードルが急激に上がるので結構つらい。

この記事が気に入ったらサポートをしてみませんか？