Tutoriel Flask # 3 ~ DB Linkage ~

Synopsis jusqu'à la dernière fois et quoi mettre en œuvre cette fois

# 1 a implémenté le traitement des requêtes GET. # 2 a implémenté le traitement des requêtes POST, PUT et DELETE. Cette fois, qui sera \ # 3, nous réécrirons tous les processus implémentés jusqu'à présent dans les processus liés à la base de données!

Pour plus de détails, voir ici.

Code source

.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': 'Titre 1', 'body': 'Corps 1'},
    {'title': 'Titre 2', 'body': 'Corps 2'},
    {'title': 'Titre 3', 'body': 'Corps 3'},
    {'title': 'Titre 4', 'body': 'Corps 4'},
    {'title': 'Titre 5', 'body': 'Corps 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()