Desenvolvimento com Flask: Conectando ao Banco de Dados e Usando Blueprints

3. Conexão com o banco de dados MySQL e criação de pool de conexões

A conexão direta com o banco de dados a cada requisição é ineficiente, pois consome tempo e recursos valiosos. Uma solução melhor é utilizar um pool de conexões, que mantém um conjunto de conexões pré-estabeelcidas prontas para uso.

Primeiro, vejamos uma implementação básica sem pool:

def buscar_registro(sql, parametros):
    conexao = pymysql.connect(host='127.0.0.1', port=3306, user='root', passwd='MinhaSenha2024', db='projeto2025')
    cursor = conexao.cursor()
    cursor.execute(sql, parametros)
    resultado = cursor.fetchone()
    cursor.close()
    conexao.close()
    return resultado

resultado = buscar_registro('SELECT * FROM usuarios WHERE token = %s', [token])

Para melhorar o desempenho, implementamos um pool de conexões usando o módulo DButils:

from dbutils.pooled_db import PooledDB

POOL = PooledDB(
    creator=pymysql,  # Módulo para conexão com o banco
    mincached=2,  # Número mínimo de conexões ociosas no pool
    maxcached=5,  # Número máximo de conexões ociosas no pool
    maxconnections=15,  # Número máximo total de conexões
    blocking=True,  # Comportamento quando não há conexões disponíveis
    setsession=[],  # Comandos a serem executados no início da sessão
    ping=0,  # Teste de conexão
    host='127.0.0.1', port=3306, user='root', passwd='MinhaSenha2024', db='projeto2025', charset="utf-8"
)

def buscar_registro(sql, parametros):
    conexao = POOL.connection()  # Obtém uma conexão do pool
    cursor = conexao.cursor()
    cursor.execute(sql, parametros)
    resultado = cursor.fetchone()
    cursor.close()
    conexao.close()  # Retorna a conexão ao pool, não fecha a conexão
    return resultado

4. Instalação e configuração do Redis com pool de conexões

Instalação

No Windows, você pode instalar o Redis diretamente usando o executável *Redis-Windows-x64.msi*. Após a instalação, o serviço será iniciado automaticamente.

Comanods úteis:

  • redis-server --service-install redis.windows-service.conf --loglevel verbose - Instala o serviço
  • redis-server --service-stop - Para o serviço
  • redis-server --service-start - Inicia o serviço
  • redis-server --service-uninstall - Desinstala o serviço

Configuração de senha

Por padrão, o Redis não tem senha. Para definir uma:

  1. Encontre o arquivo redis.conf ou redis.windows.conf
  2. Localize a linha # requirepass foobared
  3. Remova o comentário # e substitua "foobared" pela sua senha desejada

Acessando via cliente

Use o comando redis-cli.exe para acessar o Redis. O comando ping deve retornar pong.

Comandos básicos:

  • dbsize - Mostra o número de chaves no banco atual
  • <índice> - Alterna entre os 16 bancos disponíveis (0-15)
  • flushdb - Apaga todos os dados do banco atual
  • exit ou quit - Sai do cliente

Comandos de manipulação de dados:

  • KEYS * - Lista todas as chaves
  • TYPE <chave> - Mostra o tipo de dado da chave
  • GET <chave> - Obtém o valor de uma chave
  • SET <chave> <valor> - Define um valor para uma chave
  • DEL <chave> - Remove uma chave

Para tipos específicos:

  • Listas: LRANGE <chave> 0 -1
  • Sets: SMEMBERS <chave>
  • Hashes: HGETALL <chave> ou HGET <chave> <campo>

Conexão com Python

Conexão direta
PARAMETROS_CONEXAO = {
    "host": "localhost",
    "port": 6379,
    "encoding": "utf-8"
}
conexao = redis.Redis(**PARAMETROS_CONEXAO)
Conexão via pool
import redis
# Criação do pool de conexões
REDIS_POOL = redis.ConnectionPool(
    host='127.0.0.1', 
    port=6379,
    encoding="utf-8",
    max_connections=1000
)
conexao = redis.Redis(connection_pool=REDIS_POOL)

5. Utilizando o Redis para filas de mensagens e cache

Para operações que consomem muito tempo, podemos utilizar o Redis para criar filas de tarefas, onde as tarefas são colocadas em uma fila e processadas separadamente.

Aplicação Flask

O código abaixo implementa uma API que:

  • Recebe requisições e gera IDs de tarefa
  • Coloca as tarefas em uma fila no Redis
  • Permite verificar o status das tarefas
