Flaskチュートリアル#3 ~ DB連携編 ~

前回までのあらすじと今回実装するもの

#1ではGETリクエストに対する処理を実装しました。
#2ではPOST, PUT, DELETEリクエストに対する処理を実装しました。
#3となる今回はこれまで実装した処理を全てDB連携した処理に書き換えていきます！

詳しくはこちらで紹介しています。

ソースコード

.env

DB_USER=root
DB_PASS=*********

init_db.py

import sqlalchemy
import os
from os.path import join, dirname
from dotenv import load_dotenv

load_dotenv(verbose=True)
dotenv_path = join(dirname(__file__), '.env')
load_dotenv(dotenv_path)

DB_USER = os.environ.get('DB_USER')
DB_PASS = os.environ.get('DB_PASS')

url = f'mysql+mysqldb://{DB_USER}:{DB_PASS}@localhost/test?charset=utf8'

engine = sqlalchemy.create_engine(url, echo=True)

engine.execute(f'DROP TABLE IF EXISTS posts')

engine.execute('''
    CREATE TABLE posts (
        id INTEGER PRIMARY KEY AUTO_INCREMENT,
        title CHAR(30),
        body TEXT
    )
''')

db.py

import sqlalchemy
import os
from os.path import join, dirname
from dotenv import load_dotenv

load_dotenv(verbose=True)

dotenv_path = join(dirname(__file__), '.env')
load_dotenv(dotenv_path)

DB_USER = os.environ.get("DB_USER")
DB_PASS = os.environ.get("DB_PASS")


url = f'mysql+mysqldb://{DB_USER}:{DB_PASS}@localhost/test?charset=utf8'

engine = sqlalchemy.create_engine(url, echo=True)

app.py

from flask import Flask, jsonify, request
from db import engine
import json

app = Flask(__name__)

db_data = [
    {'title': 'タイトル1', 'body': '本文1'},
    {'title': 'タイトル2', 'body': '本文2'},
    {'title': 'タイトル3', 'body': '本文3'},
    {'title': 'タイトル4', 'body': '本文4'},
    {'title': 'タイトル5', 'body': '本文5'},
]


app.config['JSON_AS_ASCII'] = False 

@app.route('/', methods=['GET'])
def index():
    return 'hello world'


@app.route('/posts', methods=['GET'])
def get_all_posts():
    if 'limit' in request.args:
        limit = request.args['limit']
        posts = engine.execute(f'''
            SELECT * FROM posts
            LIMIT {int(limit)}
        ''')
    else:
        posts = engine.execute(f'''
            SELECT * FROM posts
        ''')
    result = [dict(row) for row in posts]
    return jsonify(result)


@app.route('/post/<id>', methods=['GET'])
def get_post(id):
    posts = engine.execute(f'''
        SELECT * FROM posts
        WHERE id={int(id)}
    ''')
    result = [dict(row) for row in posts]
    return jsonify(result)


@app.route('/post/add', methods=['POST'])
def create_post():
    post = request.json
    engine.execute(f'''
        INSERT INTO posts (title, body)
        VALUES ('{post['title']}', '{post['body']}')
    ''')
    return 'success'


@app.route('/post/update/<id>', methods=['PUT'])
def update_post(id):
    post = request.json
    engine.execute(f'''
        UPDATE posts 
        SET title='{post['title']}', body='{post['body']}'
        WHERE id={int(id)}
    ''')
    return 'success'


@app.route('/post/delete/<id>', methods=['DELETE'])
def delete_post(id):
    engine.execute(f'''
        DELETE FROM posts
        WHERE id={int(id)}
    ''')
    return 'success'


if __name__ == '__main__':
    app.debug = True
    app.run()