Implementação de Blueprints e Pools de Conexão em Aplicações Flask

Organização com Blueprints

Em projetos Flask com múltiplas funções de visão (view functions), os blueprints perimtem modularizar o código em arquivos separados, promovendo uma estrutura de diretórios escalável e organizada.

Estrutura de Diretórios Recomendada

meu_projeto_flask/
├── app.py                  # Arquivo principal da aplicação
├── README.md
├── meu_modulo/             # Módulo contendo blueprints, templates e estáticos
│   ├── __init__.py         # Inicialização do módulo
│   ├── rotas/              # Diretório para arquivos de rotas usando blueprints
│   │   ├── perfil.py
│   │   └── compras.py
│   ├── templates/          # Templates específicos do módulo
│   │   └── autenticacao/
│   │       └── login.html
│   └── static/             # Arquivos estáticos
│       └── estilos/
│           ├── css/
│           └── js/
└── utilitarios/            # Ferramentas auxiliares
    └── bd.py               # Configuração de conexão com banco de dados

Definição e Registro de Blueprints

Cada arquivo de rotas define um blueprint, que é registrado na aplicação principal para integrar as rotas.

# rotas/perfil.py
from flask import Blueprint

bp_perfil = Blueprint('perfil', __name__)

@bp_perfil.route('/dashboard')
def painel_controle():
    return 'Painel de controle do usuário'

@bp_perfil.route('/configuracoes')
def ajustes():
    return 'Página de configurações'

# __init__.py do módulo meu_modulo
from flask import Flask

def criar_aplicacao():
    app = Flask(__name__)
    app.config['SECRET_KEY'] = 'chave_secreta_complexa_123'

    from .rotas import perfil, compras
    app.register_blueprint(perfil.bp_perfil)
    # Para adicionar prefixo de URL, use: app.register_blueprint(perfil.bp_perfil, url_prefix='/usuario')
    app.register_blueprint(compras.bp_compras)

    return app

# app.py
from meu_modulo import criar_aplicacao

app = criar_aplicacao()

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

Conceitos Fundamentais

  • O Flask gernecia requisições e sessões através de contextos, diferenciando-se de frameworks como o Django.
  • A sessão é armazenada em cookies criptografados no navegador, dependendo de uma chave secreta (SECRET_KEY).
  • Decoradores devem ser implmeentados com functools para preservar metadados das funções.
  • Manter uma estrutura de diretórios consistente facilita a manutenção e expansão do projeto.

Pools de Conexão com Banco de Dados

Para otimizar o acesso a bancos de dados MySQL, utilize pools de conexão com bibliotecas como DBUtils e PyMySQL, reduzindo sobrecarga de estabelecer conexões repetidamente.

Instalação e Configuração

pip install dbutils pymysql

import pymysql
from dbutils.pooled_db import PooledDB
from pymysql.cursors import DictCursor

# Configuração do pool de conexões
pool_bd = PooledDB(
    creator=pymysql,
    maxconnections=6,       # Máximo de conexões simultâneas (0 para ilimitado)
    mincached=2,            # Conexões mínimas iniciais
    blocking=True,          # Esperar por conexão disponível se o pool estiver cheio
    ping=0,                 # Verificar disponibilidade do servidor (0: sem verificação)
    host='localhost',
    port=3306,
    user='admin',
    password='senha_segura_2024',
    database='meu_banco',
    charset='utf8mb4'
)

# Utilizando uma conexão do pool
conexao = pool_bd.connection()
cursor = conexao.cursor(cursor=DictCursor)
cursor.execute("SELECT * FROM usuarios WHERE id=%s AND nome=%s", (1, 'Alice'))
registros = cursor.fetchall()
cursor.close()
conexao.close()  # Retorna a conexão ao pool em vez de fechá-la

Teste com Múltiplas Threads

import pymysql
from dbutils.pooled_db import PooledDB
from pymysql.cursors import DictCursor
from threading import Thread

pool_bd = PooledDB(
    creator=pymysql,
    maxconnections=10,
    mincached=3,
    blocking=True,
    ping=0,
    host='localhost',
    port=3306,
    user='admin',
    password='senha_segura_2024',
    database='meu_banco',
    charset='utf8mb4'
)

def executar_consulta(identificador):
    conexao = pool_bd.connection()
    cursor = conexao.cursor(cursor=DictCursor)
    cursor.execute("SELECT * FROM produtos WHERE id=%s", (identificador,))
    dados = cursor.fetchall()
    cursor.close()
    conexao.close()
    print(f"Thread {identificador}: {dados}")

# Criando múltiplas threads para consultas simultâneas
for contador in range(20):
    t = Thread(target=executar_consulta, args=(contador,))
    t.start()

Implementação Baseada em Classes

Para reutilização e encapsulamento, crie uma classe que gerencia o pool e fornece métodos para consultas.

import pymysql
from dbutils.pooled_db import PooledDB
from pymysql.cursors import DictCursor

class GerenciadorBD:
    def __init__(self):
        self.pool = PooledDB(
            creator=pymysql,
            maxconnections=8,
            mincached=2,
            blocking=True,
            ping=0,
            host='localhost',
            port=3306,
            user='admin',
            password='senha_segura_2024',
            database='meu_banco',
            charset='utf8mb4'
        )

    def obter_conexao(self):
        conexao = self.pool.connection()
        cursor = conexao.cursor(cursor=DictCursor)
        return conexao, cursor

    def liberar_recursos(self, cursor, conexao):
        cursor.close()
        conexao.close()

    def buscar_todos(self, consulta, parametros=None):
        conexao, cursor = self.obter_conexao()
        cursor.execute(consulta, parametros)
        resultados = cursor.fetchall()
        self.liberar_recursos(cursor, conexao)
        return resultados

    def buscar_um(self, consulta, parametros=None):
        conexao, cursor = self.obter_conexao()
        cursor.execute(consulta, parametros)
        resultado = cursor.fetchone()
        self.liberar_recursos(cursor, conexao)
        return resultado

# Uso da classe
gerenciador = GerenciadorBD()
dados = gerenciador.buscar_todos("SELECT * FROM pedidos WHERE status=%s", ('pendente',))
print(dados)

Tags: Flask Blueprints DBUtils PyMySQL Connection Pooling

Publicado em 7-14 13:35