# Arquivo: app_principal.py
from flask import Flask, request, jsonify
import uuid, redis, json

app = Flask(__name__)

@app.route('/tarefa', methods=['POST'])
def criar_tarefa():
    '''
    Espera dados no formato: {"dados_processar": "...."}
    O token é gerado automaticamente usando uuid
    '''
    dados_entrada = request.json.get('dados_processar')
    if not dados_entrada:
        return jsonify({"status": False, "erro": "Parâmetros inválidos"})

    # 1. Gera ID único para a tarefa
    id_tarefa = str(uuid.uuid4())
    
    # 2. Adiciona à fila de tarefas no Redis
    tarefa = {"id": id_tarefa, "dados": dados_entrada}
    PARAMETROS_CONEXAO = {
        "host": "localhost",
        "port": 6379,
        "encoding": "utf-8"
    }
    conexao = redis.Redis(**PARAMETROS_CONEXAO)
    conexao.lpush("fila_tarefas", json.dumps(tarefa))
    
    # 3. Retorna o ID da tarefa para o cliente
    return jsonify({
        "status": True, 
        "id_tarefa": id_tarefa, 
        "mensagem": "Tarefa recebida, processamento iniciado"
    })

@app.route('/resultado_tarefa', methods=['GET'])
def obter_resultado():
    # Exemplo: /resultado_tarefa?id=....
    id_tarefa = request.args.get('id')
    if not id_tarefa:
        return jsonify({"status": False, "erro": "ID da tarefa não fornecido"})
    
    PARAMETROS_CONEXAO = {
        "host": "localhost",
        "port": 6379,
        "encoding": "utf-8"
    }
    conexao = redis.Redis(**PARAMETROS_CONEXAO)
    resultado = conexao.hget("resultados", id_tarefa)
    
    if not resultado:
        return jsonify({
            "status": True, 
            "resultado": "", 
            "mensagem": "Tarefa ainda em processamento"
        })
    
    # Decodifica o resultado
    resultado_str = resultado.decode("utf-8")
    # Remove o resultado após recuperá-lo
    conexao.hdel("resultados", id_tarefa)

    return jsonify({
        "status": True, 
        "resultado": resultado_str, 
        "mensagem": "Tarefa concluída com sucesso"
    })

if __name__ == '__main__':
    app.run(host="0.0.0.0", port=5000, debug=True)

Worker para processamento das tarefas

O worker abaixo:

  • Busca tarefas da fila
  • Executa o processamento (ex: assinatura digital)
  • Armazena os resultados
# Arquivo: worker.py
import json
import hashlib
import redis

def obter_tarefa():
    # 1. Busca tarefa do Redis
    PARAMETROS_CONEXAO = {
        "host": "localhost",
        "port": 6379,
        "encoding": "utf-8"
    }
    conexao = redis.Redis(**PARAMETROS_CONEXAO)
    dados = conexao.brpop("fila_tarefas", timeout=10)  # Espera por 10 segundos
    
    if not dados:
        return None
    return json.loads(dados[1].decode("utf-8"))

def salvar_resultado(id_tarefa, valor):
    PARAMETROS_CONEXAO = {
        "host": "localhost",
        "port": 6379,
        "encoding": "utf-8"
    }
    conexao = redis.Redis(**PARAMETROS_CONEXAO)
    conexao.hset("resultados", id_tarefa, valor)

def executar_processamento():
    while True:
        # 1. Obtém uma tarefa da fila
        tarefa = obter_tarefa()
        if not tarefa:
            continue

        # 2. Executa o processamento pesado
        dados_entrada = tarefa["dados"]
        dados_processados = dados_entrada + "chave_secreta_de_seguranca"
        resultado = hashlib.md5(dados_processados.encode('utf-8')).hexdigest()

        # 3. Armazena o resultado
        salvar_resultado(tarefa["id"], resultado)

if __name__ == '__main__':
    executar_processamento()

Otimizações possíveis:

  1. Utilizar variáveis globais para os parâmetros de conexão e nomes de filas: ``` FILA_TAREFAS = "fila_tarefas" FILA_RESULTADOS = "resultados"

  2. Implementar um pool de conexões para o Redis: ``` REDIS_POOL = redis.ConnectionPool( host='127.0.0.1', port=6379, encoding="utf-8", max_connections=1000 ) conexao = redis.Redis(connection_pool=REDIS_POOL)

