Categorias
API Desenvolvimento Python Tutorial

Construindo uma API em Python

Fui programador Java por 8 anos e este tutorial é para te dar uma dica: Use Python! Veja os motivos do porque eu indico esta linguagem e como fazer uma simples API com acesso ao banco de dados PostgreSQL

Manipulando dados SQL

Bom chegamos até aqui, agora sim nossa API vai começar a ter um corpo mais “profissional”, e também, para quem veio do Java com Springboot e Spring Data ou até mesmo já trabalhou com algum ORM como Hibernate ou EclipseLink verá que agora distoaremos totalmente destes padrões. Focaremos em agilidade de software e otimização de código. No que no Java parece ser feio ou errado, aqui fazemos de forma otimizada e “certa”.

Vamos começar criando nossa database:

create database sample_w_flask;

E então criaremos nossa tabela:

create table if not exists swf_user (
     id VARCHAR(36) primary key,
     name VARCHAR(100),
     email VARCHAR(100),
     active INTEGER default 1
 )

Criando módulo para conexão com PostgreSQL

Optei então por usar o PostgreSQL para nosso banco, é o banco que estou mais familiarizado e também porque eu não gosto muito do MySQL ou MariaDB, mas fiquem à vontade para trocar, talvez não haverá nenhum problema na troca e se tiver será alguma sintaxe diferente nos SQL. Bora lá vamos começar nosso código!

Começaremos então criando uma nova pasta chamada de database e dentro da pasta criaremos um arquivo chamado user_sql.py e então começaremos nossa implementação:

Importaremos psycopg2 que é o módulo responsável pela comunicação com o banco de dados. Importaremos tabém do pacte psycopg2.extras o módulo DictCursor que possibilita a captura a partir do resultado dos selects os atributos como dicionário, por exemplo: result['name']

import psycopg2
from psycopg2.extras import DictCursor

Criaremos um método que cria uma conexão com o Postgres, nele definimos o parametros de acesso ao banco e o nome da nossa database, que neste caso coloquei sample_w_flask


def get_connection():
    return psycopg2.connect(
        host = '127.0.0.1',
        port = '5432',
        user = 'postgres',
        password = 'postgres',
        database = 'sample_w_flask'
    )
Trabalhando com SELECT

Criaremos um método para a listagem de usuários, perceba que abri o método get_connection() dentro de um bloco with pois assim no final da execução o Python se encarrega de fechar a conexão. Inicio o cursor com do DictCursor como factory do cursor. Em seguida, executamos o comando SELECT que queremos obter como resposta, o comando cur.fetchall() traz os resultados do nosso select como lista. Fechamos o cursor e então com o uso de lambda transformamos a respresentação de tupla em dicionário do nosso schema de usuário já definido aqui neste tutorial.

def list_user():
    with get_connection() as conn:
        cur = conn.cursor(cursor_factory = DictCursor)
        cur.execute("SELECT * FROM swf_user")
        rows = cur.fetchall()
        cur.close()
        data = [row_to_dict(x) for x in rows]
        return data
Trabalhando com INSERT

Seguindo praticamente o mesmo conceito, o que mudaremos no método de salvar um novo usuário é substituir a query agora por um INSERT simples, e ao invés de invocar o método fetchall(), invocaremos o método commit() da própria conexão.

def save_user(user):
    with get_connection() as conn:
        cur = conn.cursor()
        cur.execute("""INSERT INTO swf_user 
                        (id, name, email, active) VALUES (%s, %s, %s, %s)""", 
                    (user['id'], user['name'], user['email'], user['active'],))
        conn.commit()
        cur.close()
Trablhando com SELECT com WHERE

E nossa busca por nome? Simples, podemos até reutilizar o mesmo código de listar e apenas alterar a query SQL para adicionarmos uma condição WHERE name ilike 'Obi-Wan', e novamente utilizaremos lambda para transformar a representação de tuplas em dicionário.

def find_by_name(name):
    with get_connection() as conn:
        cur = conn.cursor(cursor_factory = DictCursor)
        name = '%'+name+'%'
        cur.execute("SELECT * FROM swf_user WHERE name ilike %s", (name,))
        rows = cur.fetchall()
        cur.close()
        data = [row_to_dict(x) for x in rows]
        
return data
Trabalhando com DELETE

E para deletar, a mesma coisa, mudaremos nossa query para adequar ao delete.

def delete(uuid):
    with get_connection() as conn:
        cur = conn.cursor()
        print(uuid)
        cur.execute("""DELETE FROM swf_user WHERE id = %s""", (uuid,))
        conn.commit()
        cur.close()

E por final, fiz um método que facilita a transposição e tuplas para dicionário. Quando importamos o DictCursor ele não traz as listas em forma de dict mas sim uma respresentação de tuplas em dict, porém se retornarmos diretamente as tuplas, perderemos o dict delas, por isso que uma redução é necessária, porém ela é simples e sem nenhuma complexidade.

def row_to_dict(row):
    return dict({
        'id' : row['id'],
        'name' : row['name'],
        'email' : row['email'],
        'active' : row['active']
    })

