Flask で REST API を構築する
さて、先ほど作成した SQLite データベース example.db
を使用して、Flask で REST API を構築しましょう。
app.py
ファイルに、必要なモジュールをインポートするための次のコードを追加します。
from flask import Flask, jsonify, request
import sqlite3
Flask アプリケーションを作成するための次のコードを追加します。
## Create the Flask application
app = Flask(__name__)
さて、REST API のエンドポイントを定義しましょう。
すべてのユーザーを取得するエンドポイント
## Endpoint to retrieve all users
@app.route('/users', methods=['GET'])
def get_users():
## Connect to the database
conn = sqlite3.connect('example.db')
cursor = conn.cursor()
## Execute the SQL query to retrieve all users
cursor.execute('SELECT * FROM users')
users = cursor.fetchall()
## Close the cursor and the database connection
cursor.close()
conn.close()
## Return the users as JSON
return jsonify(users)
このコードでは、まず connect()
メソッドを使用して SQLite データベースに接続します。次に、cursor()
メソッドを使用して SQL 文を実行するためのカーソルオブジェクトを作成します。そして、execute()
メソッドを使用して "users" テーブルからすべてのユーザーを取得する SQL クエリを実行します。最後に、close()
メソッドを使用してカーソルとデータベース接続を閉じ、jsonify()
メソッドを使用してユーザーを JSON 形式で返します。
ID で特定のユーザーを取得するエンドポイント
## Endpoint to retrieve a specific user by ID
@app.route('/users/<int:user_id>', methods=['GET'])
def get_user(user_id):
## Connect to the database
conn = sqlite3.connect('example.db')
cursor = conn.cursor()
## Execute the SQL query to retrieve the user by ID
cursor.execute('SELECT * FROM users WHERE id =?', (user_id,))
user = cursor.fetchone()
## Close the cursor and the database connection
cursor.close()
conn.close()
## Check if the user exists
if user:
return jsonify(user)
else:
return jsonify({'message': 'User not found'}), 404
このエンドポイントは、すべてのユーザーではなく ID で特定のユーザーを取得する点を除けば、前のエンドポイントと似ています。"users" テーブルから単一のユーザーを取得するために fetchone()
メソッドを使用します。
新しいユーザーを作成するエンドポイント
## Endpoint to create a new user
@app.route('/users', methods=['POST'])
def create_user():
## Get the user data from the request body
data = request.get_json()
name = data['name']
email = data['email']
## Connect to the database
conn = sqlite3.connect('example.db')
cursor = conn.cursor()
## Execute the SQL query to insert a new user
cursor.execute('INSERT INTO users (name, email) VALUES (?,?)', (name, email))
conn.commit()
## Close the cursor and the database connection
cursor.close()
conn.close()
## Return a success message
return jsonify({'message': 'User created successfully'})
このエンドポイントは、"users" テーブルに新しいユーザーを作成するために使用されます。get_json()
メソッドを使用してリクエストボディからユーザーデータを取得します。そして、execute()
メソッドを使用して新しいユーザーを "users" テーブルに挿入します。
既存のユーザーを更新するエンドポイント
## Endpoint to update an existing user
@app.route('/users/<int:user_id>', methods=['PUT'])
def update_user(user_id):
## Get the updated user data from the request body
data = request.get_json()
name = data['name']
email = data['email']
## Connect to the database
conn = sqlite3.connect('example.db')
cursor = conn.cursor()
## Execute the SQL query to update the user
cursor.execute('UPDATE users SET name =?, email =? WHERE id =?', (name, email, user_id))
conn.commit()
## Close the cursor and the database connection
cursor.close()
conn.close()
## Return a success message
return jsonify({'message': 'User updated successfully'})
このエンドポイントは、"users" テーブルの既存のユーザーを更新するために使用されます。get_json()
メソッドを使用してリクエストボディから更新されたユーザーデータを取得します。そして、execute()
メソッドを使用して "users" テーブルのユーザーを更新します。
ユーザーを削除するエンドポイント
## Endpoint to delete a user
@app.route('/users/<int:user_id>', methods=['DELETE'])
def delete_user(user_id):
## Connect to the database
conn = sqlite3.connect('example.db')
cursor = conn.cursor()
## Execute the SQL query to delete the user
cursor.execute('DELETE FROM users WHERE id =?', (user_id,))
conn.commit()
## Close the cursor and the database connection
cursor.close()
conn.close()
## Return a success message
return jsonify({'message': 'User deleted successfully'})
このエンドポイントは、"users" テーブルからユーザーを削除するために使用されます。execute()
メソッドを使用して "users" テーブルからユーザーを削除します。
最後に、Flask アプリケーションを実行するための次のコードを追加しましょう。
if __name__ == '__main__':
app.run(debug=True, host='0.0.0.0', port=8080)