Tutorial: Web Services em Python para Todas as Tabelas da Loja de Açaí
Neste tutorial, aprenderemos a criar uma API REST simples em Python utilizando o Flask para consumir as tabelas do banco de dados de uma loja de açaí. Serão abordadas as tabelas:
- Clientes
- Produtos
- Materiais
- Vendas
- Consumo
Utilizaremos os verbos HTTP GET, POST, PUT e DELETE para realizar operações de leitura, criação, atualização e exclusão dos registros.
Requisitos
- Python 3 instalado
- Pacote Flask (instalável via
pip install flask)
- Banco de dados SQLite (
loja_acai.db) já criado com as tabelas definidas conforme o tutorial anterior.
Estrutura das Tabelas
Para referência, veja abaixo a estrutura básica de cada tabela:
-- Clientes:
CREATE TABLE clientes (
id_cliente INTEGER PRIMARY KEY AUTOINCREMENT,
nome TEXT NOT NULL,
telefone TEXT,
email TEXT
);
-- Produtos:
CREATE TABLE produtos (
id_produto INTEGER PRIMARY KEY AUTOINCREMENT,
nome TEXT NOT NULL,
preco REAL NOT NULL
);
-- Materiais:
CREATE TABLE materiais (
id_material INTEGER PRIMARY KEY AUTOINCREMENT,
nome TEXT NOT NULL,
custo REAL
);
-- Vendas:
CREATE TABLE vendas (
id_venda INTEGER PRIMARY KEY AUTOINCREMENT,
id_cliente INTEGER,
data_venda TEXT,
total REAL,
FOREIGN KEY (id_cliente) REFERENCES clientes(id_cliente)
);
-- Consumo:
CREATE TABLE consumo (
id_consumo INTEGER PRIMARY KEY AUTOINCREMENT,
id_venda INTEGER,
id_produto INTEGER,
quantidade INTEGER,
FOREIGN KEY (id_venda) REFERENCES vendas(id_venda),
FOREIGN KEY (id_produto) REFERENCES produtos(id_produto)
);
Com as tabelas criadas, já podemos construir nossa API.
Código Completo do Web Service
O código a seguir implementa os endpoints para cada uma das tabelas, seguindo o mesmo padrão para facilitar a manutenção e a compreensão:
from flask import Flask, request, jsonify, abort
import sqlite3
app = Flask(__name__)
DATABASE = 'loja_acai.db' # Nome do banco de dados SQLite
def get_db():
"""Retorna uma conexão com o banco de dados configurada para retornar dicionários."""
conn = sqlite3.connect(DATABASE)
conn.row_factory = sqlite3.Row
return conn
# ===================== Clientes =====================
@app.route('/clientes', methods=['GET'])
def get_clientes():
conn = get_db()
clientes = conn.execute("SELECT * FROM clientes").fetchall()
conn.close()
return jsonify([dict(row) for row in clientes])
@app.route('/clientes/<int:id>', methods=['GET'])
def get_cliente(id):
conn = get_db()
cliente = conn.execute("SELECT * FROM clientes WHERE id_cliente = ?", (id,)).fetchone()
conn.close()
if cliente is None:
abort(404)
return jsonify(dict(cliente))
@app.route('/clientes', methods=['POST'])
def create_cliente():
if not request.json or 'nome' not in request.json:
abort(400)
cliente = {
'nome': request.json['nome'],
'telefone': request.json.get('telefone', ''),
'email': request.json.get('email', '')
}
conn = get_db()
cur = conn.cursor()
cur.execute("INSERT INTO clientes (nome, telefone, email) VALUES (?, ?, ?)",
(cliente['nome'], cliente['telefone'], cliente['email']))
conn.commit()
cliente['id_cliente'] = cur.lastrowid
conn.close()
return jsonify(cliente), 201
@app.route('/clientes/<int:id>', methods=['PUT'])
def update_cliente(id):
if not request.json:
abort(400)
campos = []
valores = []
for campo in ['nome', 'telefone', 'email']:
if campo in request.json:
campos.append(f"{campo} = ?")
valores.append(request.json[campo])
if not campos:
abort(400)
valores.append(id)
conn = get_db()
conn.execute(f"UPDATE clientes SET {', '.join(campos)} WHERE id_cliente = ?", valores)
conn.commit()
conn.close()
return jsonify({'result': True})
@app.route('/clientes/<int:id>', methods=['DELETE'])
def delete_cliente(id):
conn = get_db()
conn.execute("DELETE FROM clientes WHERE id_cliente = ?", (id,))
conn.commit()
conn.close()
return jsonify({'result': True})
# ===================== Produtos =====================
@app.route('/produtos', methods=['GET'])
def get_produtos():
conn = get_db()
produtos = conn.execute("SELECT * FROM produtos").fetchall()
conn.close()
return jsonify([dict(row) for row in produtos])
@app.route('/produtos/<int:id>', methods=['GET'])
def get_produto(id):
conn = get_db()
produto = conn.execute("SELECT * FROM produtos WHERE id_produto = ?", (id,)).fetchone()
conn.close()
if produto is None:
abort(404)
return jsonify(dict(produto))
@app.route('/produtos', methods=['POST'])
def create_produto():
if not request.json or 'nome' not in request.json or 'preco' not in request.json:
abort(400)
produto = {
'nome': request.json['nome'],
'preco': request.json['preco']
}
conn = get_db()
cur = conn.cursor()
cur.execute("INSERT INTO produtos (nome, preco) VALUES (?, ?)",
(produto['nome'], produto['preco']))
conn.commit()
produto['id_produto'] = cur.lastrowid
conn.close()
return jsonify(produto), 201
@app.route('/produtos/<int:id>', methods=['PUT'])
def update_produto(id):
if not request.json:
abort(400)
campos = []
valores = []
for campo in ['nome', 'preco']:
if campo in request.json:
campos.append(f"{campo} = ?")
valores.append(request.json[campo])
if not campos:
abort(400)
valores.append(id)
conn = get_db()
conn.execute(f"UPDATE produtos SET {', '.join(campos)} WHERE id_produto = ?", valores)
conn.commit()
conn.close()
return jsonify({'result': True})
@app.route('/produtos/<int:id>', methods=['DELETE'])
def delete_produto(id):
conn = get_db()
conn.execute("DELETE FROM produtos WHERE id_produto = ?", (id,))
conn.commit()
conn.close()
return jsonify({'result': True})
# ===================== Materiais =====================
@app.route('/materiais', methods=['GET'])
def get_materiais():
conn = get_db()
materiais = conn.execute("SELECT * FROM materiais").fetchall()
conn.close()
return jsonify([dict(row) for row in materiais])
@app.route('/materiais/<int:id>', methods=['GET'])
def get_material(id):
conn = get_db()
material = conn.execute("SELECT * FROM materiais WHERE id_material = ?", (id,)).fetchone()
conn.close()
if material is None:
abort(404)
return jsonify(dict(material))
@app.route('/materiais', methods=['POST'])
def create_material():
if not request.json or 'nome' not in request.json:
abort(400)
material = {
'nome': request.json['nome'],
'custo': request.json.get('custo', 0.0)
}
conn = get_db()
cur = conn.cursor()
cur.execute("INSERT INTO materiais (nome, custo) VALUES (?, ?)",
(material['nome'], material['custo']))
conn.commit()
material['id_material'] = cur.lastrowid
conn.close()
return jsonify(material), 201
@app.route('/materiais/<int:id>', methods=['PUT'])
def update_material(id):
if not request.json:
abort(400)
campos = []
valores = []
for campo in ['nome', 'custo']:
if campo in request.json:
campos.append(f"{campo} = ?")
valores.append(request.json[campo])
if not campos:
abort(400)
valores.append(id)
conn = get_db()
conn.execute(f"UPDATE materiais SET {', '.join(campos)} WHERE id_material = ?", valores)
conn.commit()
conn.close()
return jsonify({'result': True})
@app.route('/materiais/<int:id>', methods=['DELETE'])
def delete_material(id):
conn = get_db()
conn.execute("DELETE FROM materiais WHERE id_material = ?", (id,))
conn.commit()
conn.close()
return jsonify({'result': True})
# ===================== Vendas =====================
@app.route('/vendas', methods=['GET'])
def get_vendas():
conn = get_db()
vendas = conn.execute("SELECT * FROM vendas").fetchall()
conn.close()
return jsonify([dict(row) for row in vendas])
@app.route('/vendas/<int:id>', methods=['GET'])
def get_venda(id):
conn = get_db()
venda = conn.execute("SELECT * FROM vendas WHERE id_venda = ?", (id,)).fetchone()
conn.close()
if venda is None:
abort(404)
return jsonify(dict(venda))
@app.route('/vendas', methods=['POST'])
def create_venda():
if not request.json or 'id_cliente' not in request.json or 'data_venda' not in request.json or 'total' not in request.json:
abort(400)
venda = {
'id_cliente': request.json['id_cliente'],
'data_venda': request.json['data_venda'],
'total': request.json['total']
}
conn = get_db()
cur = conn.cursor()
cur.execute("INSERT INTO vendas (id_cliente, data_venda, total) VALUES (?, ?, ?)",
(venda['id_cliente'], venda['data_venda'], venda['total']))
conn.commit()
venda['id_venda'] = cur.lastrowid
conn.close()
return jsonify(venda), 201
@app.route('/vendas/<int:id>', methods=['PUT'])
def update_venda(id):
if not request.json:
abort(400)
campos = []
valores = []
for campo in ['id_cliente', 'data_venda', 'total']:
if campo in request.json:
campos.append(f"{campo} = ?")
valores.append(request.json[campo])
if not campos:
abort(400)
valores.append(id)
conn = get_db()
conn.execute(f"UPDATE vendas SET {', '.join(campos)} WHERE id_venda = ?", valores)
conn.commit()
conn.close()
return jsonify({'result': True})
@app.route('/vendas/<int:id>', methods=['DELETE'])
def delete_venda(id):
conn = get_db()
conn.execute("DELETE FROM vendas WHERE id_venda = ?", (id,))
conn.commit()
conn.close()
return jsonify({'result': True})
# ===================== Consumo =====================
@app.route('/consumo', methods=['GET'])
def get_consumo():
conn = get_db()
consumos = conn.execute("SELECT * FROM consumo").fetchall()
conn.close()
return jsonify([dict(row) for row in consumos])
@app.route('/consumo/<int:id>', methods=['GET'])
def get_consumo_item(id):
conn = get_db()
consumo = conn.execute("SELECT * FROM consumo WHERE id_consumo = ?", (id,)).fetchone()
conn.close()
if consumo is None:
abort(404)
return jsonify(dict(consumo))
@app.route('/consumo', methods=['POST'])
def create_consumo():
if not request.json or 'id_venda' not in request.json or 'id_produto' not in request.json or 'quantidade' not in request.json:
abort(400)
consumo = {
'id_venda': request.json['id_venda'],
'id_produto': request.json['id_produto'],
'quantidade': request.json['quantidade']
}
conn = get_db()
cur = conn.cursor()
cur.execute("INSERT INTO consumo (id_venda, id_produto, quantidade) VALUES (?, ?, ?)",
(consumo['id_venda'], consumo['id_produto'], consumo['quantidade']))
conn.commit()
consumo['id_consumo'] = cur.lastrowid
conn.close()
return jsonify(consumo), 201
@app.route('/consumo/<int:id>', methods=['PUT'])
def update_consumo(id):
if not request.json:
abort(400)
campos = []
valores = []
for campo in ['id_venda', 'id_produto', 'quantidade']:
if campo in request.json:
campos.append(f"{campo} = ?")
valores.append(request.json[campo])
if not campos:
abort(400)
valores.append(id)
conn = get_db()
conn.execute(f"UPDATE consumo SET {', '.join(campos)} WHERE id_consumo = ?", valores)
conn.commit()
conn.close()
return jsonify({'result': True})
@app.route('/consumo/<int:id>', methods=['DELETE'])
def delete_consumo(id):
conn = get_db()
conn.execute("DELETE FROM consumo WHERE id_consumo = ?", (id,))
conn.commit()
conn.close()
return jsonify({'result': True})
if __name__ == '__main__':
app.run(debug=True)
Entendendo os Verbos HTTP Utilizados
Nesta API, utilizamos os seguintes métodos HTTP:
- GET: Recupera dados dos registros. Por exemplo,
GET /clientes retorna todos os clientes e GET /clientes/<id> retorna um cliente específico.
- POST: Cria um novo registro. Por exemplo,
POST /produtos espera dados em JSON para criar um novo produto.
- PUT: Atualiza um registro existente. Por exemplo,
PUT /materiais/<id> atualiza as informações do material indicado.
- DELETE: Remove um registro do banco. Exemplo:
DELETE /vendas/<id> remove a venda informada.
Como Executar e Testar a API
Siga os passos abaixo para executar e testar a API:
- Salve o código em um arquivo, por exemplo,
app.py.
- Verifique se o banco de dados
loja_acai.db existe e está configurado com as tabelas apresentadas.
- Instale o Flask, se ainda não o fez:
pip install flask
- Execute o aplicativo:
python app.py
- Utilize ferramentas como Postman ou curl para testar cada endpoint.
Conclusão
Este tutorial demonstrou como criar web services em Python para consumir as tabelas de um banco de dados SQLite da loja de açaí, abrangendo as tabelas Clientes, Produtos, Materiais, Vendas e Consumo. Ao utilizar o Flask e os principais verbos HTTP, você obteve uma API REST simples e modular, facilitando a integração com outras aplicações.
Experimente testar cada endpoint e ajustar o código conforme as necessidades do seu projeto!
English 
Portuguese