Construir uma API REST em Flask
Agora, vamos construir uma API REST em Flask usando o banco de dados SQLite example.db que criamos anteriormente.
No arquivo app.py, adicione o seguinte código para importar os módulos necessários:
from flask import Flask, jsonify, request
import sqlite3
Adicione o seguinte código para criar a aplicação Flask:
## Create the Flask application
app = Flask(__name__)
Agora, vamos definir os endpoints para nossa API REST:
Endpoint para recuperar todos os usuários
## 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)
Neste código, primeiro conectamos ao banco de dados SQLite usando o método connect(). Em seguida, criamos um objeto cursor para executar as instruções SQL usando o método cursor(). Depois, executamos a consulta SQL para recuperar todos os usuários da tabela "users" usando o método execute(). Finalmente, fechamos o cursor e a conexão com o banco de dados usando o método close() e retornamos os usuários como JSON usando o método jsonify().
Endpoint para recuperar um usuário específico por 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
Este endpoint é semelhante ao anterior, exceto que ele recupera um usuário específico por ID em vez de todos os usuários. Usamos o método fetchone() para recuperar um único usuário da tabela "users".
Endpoint para criar um novo usuário
## 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'})
Este endpoint é usado para criar um novo usuário na tabela "users". Usamos o método get_json() para obter os dados do usuário do corpo da requisição. Em seguida, usamos o método execute() para inserir o novo usuário na tabela "users".
Endpoint para atualizar um usuário existente
## 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'})
Este endpoint é usado para atualizar um usuário existente na tabela "users". Usamos o método get_json() para obter os dados atualizados do usuário do corpo da requisição. Em seguida, usamos o método execute() para atualizar o usuário na tabela "users".
Endpoint para deletar um usuário
## 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'})
Este endpoint é usado para deletar um usuário da tabela "users". Usamos o método execute() para deletar o usuário da tabela "users".
Finalmente, vamos adicionar o seguinte código para executar a aplicação Flask:
if __name__ == '__main__':
app.run(debug=True, host='0.0.0.0', port=8080)
