Finalizando a web api

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:

1. Salve o código em um arquivo, por exemplo, app.py.
2. Verifique se o banco de dados loja_acai.db existe e está configurado com as tabelas apresentadas.
3. Instale o Flask, se ainda não o fez: pip install flask
4. Execute o aplicativo: python app.py
5. 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!

Exemplo prático de Python para loja Açaí

Tutorial: Web Services em Python para a Tabela Clientes

Neste tutorial, vamos criar uma API REST simples em Python utilizando o Flask para consumir a tabela Clientes de um banco de dados SQLite. Você aprenderá como utilizar os principais verbos HTTP (GET, POST, PUT e DELETE) para realizar operações de leitura, criação, atualização e exclusão de registros.

Requisitos

• Python 3 instalado
• Pacote Flask (instalável via pip install flask)
• Banco de dados SQLite já criado com a tabela clientes

Estrutura da Tabela Clientes

Considere que a tabela clientes foi criada com a seguinte estrutura:

CREATE TABLE clientes (
  id_cliente INTEGER PRIMARY KEY AUTOINCREMENT,
  nome TEXT NOT NULL,
  telefone TEXT,
  email TEXT
);

Código do Web Service

O código a seguir implementa uma API REST utilizando Flask. Ele define uma rota para cada operação (GET, POST, PUT e DELETE) na tabela clientes:

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():
    """Abre uma conexão com o banco de dados e define o row_factory para retornar dicts."""
    conn = sqlite3.connect(DATABASE)
    conn.row_factory = sqlite3.Row
    return conn

# 1. GET /clientes - Retorna todos os 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])

# 2. GET /clientes/<id> - Retorna um cliente específico
@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))

# 3. POST /clientes - Cria um novo 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

# 4. PUT /clientes/<id> - Atualiza um cliente existente
@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})

# 5. DELETE /clientes/<id> - Remove um cliente
@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})

if __name__ == '__main__':
    app.run(debug=True)

Entendendo os Verbos HTTP

Neste exemplo, utilizamos os seguintes métodos HTTP:

• GET: Utilizado para recuperar dados. Temos dois endpoints:
  • GET /clientes – retorna todos os clientes;
  • GET /clientes/<id> – retorna um cliente específico.
• POST: Utilizado para criar um novo recurso. O endpoint POST /clientes recebe dados no formato JSON para inserir um novo cliente.
• PUT: Utilizado para atualizar um recurso existente. O endpoint PUT /clientes/<id> recebe os campos que devem ser atualizados do cliente informado.
• DELETE: Utilizado para remover um recurso. O endpoint DELETE /clientes/<id> exclui o cliente com o id informado.

Como Executar e Testar a API

Siga os passos abaixo para executar e testar a API:

1. Salve o código em um arquivo, por exemplo, app.py.
2. Certifique-se de que o banco de dados loja_acai.db existe e contém a tabela clientes.
3. Instale o Flask se ainda não o fez: pip install flask
4. Execute o aplicativo: python app.py
5. Acesse os endpoints utilizando ferramentas como Postman ou o curl via linha de comando.

Conclusão

Este tutorial apresentou de forma didática como criar web services em Python para a tabela clientes utilizando Flask, explorando os verbos HTTP básicos necessários para a manipulação dos dados (GET, POST, PUT e DELETE). Essa estrutura pode ser facilmente expandida para incluir outras tabelas e funcionalidades em seu projeto.

Experimente testar e modificar os endpoints para adaptar à sua aplicação.