Streamlit でWeb APIを呼び出してアプリを公開

以前にFAST API(Deta cloud)を使ってWeb APIを公開する方法とStreamlit sharingを使ってWebアプリを公開する方法に関する記事を書きました。

今回は、2つのサービスを使って、簡易アプリを構築できるということを紹介致します。

最後にStreamlit sharingで問題なく動作することを確認してます。

1.予約システムの概要
2.Deta cloud でWeb APIを用意する
3.Streamlit でWeb APIを呼び出す
4.Streamlit sharingデプロイ後の動作確認
5.まとめ

1.予約システムの概要

簡易的に会議室の予約をするWebアプリを作ります。

用意するWeb API

method	URI	概要
GET	/users	ユーザ情報を全件取得
POST	/users	ユーザ情報を1件登録
GET	/rooms	会議室情報を全件取得
POST	/rooms	会議室情報を1件登録
GET	/bookings	予約情報を全件取得
POST	/bookings	予約情報を1件登録

用意する画面

ユーザー登録画面
会議室登録画面
会議室予約画面

用意するテーブル

users
rooms
bookings

capacity: 会議室の席数(定員)
reserved_num: 予約人数 (capacity >= reserved_num)

予約画面では、ユーザーと会議室を選択して、指定した時間に予約が取れるといった感じです。予約した内容は、一覧として予約登録画面にも表示します。

登録処理や一覧に表示するデータの取得には、Deta cloudを利用してWeb APIを用意します。

Streamlitでは、ユーザーの画面操作に応じて、Web APIを呼び出して、登録やデータ取得を行います。

2.Deta cloud でWeb APIを用意する

無料でサクッとWeb APIを公開できる Deta cloudを利用します。

requirements.txt

fastapi

main.py

from fastapi import FastAPI
from fastapi.responses import JSONResponse
from pydantic import BaseModel
from deta import Deta
import json

# localの場合は、project keyが必要だが、Deta base上ではいらない
deta = Deta()
users = deta.Base("fastapi-users")
rooms = deta.Base("fastapi-rooms")
bookings = deta.Base("fastapi-bookings")

app = FastAPI()

class User(BaseModel):
  name: str
  age: int
  hometown: str

class Room(BaseModel):
  room_name: str
  capacity: int

class Booking(BaseModel):
  user_key: str
  room_key: str
  reserved_num: int
  start_date_time: str
  end_date_time: str

@app.get("/users")
def read_user():
  return next(users.fetch())

@app.post("/users",status_code=200)
def create_user(user: User):
  user= users.put(user.dict())
  return json.dumps(user)

@app.get("/rooms")
def read_room():
  return next(rooms.fetch())

@app.post("/rooms",status_code=200)
def create_room(room: Room):
  room= rooms.put(room.dict())
  return json.dumps(room)

@app.get("/bookings")
def read_booking():
  return next(bookings.fetch())

@app.post("/bookings",status_code=200)
def create_booking(booking: Booking):
  booking= bookings.put(booking.dict())
  return json.dumps(booking)

deta.Baseでテーブルを定義
User, Room, Booking クラスを定義
それぞれのテーブルとクラスを使って、getとpostメソッドを定義
※各メソッドはjsonを返すようにしている

Deta cloud上へのデプロイと、Swaggerを使ったAPIの確認は、以下の記事にて紹介しています。

Python FAST API Deta cloudを使って簡単にWeb APIを公開する

FAST APIとDeta cloudを使って簡単にWeb APIを公開する方法をご紹介致します。Deta cloudは無料で利用することができるホスティングサービスです。Detaを使って簡易なWeb APIを素早くデプロイすることができます。

3.Streamlit でWeb APIを呼び出す

Deta cloudにデプロイしたWeb APIをStreamlit側で呼び出すようにします。

フォルダ構成

├─ app.py
├─ multiapp.py
├─ requirements.txt
├─ apps
      │─ users.py
      │─ rooms.py
      │─ bookings.py

requirements.txt

click<=8.0.4
streamlit

clickとそのバージョン指定は、おそらく必要になるかと思います。

clickをrequirements.txtに追加しなかったとき、こちらのようなエラーが出るのではないかと思います。

app.py

import streamlitas st
from multiappimport MultiApp
from appsimport users, rooms, bookings# import your app modules here

app= MultiApp()

app.add_app("users", users.app)
app.add_app("rooms", rooms.app)
app.add_app("bookings", bookings.app)

app.run()

app.pyではmultiapp.pyで用意しているadd_app関数を読んで、実行してます。

multiapp.py

import streamlitas st

class MultiApp:
    def __init__(self):
        self.apps= []

    def add_app(self,title,func):
        self.apps.append({
            "title": title,
            "function": func
        })

    def run(self):
        app= st.sidebar.selectbox(
            'choose your page',
            self.apps,
            format_func=lambda app: app['title'])
        app['function']()

複数ファイルでも構成できるようにmultiappを用意します。

apps/users.py

import streamlitas st 
import requests
import json

endpoint= 'https://[your domain].deta.dev'

def app():
  st.title('ユーザー登録')

  with st.form(key='user'):
    name:str = st.text_input('ユーザー名',max_chars=12)
    age:int = st.text_input('年齢',max_chars=3)
    hometown:str = st.text_input('出身',max_chars=15)
    data= {
      'name': name,
      'age': age,
      'hometown': hometown
    }
    submit_button= st.form_submit_button(label='登録')

  if submit_button:
    url= endpoint+ '/users'
    res= requests.post(
      url,
      data = json.dumps(data)
    )

    if res.status_code== 200:
      st.success('ユーザー登録完了')

