はじめに
- 前回、sqlalchemy-migrateを利用してUserテーブルを作成したので、そのユーザーテーブルにアクセスしてデータを返すサンプルを作成する px-wing.hatenablog.com
インストール
- 必要なライブラリは下記となります。利用しているデータベースはmysqlサーバーとなります。
pip install uvicorn pip install fastapi pip install pymysql pip install mysqlclient pip install sqlalchemy pip install sqlalchemy-migrate
コード
- DB接続部分のコード
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, scoped_session
ENGINE = create_engine('mysql://<ユーザー名>:<パスワード>@<DBのホスト名>:<DBポート番号>/<データベース名>',encoding="utf-8",echo=True)
session = scoped_session(
sessionmaker(
autocommit=False,
autoflush=False,
expire_on_commit=False,
bind=ENGINE
)
)
# modelで使用する
ModelBase = declarative_base()
# DB接続用のセッションクラス、インスタンスが作成されると接続する
ModelBase.query = session.query_property()
- Userテーブルのモデルは下記となります。ファイル名は「model.py」とします
from sqlalchemy import Column, Integer, String
from db import ModelBase
from sqlalchemy import Table, Column, MetaData,BIGINT, VARCHAR, DateTime, Text
import hashlib
from datetime import datetime
class User(ModelBase):
__tablename__ = 'users'
id = Column('id', BIGINT(), primary_key=True, autoincrement=True)
mail_address = Column('mail_address', VARCHAR(40), nullable=False, unique=True)
password = Column('password', VARCHAR(40), nullable=False)
last_login_at = Column('last_login_at', DateTime)
created_at = Column('created_at', DateTime, nullable=False, default=datetime.now())
updated_at = Column('updated_at', DateTime, nullable=False, default=datetime.now())
def __init__(self, mail_address, password):
self.mail_address = mail_address
# パスワードはハッシュ化して保存
self.password = hashlib.md5(password.encode()).hexdigest()
- メイン処理となるmain.pyの処理です。 http://localhost:8000/usersにアクセスするとusersテーブルに登録されているすべての情報を取得してjsonで返します。
from typing import List
from fastapi import FastAPI
from model import User
from db import session
from pydantic import BaseModel
import datetime
class UserSelect(BaseModel):
mail_address: str
password: str
last_login_at: datetime.datetime
created_at: datetime.datetime
updated_at: datetime.datetime
app = FastAPI(
title='FastAPI HOGE HOGE アプリケーション',
description='FastApiを利用したAPIアプリケーションを開発する',
version='0.0.1 beta'
)
@app.get("/users")
async def user_list(date: List[UserSelect]):
return session.query(User.mail_address,User.password,User.created_at,User.updated_at,User.last_login_at).all()
実行結果
http://localhost:8000/usersにアクセスすると下記のJSONが返ってくる
[
[
"hoge@example.com",
"0d32bced91aa5c2ee5696fc7995370ae",
"2020-11-20T08:16:47",
"2020-11-20T08:16:47",
"2020-11-20T22:19:39"
],
[
"hoge@example.com",
"0d32bced91aa5c2ee5696fc7995370ae",
"2020-11-19T23:21:34",
"2020-11-19T23:21:34",
"2020-11-20T22:19:39"
]
]
- DBに登録されている内容と同じ内容が変わってきていることを確認できる。
mysql> select * from users\G
*************************** 1. row ***************************
id: 1
mail_address: hoge@example.com
password: 0d32bced91aa5c2ee5696fc7995370ae
last_login_at: 2020-11-20 22:19:39
created_at: 2020-11-20 08:16:47
updated_at: 2020-11-20 08:16:47
*************************** 2. row ***************************
id: 2
mail_address: hoge@example.com
password: 0d32bced91aa5c2ee5696fc7995370ae
last_login_at: 2020-11-20 22:19:39
created_at: 2020-11-19 23:21:34
updated_at: 2020-11-19 23:21:34
2 rows in set (0.00 sec)
