Implementação de um Sistema de Autenticação com Tornado e SQLAlchemy

O framework Tornado é conhecido por sua eficiência em lidar com conexões assíncronas, sendo uma excelente escolha para aplicações Web escaláveis em Python. Abaixo, detalhamos como estruturar um sistema completo de gerenciamento de usuários, abrangendo desde a configuração do servidor até a persistência de dados com SQLAlchemy e gerenciamento de sessões com Redis.

1. Configuração Principal do Aplicativo (main.py)

O ponto de entrada da aplicação define as rotas, as configurações de segurança (XSRF e Cookies) e a integração com o Redis para armazenamento de sessões via pycket.

import tornado.ioloop
import tornado.web
import tornado.options
from tornado.options import define, options

from handlers.core import HomeHandler
from handlers.auth import SignupHandler, AuthHandler

define('port', default=8889, help='Porta de execução', type=int)

class TornadoWebApp(tornado.web.Application):
    def __init__(self):
        # Definição de rotas do sistema
        routes = [
            (r'/', HomeHandler),
            (r'/registro', SignupHandler),
            (r'/login', AuthHandler),
        ]

        # Configurações de ambiente e middlewares
        config = dict(
            debug=True,
            static_path='static',
            template_path='templates',
            cookie_secret='CHAVE_SECRETA_ALEATORIA',
            xsrf_cookies=True,
            pycket={
                'engine': 'redis',
                'storage': {
                    'host': '127.0.0.1',
                    'port': 6379,
                    'db_sessions': 10,
                    'max_connections': 2 ** 31,
                },
                'cookies': {
                    'expires_days': 2,
                },
            },
            login_url='/login',
        )
        super().__init__(routes, **config)

if __name__ == "__main__":
    tornado.options.parse_command_line()
    app = TornadoWebApp()
    app.listen(options.port)
    print(f"Servidor rodando na porta: {options.port}")
    tornado.ioloop.IOLoop.current().start()

2. Handlers de Processamento

Os handlers gerenciam o fluxo de requisições. Utilziamos uma classe base para integrar o mixin de sessão e validar o estado de login do usuário.

Base e Página Inicial (hendlers/core.py)

import tornado.web
from pycket.session import SessionMixin

class BaseHandler(tornado.web.RequestHandler, SessionMixin):
    def get_current_user(self):
        # Retorna o identificador do usuário salvo na sessão
        return self.session.get('user_id')

class HomeHandler(BaseHandler):
    def get(self):
        self.write('Bem-vindo à página inicial!')

Autenticação (handlers/auth.py)

import tornado.web
import re
from models.account import Usuario
from utils.security import gerar_hash_senha
from handlers.core import BaseHandler

class SignupHandler(tornado.web.RequestHandler):
    def get(self):
        self.render('register.html')

    def post(self):
        user_name = self.get_argument('username', '').strip()
        pwd = self.get_argument('password', '').strip()
        pwd_confirm = self.get_argument('confirm_password', '').strip()

        if not all([user_name, pwd, pwd_confirm]):
            return self.write('Erro: Preencha todos os campos.')

        if pwd != pwd_confirm:
            return self.write('Erro: As senhas não coincidem.')

        if Usuario.buscar_por_nome(user_name):
            return self.write('Erro: Usuário já cadastrado.')

        # Persistência do novo usuário
        hashed_pwd = gerar_hash_senha(pwd)
        Usuario.salvar_novo_usuario(user_name, hashed_pwd)
        
        self.redirect('/login')

class AuthHandler(BaseHandler):
    def get(self):
        self.render('login.html')

    def post(self):
        user_name = self.get_argument('username', '').strip()
        pwd = self.get_argument('password', '').strip()

        usuario_instancia = Usuario.buscar_por_nome(user_name)
        if usuario_instancia and usuario_instancia.validar_acesso(pwd):
            self.session.set('user_id', user_name)
            redirecionar = self.get_argument('next', '/')
            self.redirect(redirecionar)
        else:
            self.write('Credenciais inválidas.')

3. Camada de Modelagem de Dados (SQLAlchemy)

Utilizamos o SQLAlchemy para mapear as tabelas do banco de dados MySQL para classes Python.

Definição de Modelos (models/acccount.py)

from datetime import datetime
from sqlalchemy import Column, Integer, String, DateTime, Boolean, ForeignKey
from sqlalchemy.orm import relationship
from models.database import Base, db_session

class TimestampMixin:
    data_criacao = Column(DateTime, default=datetime.now)
    ativo = Column(Boolean, default=True)

class Usuario(Base, TimestampMixin):
    __tablename__ = 'usuarios'
    
    id = Column(Integer, primary_key=True, autoincrement=True)
    username = Column(String(50), nullable=False, unique=True)
    password = Column(String(255), nullable=False)
    email = Column(String(120))

    @classmethod
    def salvar_novo_usuario(cls, name, secret):
        novo_user = cls(username=name, password=secret)
        db_session.add(novo_user)
        db_session.commit()

    @classmethod
    def buscar_por_nome(cls, name):
        return db_session.query(cls).filter_by(username=name).first()

    def validar_acesso(self, tentativa_senha):
        # Lógica de comparação de hash (exemplo simplificado)
        return self.password == tentativa_senha

class Galeria(Base, TimestampMixin):
    __tablename__ = 'galerias'
    
    id = Column(Integer, primary_key=True)
    url_midia = Column(String(500))
    dono_id = Column(Integer, ForeignKey('usuarios.id'))
    dono = relationship('Usuario', backref='midias')

Conexão com Banco de Dados (models/database.py)

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

DB_URI = 'mysql+pymysql://root:senha@127.0.0.1:3306/meu_projeto?charset=utf8'

engine = create_engine(DB_URI, echo=False)
Base = declarative_base(bind=engine)
SessionLocal = sessionmaker(bind=engine)
db_session = SessionLocal()

4. Templates HTML

O sistema de templates do Tornado permite herança de blocos, facilitando a reutilização de código front-end.

Estrutura Base (templates/base.html)


<html lang="pt-br">
<head>
    <meta charset="UTF-8">
    <title>{% block title %}Sistema Tornado{% end %}</title>
</head>
<body>
    <main>
        {% block content %}{% end %}
    </main>
</body>
</html>

Página de Registro (templates/register.html)

{% extends 'base.html' %}

{% block title %}Cadastro{% end %}

{% block content %}
    <h2>Criar Conta</h2>
    <form method="post">
        {% module xsrf_form_html() %}
        <input type="text" name="username" placeholder="Usuário" required><br>
        <input type="password" name="password" placeholder="Senha" required><br>
        <input type="password" name="confirm_password" placeholder="Confirme a Senha" required><br>
        <button type="submit">Registrar</button>
    </form>
{% end %}

Tags: Tornado SQLAlchemy Python Redis Web Development

Publicado em 7-15 09:08