【FastAPI入門】MｙSQL v8.0系と連携したCRUD処理の実装

services:
  db:
    image: mysql:8.3.0 
    ports:
      - 3316:3306
    volumes:
      - './db:/var/lib/mysql'  
    environment:
      MYSQL_ROOT_PASSWORD: root_pass
      MYSQL_DATABASE: st_db 
      MYSQL_USER: st_user
      MYSQL_PASSWORD: st_pass
  phpmyadmin:
    image: phpmyadmin/phpmyadmin:latest
    depends_on:
      - db
    environment:
      - PMA_ARBITRARY=1
      - PMA_HOSTS=db
      - PMA_USER=root
      - PMA_PASSWORD=root_pass
    ports:
      - 8092:80

pip install fastapi
pip install uvicorn[standard]
pip install mysql-connector 
pip install pymysql

uvicorn  server:app --reload --port 8000

mysql>  drop table if exists users;
mysql>  create table users (
    id int auto_increment PRIMARY KEY,
    nickname VARCHAR(50) NOT NULL UNIQUE,
    name VARCHAR(50) NOT NULL,
    password VARCHAR(100) not NULL,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP NOT NULL
)DEFAULT CHARSET=utf8mb4;

mysql> desc users;
+-------------+------------------+------+-----+-----------------------------+-------------------------------+
| Field           | Type                  | Null | Key | Default                              | Extra                                     |
+-------------+------------------+------+-----+------------------------------+------------------------------+
| id                  | int                      | NO   | PRI | NULL                                   | auto_increment              |
| nickname  | varchar(50)    | NO   | UNI | NULL                                  |                                                |
| name           | varchar(50)   | NO   |         | NULL                                    |                                                |
| password   | varchar(100) | NO  |         | NULL                                    |                                                |
| created_at | timestamp    | NO   |        | CURRENT_TIMESTAMP | DEFAULT_GENERATED |
+---------------+----------------+------+-----+-------------------------------+------------------------------+
5 rows in set (0.00 sec)

import requests

res = requests.get("http://localhost:8000/users")
print(res.status_code)
print(res.text)

# 実行結果

200
null

from fastapi import FastAPI
import mysql.connector as mysqlcon

app = FastAPI()

# 全ユーザー抽出
@app.get("/users")
def get_allusers():
    try:
        conn = mysqlcon.connect(
            host="localhost",
            port = 3316,
            user="root",　# ← Dockerコンテナではroot権限でないと接続できなかった!!
            password="root_pass",   # ← rootユーザーのpassword、テストなので直書きしてます。
            database = "st_db",
        )

        cur = conn.cursor()
        cur.execute("select * from users")
        rows = cur.fetchall()  # ← 全件抽出
        conn.close()
        print("成功")
        # FastAPIのレスポンスは通常、JSON形式で返されます。
        return rows

    except Exception as e:
        print(f"エラー発生しました: {e}")

# 実行結果

エラー発生しました: Authentication plugin 'caching_sha2_password' is not supported

from fastapi import FastAPI
#import mysql.connector as mysqlcon  # ← 使用せず
import pymysql  # ← 新規追加

@app.get("/users")
def get_allusers():
    try:
        #conn = mysqlcon.connect(   # ← 使用せず
        conn = pymysql.connect(       # ← こちらを採用
    ...省略...

# server.pyの実行結果

成功
INFO:     127.0.0.1:50404 - "GET /users HTTP/1.1" 200 OK

# client.pyの実行結果

200
[]

res = requests.post('http://localhost:8000/user/create',
            json={"nickname":"robinchan","name":"ニコ ロビン", "password": "pass123"}  )
print(res.status_code)
print(res.text)

# 実行結果
200
true

from pydantic import BaseModel 

class User(BaseModel):
    nickname : str
    name : str
    password : str

@app.post("/user/create")
def save_user(new_user :User):
    print(new_user)  # nickname='robin' name='ニコ ロビン' password='robin123'
    # 辞書に変換
    user_dict = dict(new_user)
    # タプルに変換
    user_tupple = (
        user_dict["nickname"],
        user_dict["name"],
        user_dict["password"]
    )
    print(user_tupple)  # ('robin', 'ニコ ロビン', 'robin123')
    try:
        conn = pymysql.connect(
            host="localhost",
            port = 3316,
            user="root",
            password="root_pass",
            database = "st_db",
        )

        cur = conn.cursor()
        cur.execute("insert into users (nickname, name, password) values(%s, %s, %s) ", user_tupple)
        conn.commit()
        conn.close()
        print("登録成功")
        return True

    except Exception as e:
        print(f"エラー発生しました: {e}")
        return False

user_id =3
res = requests.delete(f'http://localhost:8000/user/{user_id}/destroy')
print(res.status_code)
print(res.text)

# 削除
@app.delete("/user/{user_id}/destroy")
def delete_user_byid(user_id):
    id= int(user_id)
    try:
        conn = pymysql.connect(
            host="localhost",
            port = 3316,
            user="root",
            password="root_pass",
            database = "st_db",
        )

        cur = conn.cursor()
        cur.execute("select count(*) from users where id=%s",(id,))
        result = cur.fetchone()
        cnt = result[0]
        print(f"id={id} は {cnt} 件です")
        if cnt > 0:
            #削除
            print("削除します")
            cur.execute("delete from users where id = %s",(id,))
            conn.commit()
            conn.close()
            return True
        else :
            print("該当データがありません")
            conn.close()
            return False
        
    except Exception as e:
        print(f"エラー発生しました: {e}")
        return False

user_id=5
res = requests.put(f'http://localhost:8000/user/{user_id}/update',
            json={"nickname":"zoro","name":"ロロノア ゾロ", "password": "zoro123"}  )
print(res.status_code)
print(res.text)

@app.put("/user/{user_id}/update")
def update_user(user_id:int, newuser: User):
    id= int(user_id)
    newuser_dic = dict(newuser)
    newuser_tupple = (
        newuser_dic["nickname"],
        newuser_dic["name"],
        newuser_dic["password"],
        id
    )
    try:
        conn = pymysql.connect(
            host="localhost",
            port = 3316,
            user="root",
            password="root_pass",
            database = "st_db",
        )

        cur = conn.cursor()
        cur.execute("select count(*) from users where id=%s",(id,))
        result = cur.fetchone()
        cnt = result[0]
        print(f"id={id} は {cnt} 件です")
        if cnt > 0:
            #更新処理
            cur.execute("update users set nickname = %s, name = %s, password = %s
 　　　　　　　　　　　　　　　where id = %s" , newuser_tupple)
            conn.commit()
            conn.close()
            return True
        else:
            #何もしない
            return False
        
    except Exception as e:
        print(f"エラー発生しました: {e}")
        return False