[PYTHON] Flask Tutorial # 3 ~ DB Linkage ~

Synopse bis zum letzten Mal und was diesmal implementiert werden soll

# 1 implementierte die Verarbeitung für GET-Anforderungen. # 2 hat die Verarbeitung für POST-, PUT- und DELETE-Anforderungen implementiert. Dieses Mal, das wird \ # 3 sein, werden wir alle bisher implementierten Prozesse in DB-verknüpfte Prozesse umschreiben!

Einzelheiten finden Sie unter hier.

Quellcode

.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': 'Titel 1', 'body': 'Körper 1'},
    {'title': '', 'body': 'Körper 2'},
    {'title': 'Titel 3', 'body': 'Körper 3'},
    {'title': 'Titel 4', 'body': 'Körper 4'},
    {'title': 'Titel 5', 'body': 'Körper 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()