endpointが Deta cloudでデプロイした時に生成されたURLです。

/users (post)でユーザー登録を行います。

rooms.py

import streamlitas st 
import requests
import json

endpoint= 'https://[your domain].deta.dev'

def app():
  st.title('会議室登録')

  with st.form(key='room'):
    room_name:str = st.text_input('会議室名',max_chars=12)
    capacity:int = st.number_input('定員',step=1)
    data= {
      'room_name': room_name,
      'capacity': capacity
    }
    submit_button= st.form_submit_button(label='登録')

  if submit_button:
    url= endpoint+ '/rooms'
    res= requests.post(
      url,
      data = json.dumps(data)
    )

    if res.status_code== 200:
      st.success('会議室登録完了')

/rooms (post)で会議室登録を行います。

apps/booking.py

import streamlitas st 
import requests
import json
import pandasas pd
import datetime

endpoint= 'https://[your domain].deta.dev'

def app():
  st.title('会議室予約')
 
  # ユーザー一覧を取得
  url_users= endpoint+ '/users'
  res_user= requests.get(url_users)
  users= res_user.json()

  users_dict=[]
  for userin users:
    user_dict= {}
    user_dict['name']= user['name']
    user_dict['age']= user['age']
    user_dict['hometown']= user['hometown']
    users_dict.append(user_dict)

  user_list= pd.DataFrame(users_dict) 

  st.write('### ユーザー一覧')
  st.table(user_list)

  users_name= {}
  for userin users :
    users_name[user['name']]= user['key'] 

  # 会議室一覧を取得
  url_rooms= endpoint+ '/rooms'
  res_room= requests.get(url_rooms)
  rooms= res_room.json()

  rooms_dict=[]
  for roomin rooms:
    room_dict= {}
    room_dict['room name']= room['room_name']
    room_dict['capacity']= room['capacity']
    rooms_dict.append(room_dict)

  room_list= pd.DataFrame(rooms_dict)

  st.write('### 会議室一覧')
  st.table(room_list)
 
  # 予約一覧を取得
  url_bookings= endpoint+ '/bookings'
  res_booking= requests.get(url_bookings)
  bookings= res_booking.json()

  users_key= {}
  for userin users:
    users_key[user['key']]= user['name']

  rooms_key= {}
  for roomin rooms:
    rooms_key[room['key']]= {
      'room_name': room['room_name'],
      'capacity': room['capacity']
    }
 
  bookings_dict=[]
  for bookingin bookings:
    booking_dict= {}
    booking_dict['name']= users_key[booking['user_key']]
    booking_dict['room name']= rooms_key[booking['room_key']]['room_name']
    booking_dict['reserved num']= booking['reserved_num']
    booking_dict['start time']= datetime.datetime.fromisoformat(booking['start_date_time']).strftime('%Y/%m/%d %H:%M')
    booking_dict['end time']= datetime.datetime.fromisoformat(booking['end_date_time']).strftime('%Y/%m/%d %H:%M')
    bookings_dict.append(booking_dict)

  booking_list= pd.DataFrame(bookings_dict)   
  
  st.write('### 予約一覧')
  st.table(booking_list)

  room_dict= {}
  for roomin rooms :
    room_dict[room['room_name']]= {
      'room_key': room['key'],
      'capacity': room['capacity']
    }

  with st.form(key='booking'):
    username:str = st.selectbox('予約者名', users_name.keys())
    roomname:str = st.selectbox('会議室名', room_dict.keys())
    reserved_num:int = st.number_input('予約人数',step=1,min_value=1)
    date= st.date_input('日付を入力',min_value=datetime.date.today())
    start_time= st.time_input('開始時刻: ',value=datetime.time(hour=9,minute=0))
    end_time= st.time_input('終了時刻: ',value=datetime.time(hour=20,minute=0))
 
    submit_button= st.form_submit_button(label='予約')

  if submit_button:
    user_key:int = users_name[username]
    room_key:int = room_dict[roomname]['room_key']
    capacity:int = room_dict[roomname]['capacity']
    data= {
      'user_key': user_key,
      'room_key': room_key,
      'reserved_num': reserved_num,
      'start_date_time': datetime.datetime(
        year=date.year,
        month=date.month,
        day=date.day,
        hour=start_time.hour,
        minute=start_time.minute
      ).isoformat(),
      'end_date_time': datetime.datetime(
        year=date.year,
        month=date.month,
        day=date.day,
        hour=end_time.hour,
        minute=end_time.minute
      ).isoformat()
    }

    # 定員以下の場合、予約登録
    if reserved_num> capacity:
      st.error(f'{roomname}の定員は{capacity}名です。')
    # 開始時刻>=終了時刻
    elif start_time>= end_time:
      st.error('開始時刻が終了時刻を越えています')
    elif start_time< datetime.time(hour=9,minute=0,second=0)or end_time> datetime.time(hour=20,minute=0):
      st.error('利用時間は、9:00～20:00になります。')
    else:
      url= endpoint+ '/bookings'
      res= requests.post(
        url,
        data = json.dumps(data)
      )
      if res.status_code== 200:
        st.success('予約完了しました。')
      # st.write(res.status_code)
      elif res.status_code== 404 and res.json()['detail']== 'Already booked':
        st.error('指定の時間には既に予約が入っています')