E nosso código deve ficar assim:

import psycopg2
from psycopg2.extras import DictCursor

def get_connection():
    return psycopg2.connect(
        host = '127.0.0.1',
        port = '5432',
        user = 'postgres',
        password = 'postgres',
        database = 'sample_w_flask'
    )

def list_user():
    with get_connection() as conn:
        cur = conn.cursor(cursor_factory = DictCursor)
        cur.execute("SELECT * FROM swf_user")
        rows = cur.fetchall()
        cur.close()
        data = [row_to_dict(x) for x in rows]
        return data

def save_user(user):
    with get_connection() as conn:
        cur = conn.cursor()
        cur.execute("""INSERT INTO swf_user 
                        (id, name, email, active) VALUES (%s, %s, %s, %s)""", 
                    (user['id'], user['name'], user['email'], user['active'],))
        conn.commit()
        cur.close()

def find_by_name(name):
    with get_connection() as conn:
        cur = conn.cursor(cursor_factory = DictCursor)
        name = '%'+name+'%'
        cur.execute("SELECT * FROM swf_user WHERE name ilike %s", (name,))
        rows = cur.fetchall()
        cur.close()
        data = [row_to_dict(x) for x in rows]
        return data

def delete(uuid):
    with get_connection() as conn:
        cur = conn.cursor()
        print(uuid)
        cur.execute("""DELETE FROM swf_user WHERE id = %s""", (uuid,))
        conn.commit()
        cur.close()

def row_to_dict(row):
    return dict({
        'id' : row['id'],
        'name' : row['name'],
        'email' : row['email'],
        'active' : row['active']
    })

Reajustando módulo de Rotas

Bom, agora precisamos alterar nosso módulo de rotas para que chame os métodos que fizemos no módulo de conexão com o banco, para que agora todas as rotas trabalhem com o banco de dados e não em um JSON pré estabelecido. Voltaremos a abrir então o arquivo user_route.py e faremos os seguitnes ajustes:

No cabeçalho de importação, adicionaremos a importação do nosso módulo e chamaremos esse módulo de sql dessa forma:

import database.user_sql as sql

Para o método list_user() atribuiremos à uma variável o retorno do metodo list_user() lá do nosso módulo user_sql.py ficará assim:

@user_route.route('/list/', methods=['GET'])
def list_user():
    user_list = sql.list_user()
    return jsonify(user_list), 200

Para o método create_user() apenas chamaremos o método que salva, ficará desta forma:

@user_route.route('/create/', methods=['PUT'])
def create_user():
    user = request.json
    user['id'] = str(uuid4())
    sql.save_user(user)
    return jsonify(user), 201

Para o método find_user() também não há nenhum segredo, passaremos o nome por parâmetro e retornamos o resultado, desta forma:

@user_route.route('/find/', methods=['GET'])
def find_user():
    name = request.args.get('name')
    founded = sql.find_by_name(name)
    return jsonify(founded), 200

E finalmente para nosso método delete(), apenas substituindo pelo método presente em nosso módulo user_sql.py, ficará assim:

@user_route.route('/<uuid>/', methods=['DELETE'])
def delete_user(uuid):
    sql.delete(uuid)
    return jsonify({'message': 'User removed'}), 200

Testando as alterações

Chamando nossa rota de listagem, agora, recebemos um array vazio, tudo certo, pois ainda não temos nada populado no banco de dados
Salvaremos nosso primeiro usuário, enviamos o JSON como corpo e recebemos o usuário com o id.
Chamando a rota de listagem novamente vemos que agora está vindo o registro que acabamos de salvar.
Agora chamando a rota que filtra pelo nome, vemos que está funcionando e trazendo o dado conforme nos cláusula
E se passarmos um nome que não existe na base, traz um array vazio
Chamando a rota de DELETE passando a UUID do usuário que acabamos de criar
E na nova chamada à nossa rota de Listagem vemos o array vazio novamente pois removemos o único registro que tínhamos

TUDO FUNCIONANDO LINDO E PERFEITO

Porque escolher o Python

Bom, você deve ter percebido o quanto enxuto ficou nossa aplicação, fizemos rotas e comunicação com o banco além do gerenciamento de dependências em apenas 4 arquivos e um sistema de hierarquia de pastas muito simples, para um projeto pequeno ou apenas para aprender é muito fácil fazer o start com o Python, e caso esse teu projeto simples comece a crescer e precisar industrializar, é facilmente refatorado.

Depois de 8 anos programando com Java, eu chego na conclusão que meus próximos projetos com certeza será Python, uma linguagem fácil de aprender, rápido para desenvolver, rápido para iniciar e parar servidor e com uma facilidade muito grande de refatoração.

Gostou do tutorial? Então compartilhe… no WhatsApp, Facebook, Linkedin, TikTok, Email, Orkut… dê-me uma força! Para o próximo quero trazer um tutorial para integração com o banco NoSQL MongoDB.

Este projeto está no GitHub

Até a próxima!

Deixe uma resposta

O seu endereço de e-mail não será publicado. Campos obrigatórios são marcados com *