6. Utilização de Blueprints no Flask

Blueprints no Flask são uma forma de modularizar aplicações grandes, permitindo dividir o código em componentes independentes, cada um com suas próprias rotas, templates e arquivos estáticos. Esses componentes são então registrados na aplicação principal.

6.1 Passos básicos de utilização

(1) Criação de um Blueprint

Crie um arquivo de blueprint (ex: autenticacao.py):

from flask import Blueprint

# Cria objeto blueprint: Blueprint('nome', __name__, parâmetros)
bp_autenticacao = Blueprint('autenticacao', __name__, url_prefix='/auth')
(2) Adicionando rotas
@bp_autenticacao.route('/login')
def login():
    return "Página de Login"

@bp_autenticacao.route('/logout')
def logout():
    return "Página de Logout"
(3) Registrando na aplicação principal

No arquivo principal da aplicação (ex: app.py):

from flask import Flask
from autenticacao import bp_autenticacao

app = Flask(__name__)

# Registra o blueprint (com prefixo de URL opcional)
app.register_blueprint(bp_autenticacao)  # Acessar via /auth/login

6.2 Funcionalidades principais

(1) Prefixo de URL

  • Prefixo global: ao registrar o blueprint ``` app.register_blueprint(bp_autenticacao, url_prefix='/usuario')

  • Prefixo interno: ao criar o blueprint ``` bp_autenticacao = Blueprint('autenticacao', name, url_prefix='/auth')
(2) Arquivos estáticos do blueprint
# Especifica pasta de arquivos estáticos do blueprint
bp_admin = Blueprint('admin', __name__, static_folder='estaticos_admin')

# Acesso: /admin/estaticos_admin/estilo.css
@bp_admin.route('/demo-estatico')
def demo_estatico():
    return bp_admin.static_url_path  # Retorna: /admin/estaticos_admin
(3) Diretório de templates do blueprint
# Especifica pasta de templates do blueprint
bp_admin = Blueprint('admin', __name__, template_folder='templates_admin')

@bp_admin.route('/')
def home_admin():
    return render_template('admin.html')  # Busca em templates_admin/admin.html

6.3 Exemplo de estrutura de projeto

meu_app/
├── app.py                 # Aplicação principal
├── autenticacao/          # Módulo de autenticação
│   ├── __init__.py
│   ├── rotas.py          # Definição das rotas
│   └── templates/        # Templates específicos
│       └── auth/
│           └── login.html
├── admin/                # Painel administrativo
│   ├── __init__.py
│   └── rotas.py
└── templates/            # Templates globais
    └── base.html
Exemplo de arquivo de blueprint (autenticacao/rotas.py)
from flask import Blueprint, render_template

bp_autenticacao = Blueprint('autenticacao', __name__, url_prefix='/auth')

@bp_autenticacao.route('/login')
def login():
    return render_template('auth/login.html')  # Caminho: autenticacao/templates/auth/login.html
Registro na aplicação principal (app.py)
from flask import Flask
from autenticacao.rotas import bp_autenticacao
from admin.rotas import bp_admin

app = Flask(__name__)

app.register_blueprint(bp_autenticacao)
app.register_blueprint(bp_admin, url_prefix='/admin')

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

6.4 Técnicas avançadas

(1) Geração de URLs entre blueprints
from flask import url_for

# Formato: nome_blueprint.funcao_visao
url_for('autenticacao.login')  # Gera /auth/login
(2) Hooks de requisição no blueprint
@bp_autenticacao.before_request
def verificar_autenticacao():
    if not usuario_logado():
        return redirect(url_for('autenticacao.login'))
(3) Tratamento de erros no blueprint
@bp_autenticacao.errorhandler(404)
def erro_404_autenticacao(erro):
    return "Página 404 específica do módulo de autenticação", 404

6.5 Vantagens do uso de blueprints

  1. Modularização: Organiza a aplicação por funcionalidades (ex: autenticação, blog, painel administrativo).
  2. Reutilização de código: Permite compartilhar blueprints entre diferentes projetos.
  3. Evita conflitos de rotas: Isola as rotas de cada módulo através de prefixso.
  4. Registro diferido: Permite definir blueprints primeiro e registrá-los posteriormente na aplicação principal.

Com os blueprints, aplicações Flask podem ser facilmente expandidas, mantendo a clareza e manutenibilidade do código.

Tags: Flask MySQL Redis Blueprints Python

Publicado em 8-4 03:56

Tags em Destaque

©2026 Doido Dev