# 1 implemented the processing for GET request. # 2 implemented processing for POST, PUT, and DELETE requests. This time, which will be \ # 3, we will rewrite all the processes implemented so far into DB-linked processes!
For details, see here.
.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': 'Title 1', 'body': 'Body 1'},
{'title': 'Title 2', 'body': 'Body 2'},
{'title': 'Title 3', 'body': 'Body 3'},
{'title': 'Title 4', 'body': 'Body 4'},
{'title': 'Title 5', 'body': '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()