antonino, Autor em Área de Trampo

Classes e Objetos em Python

21/05/202622/04/2026 Por antonino

Classes são plantas arquitetônicas para criar objetos. Objetos são instâncias concretas que ocupam memória. Primeiramente, uma classe define atributos (dados) e métodos (comportamentos). Por exemplo, uma classe Cachorro pode ter nome e latir(). Além disso, cada objeto tem seus próprios valores de atributos. A voz passiva é usada aqui: “os objetos são criados a partir da classe”. Quando utilizar classes e objetos? Em sistemas complexos e reutilizáveis. Também quando você precisa modelar entidades do mundo real. Python é uma linguagem orientada a objetos desde o início. Vamos explorar sintaxe, construtores e encapsulamento. Três subtítulos guiarão você pelos fundamentos da POO. Ao final, você criará suas próprias classes com confiança.

Sintaxe básica: criando classes e instâncias

Use a palavra-chave class seguida do nome da classe. O método __init__ é o construtor especial. Ele inicializa os atributos do novo objeto. O parâmetro self refere-se à própria instância. Quando usar classes? Sempre que você agrupar dados e comportamentos. A voz passiva é aplicada: “os atributos são definidos com self”. Exemplo de classe básica:

# Definindo uma classe simples
class Pessoa:
    """Classe que representa uma pessoa."""
    
    # Construtor (inicializador)
    def __init__(self, nome, idade):
        self.nome = nome  # Atributo público
        self.idade = idade
    
    # Método de instância
    def saudacao(self):
        return f"Olá, meu nome é {self.nome}"
    
    def aniversario(self):
        self.idade += 1
        return f"{self.nome} agora tem {self.idade} anos"
    
    def __str__(self):
        """Representação amigável do objeto."""
        return f"Pessoa(nome={self.nome}, idade={self.idade})"

# Criando objetos (instâncias)
pessoa1 = Pessoa("Ana", 25)
pessoa2 = Pessoa("Carlos", 30)

# Acessando atributos e métodos
print(pessoa1.saudacao())
print(f"Idade da Ana: {pessoa1.idade}")
pessoa1.aniversario()
print(f"Após aniversário: {pessoa1.idade}")

print(pessoa2.saudacao())
print(pessoa2)  # Usa __str__ automaticamente

# Classe com atributo de classe (compartilhado)
class ContaBancaria:
    # Atributo de classe (compartilhado entre todas as instâncias)
    taxa_juros = 0.02
    
    def __init__(self, titular, saldo_inicial=0):
        self.titular = titular
        self.saldo = saldo_inicial
    
    def depositar(self, valor):
        self.saldo += valor
        return self.saldo
    
    def sacar(self, valor):
        if valor <= self.saldo:
            self.saldo -= valor
            return True
        return False
    
    def aplicar_juros(self):
        juros = self.saldo * ContaBancaria.taxa_juros
        self.saldo += juros
        return juros

# Criando contas
conta1 = ContaBancaria("João", 1000)
conta2 = ContaBancaria("Maria", 500)

print(f"\nConta de {conta1.titular}: saldo R${conta1.saldo}")
conta1.depositar(200)
print(f"Após depósito: R${conta1.saldo}")

print(f"Taxa de juros: {ContaBancaria.taxa_juros * 100}%")
juros = conta1.aplicar_juros()
print(f"Juros aplicados: R${juros:.2f}")
print(f"Novo saldo: R${conta1.saldo:.2f}")

# Definindo uma classe simples

class Pessoa:

"""Classe que representa uma pessoa."""

# Construtor (inicializador)

def __init__(self, nome, idade):

self.nome = nome # Atributo público

self.idade = idade

# Método de instância

def saudacao(self):

return f"Olá, meu nome é {self.nome}"

def aniversario(self):

self.idade += 1

return f"{self.nome} agora tem {self.idade} anos"

def __str__(self):

"""Representação amigável do objeto."""

return f"Pessoa(nome={self.nome}, idade={self.idade})"

# Criando objetos (instâncias)

pessoa1 = Pessoa("Ana", 25)

pessoa2 = Pessoa("Carlos", 30)

# Acessando atributos e métodos

print(pessoa1.saudacao())

print(f"Idade da Ana: {pessoa1.idade}")

pessoa1.aniversario()

print(f"Após aniversário: {pessoa1.idade}")

print(pessoa2.saudacao())

print(pessoa2) # Usa __str__ automaticamente

# Classe com atributo de classe (compartilhado)

class ContaBancaria:

# Atributo de classe (compartilhado entre todas as instâncias)

taxa_juros = 0.02

def __init__(self, titular, saldo_inicial=0):

self.titular = titular

self.saldo = saldo_inicial

def depositar(self, valor):

self.saldo += valor

return self.saldo

def sacar(self, valor):

if valor <= self.saldo:

self.saldo -= valor

return True

return False

def aplicar_juros(self):

juros = self.saldo * ContaBancaria.taxa_juros

self.saldo += juros

return juros

# Criando contas

conta1 = ContaBancaria("João", 1000)

conta2 = ContaBancaria("Maria", 500)

print(f"\nConta de {conta1.titular}: saldo R${conta1.saldo}")

conta1.depositar(200)

print(f"Após depósito: R${conta1.saldo}")

print(f"Taxa de juros: {ContaBancaria.taxa_juros * 100}%")

juros = conta1.aplicar_juros()

print(f"Juros aplicados: R${juros:.2f}")

print(f"Novo saldo: R${conta1.saldo:.2f}")

Cada objeto mantém seus próprios valores de atributos. Atributos de classe são compartilhados por todas as instâncias.

Encapsulamento: protegendo dados internos

Encapsulamento esconde detalhes internos da classe. Em Python, usamos convenção de nomes para indicar proteção. Um underscore _atributo significa “protegido”. Dois underscores __atributo causam name mangling. Quando usar encapsulamento? Quando um atributo não deve ser alterado diretamente. Por exemplo, um saldo bancário só muda via métodos. A voz passiva é aplicada: “as propriedades são usadas para controle de acesso”. Exemplo de encapsulamento:

class Produto:
    """Classe com atributos encapsulados."""
    
    def __init__(self, nome, preco, estoque):
        self.nome = nome  # Público
        self._preco = preco  # Protegido (convenção)
        self.__estoque = estoque  # Privado (name mangling)
    
    # Getter para preço (propriedade)
    @property
    def preco(self):
        return self._preco
    
    # Setter com validação
    @preco.setter
    def preco(self, novo_preco):
        if novo_preco < 0:
            raise ValueError("Preço não pode ser negativo")
        self._preco = novo_preco
    
    # Getter para estoque
    @property
    def estoque(self):
        return self.__estoque
    
    # Método para alterar estoque com validação
    def remover_estoque(self, quantidade):
        if quantidade <= 0:
            raise ValueError("Quantidade deve ser positiva")
        if quantidade > self.__estoque:
            raise ValueError("Estoque insuficiente")
        self.__estoque -= quantidade
        return self.__estoque
    
    def adicionar_estoque(self, quantidade):
        if quantidade <= 0:
            raise ValueError("Quantidade deve ser positiva")
        self.__estoque += quantidade
        return self.__estoque
    
    def __str__(self):
        return f"Produto: {self.nome}, Preço: R${self.preco:.2f}, Estoque: {self.estoque}"

# Usando a classe
produto = Produto("Notebook", 2500.00, 10)
print(produto)

# Acessando via propriedade (getter)
print(f"Preço do {produto.nome}: R${produto.preco:.2f}")

# Alterando via setter com validação
produto.preco = 2400.00
print(f"Novo preço: R${produto.preco:.2f}")

# Tentativa de acesso direto ao atributo privado (falha)
try:
    print(produto.__estoque)
except AttributeError as e:
    print(f"Erro: {e} (name mangling)")

# Acesso correto via métodos
produto.remover_estoque(3)
print(f"Estoque após remoção: {produto.estoque}")
produto.adicionar_estoque(5)
print(f"Estoque após adição: {produto.estoque}")

# Demonstração do name mangling
print(f"\nNome real do atributo privado: {dir(produto)[-1]}")

class Produto:

"""Classe com atributos encapsulados."""

def __init__(self, nome, preco, estoque):

self.nome = nome # Público

self._preco = preco # Protegido (convenção)

self.__estoque = estoque # Privado (name mangling)

# Getter para preço (propriedade)

@property

def preco(self):

return self._preco

# Setter com validação

@preco.setter

def preco(self, novo_preco):

if novo_preco < 0:

raise ValueError("Preço não pode ser negativo")

self._preco = novo_preco

# Getter para estoque

@property

def estoque(self):

return self.__estoque

# Método para alterar estoque com validação

def remover_estoque(self, quantidade):

if quantidade <= 0:

raise ValueError("Quantidade deve ser positiva")

if quantidade > self.__estoque:

raise ValueError("Estoque insuficiente")

self.__estoque -= quantidade

return self.__estoque

def adicionar_estoque(self, quantidade):

if quantidade <= 0:

raise ValueError("Quantidade deve ser positiva")

self.__estoque += quantidade

return self.__estoque

def __str__(self):

return f"Produto: {self.nome}, Preço: R${self.preco:.2f}, Estoque: {self.estoque}"

# Usando a classe

produto = Produto("Notebook", 2500.00, 10)

print(produto)

# Acessando via propriedade (getter)

print(f"Preço do {produto.nome}: R${produto.preco:.2f}")

# Alterando via setter com validação

produto.preco = 2400.00

print(f"Novo preço: R${produto.preco:.2f}")

# Tentativa de acesso direto ao atributo privado (falha)

try:

print(produto.__estoque)

except AttributeError as e:

print(f"Erro: {e} (name mangling)")

# Acesso correto via métodos

produto.remover_estoque(3)

print(f"Estoque após remoção: {produto.estoque}")

produto.adicionar_estoque(5)

print(f"Estoque após adição: {produto.estoque}")

# Demonstração do name mangling

print(f"\nNome real do atributo privado: {dir(produto)[-1]}")

Propriedades com @property permitem getters e setters elegantes. Name mangling transforma __estoque em _Produto__estoque.

Métodos especiais e boas práticas

Python oferece métodos especiais (dunder methods) para personalização. __init__ constrói, __str__ exibe, __repr__ depura. __len__ define o tamanho, __add__ sobrecarrega +. Quando usar métodos especiais? Para tornar objetos mais “pythônicos”. Por exemplo, permitir len(obj) ou obj1 + obj2. A voz passiva é aplicada: “as operações são definidas pelos métodos mágicos”. Exemplo de classe com métodos especiais:

class Vetor:
    """Representa um vetor matemático 2D."""
    
    def __init__(self, x, y):
        self.x = x
        self.y = y
    
    def __str__(self):
        """String amigável para usuários."""
        return f"Vetor({self.x}, {self.y})"
    
    def __repr__(self):
        """String para depuração (precisa recriar objeto)."""
        return f"Vetor({self.x}, {self.y})"
    
    def __add__(self, outro):
        """Soma de vetores (sobrecarga do operador +)."""
        if not isinstance(outro, Vetor):
            raise TypeError("Só é possível somar Vetor com Vetor")
        return Vetor(self.x + outro.x, self.y + outro.y)
    
    def __sub__(self, outro):
        """Subtração de vetores (sobrecarga do operador -)."""
        return Vetor(self.x - outro.x, self.y - outro.y)
    
    def __mul__(self, escalar):
        """Multiplicação por escalar."""
        if isinstance(escalar, (int, float)):
            return Vetor(self.x * escalar, self.y * escalar)
        raise TypeError("Multiplicação apenas por número")
    
    def __rmul__(self, escalar):
        """Multiplicação com escalar à esquerda (escalar * vetor)."""
        return self.__mul__(escalar)
    
    def __eq__(self, outro):
        """Comparação de igualdade."""
        if not isinstance(outro, Vetor):
            return False
        return self.x == outro.x and self.y == outro.y
    
    def __len__(self):
        """Retorna o número de componentes (sempre 2)."""
        return 2
    
    def __abs__(self):
        """Módulo (norma) do vetor."""
        return (self.x ** 2 + self.y ** 2) ** 0.5
    
    def __getitem__(self, indice):
        """Permite acesso via índice: v[0] e v[1]."""
        if indice == 0:
            return self.x
        elif indice == 1:
            return self.y
        raise IndexError("Índice inválido")
    
    def __call__(self, fator):
        """Permite chamar o objeto como função: v(2)."""
        return Vetor(self.x * fator, self.y * fator)

# Demonstração
v1 = Vetor(3, 4)
v2 = Vetor(1, 2)

print(f"v1: {v1}")
print(f"v2: {v2}")
print(f"v1 + v2 = {v1 + v2}")
print(f"v1 - v2 = {v1 - v2}")
print(f"v1 * 3 = {v1 * 3}")
print(f"3 * v1 = {3 * v1}")
print(f"v1 == Vetor(3,4)? {v1 == Vetor(3,4)}")
print(f"Tamanho (len): {len(v1)}")
print(f"Módulo de v1: {abs(v1):.2f}")
print(f"v1[0] = {v1[0]}, v1[1] = {v1[1]}")
print(f"v1(2) = {v1(2)}")

# Exemplo de classe com __enter__ e __exit__ (context manager)
class ArquivoGerenciado:
    """Context manager para arquivos."""
    
    def __init__(self, nome, modo):
        self.nome = nome
        self.modo = modo
        self.arquivo = None
    
    def __enter__(self):
        self.arquivo = open(self.nome, self.modo)
        return self.arquivo
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        if self.arquivo:
            self.arquivo.close()
        return False  # Não suprime exceções

# Uso do context manager
with ArquivoGerenciado("teste.txt", "w") as f:
    f.write("Olá, mundo!")

print("\nArquivo escrito com sucesso usando context manager")

100

class Vetor:

"""Representa um vetor matemático 2D."""

def __init__(self, x, y):

self.x = x

self.y = y

def __str__(self):

"""String amigável para usuários."""

return f"Vetor({self.x}, {self.y})"

def __repr__(self):

"""String para depuração (precisa recriar objeto)."""

return f"Vetor({self.x}, {self.y})"

def __add__(self, outro):

"""Soma de vetores (sobrecarga do operador +)."""

if not isinstance(outro, Vetor):

raise TypeError("Só é possível somar Vetor com Vetor")

return Vetor(self.x + outro.x, self.y + outro.y)

def __sub__(self, outro):

"""Subtração de vetores (sobrecarga do operador -)."""

return Vetor(self.x - outro.x, self.y - outro.y)

def __mul__(self, escalar):

"""Multiplicação por escalar."""

if isinstance(escalar, (int, float)):

return Vetor(self.x * escalar, self.y * escalar)

raise TypeError("Multiplicação apenas por número")

def __rmul__(self, escalar):

"""Multiplicação com escalar à esquerda (escalar * vetor)."""

return self.__mul__(escalar)

def __eq__(self, outro):

"""Comparação de igualdade."""

if not isinstance(outro, Vetor):

return False

return self.x == outro.x and self.y == outro.y

def __len__(self):

"""Retorna o número de componentes (sempre 2)."""

return 2

def __abs__(self):

"""Módulo (norma) do vetor."""

return (self.x ** 2 + self.y ** 2) ** 0.5

def __getitem__(self, indice):

"""Permite acesso via índice: v[0] e v[1]."""

if indice == 0:

return self.x

elif indice == 1:

return self.y

raise IndexError("Índice inválido")

def __call__(self, fator):

"""Permite chamar o objeto como função: v(2)."""

return Vetor(self.x * fator, self.y * fator)

# Demonstração

v1 = Vetor(3, 4)

v2 = Vetor(1, 2)

print(f"v1: {v1}")

print(f"v2: {v2}")

print(f"v1 + v2 = {v1 + v2}")

print(f"v1 - v2 = {v1 - v2}")

print(f"v1 * 3 = {v1 * 3}")

print(f"3 * v1 = {3 * v1}")

print(f"v1 == Vetor(3,4)? {v1 == Vetor(3,4)}")

print(f"Tamanho (len): {len(v1)}")

print(f"Módulo de v1: {abs(v1):.2f}")

print(f"v1[0] = {v1[0]}, v1[1] = {v1[1]}")

print(f"v1(2) = {v1(2)}")

# Exemplo de classe com __enter__ e __exit__ (context manager)

class ArquivoGerenciado:

"""Context manager para arquivos."""

def __init__(self, nome, modo):

self.nome = nome

self.modo = modo

self.arquivo = None

def __enter__(self):

self.arquivo = open(self.nome, self.modo)

return self.arquivo

def __exit__(self, exc_type, exc_val, exc_tb):

if self.arquivo:

self.arquivo.close()

return False # Não suprime exceções

# Uso do context manager

with ArquivoGerenciado("teste.txt", "w") as f:

f.write("Olá, mundo!")

print("\nArquivo escrito com sucesso usando context manager")

Métodos especiais tornam suas classes mais integradas à linguagem. A fórmula da orientação a objetos é clara: $POO = \text{classes} + \text{objetos} + \text{encapsulamento} + \text{herança}$ Classes e objetos são o alicerce de qualquer sistema Python. Comece com classes simples e adicione complexidade gradualmente. Seu código será mais organizado, reutilizável e intuitivo.

Celery Filas de Tarefas Assíncronas

22/05/202622/04/2026 Por antonino

Celery é um sistema de filas para processamento assíncrono. Ele executa tarefas em segundo plano fora do ciclo principal. Primeiramente, Celery usa um broker (Redis ou RabbitMQ) para comunicação. Por exemplo, você envia um e-mail sem bloquear a resposta HTTP. Além disso, Celery escala horizontalmente com muitos workers. A voz passiva é usada aqui: “as tarefas são enfileiradas e processadas depois”. Quando utilizar Celery? Em operações demoradas e assíncronas. Por exemplo, processamento de imagens, envio de e-mails ou web scraping. Também em tarefas periódicas (cron jobs) e pipelines de dados. Vamos explorar conceitos, instalação e exemplos práticos. Três subtítulos guiarão você pelo universo Celery. Ao final, você implementará filas robustas em seus projetos.

Conceitos fundamentais: broker, worker e tasks

O broker é o intermediário entre produtores e consumidores. Redis e RabbitMQ são os brokers mais comuns para Celery. Workers são processos que executam as tarefas da fila. Tasks são funções decoradas com @app.task. Quando usar cada componente? Broker para comunicação entre serviços. Worker para processamento paralelo de tarefas. A voz passiva é aplicada: “as tarefas são serializadas em JSON”. Exemplo básico de configuração e tarefa Celery:

# tasks.py
from celery import Celery
import time

# Configuração do Celery com Redis como broker
app = Celery(
    'tasks',
    broker='redis://localhost:6379/0',
    backend='redis://localhost:6379/0'
)

@app.task
def tarefa_demorada(segundos):
    """Tarefa que simula processamento demorado."""
    print(f"Iniciando tarefa por {segundos}s")
    for i in range(segundos):
        time.sleep(1)
        print(f"Progresso: {i+1}/{segundos}")
    return f"Tarefa concluída após {segundos}s"

@app.task
def enviar_email(destinatario, assunto, corpo):
    """Simula envio de e-mail."""
    print(f"Enviando e-mail para {destinatario}")
    time.sleep(2)  # Simula envio
    return f"E-mail enviado para {destinatario}"

@app.task
def processar_imagem(caminho, operacao):
    """Simula processamento de imagem."""
    print(f"Processando {caminho} com operação '{operacao}'")
    time.sleep(3)
    return f"Imagem {caminho} processada"

# Tarefa com agendamento
@app.task
def tarefa_periodica():
    """Tarefa executada periodicamente."""
    print("Executando tarefa periódica...")
    return "Tarefa periódica concluída"

# Como executar:
# Terminal 1: celery -A tasks worker --loglevel=info
# Terminal 2: python -c "from tasks import tarefa_demorada; tarefa_demorada.delay(5)"

# tasks.py

from celery import Celery

import time

# Configuração do Celery com Redis como broker

app = Celery(

'tasks',

broker='redis://localhost:6379/0',

backend='redis://localhost:6379/0'

)

@app.task

def tarefa_demorada(segundos):

"""Tarefa que simula processamento demorado."""

print(f"Iniciando tarefa por {segundos}s")

for i in range(segundos):

time.sleep(1)

print(f"Progresso: {i+1}/{segundos}")

return f"Tarefa concluída após {segundos}s"

@app.task

def enviar_email(destinatario, assunto, corpo):

"""Simula envio de e-mail."""

print(f"Enviando e-mail para {destinatario}")

time.sleep(2) # Simula envio

return f"E-mail enviado para {destinatario}"

@app.task

def processar_imagem(caminho, operacao):

"""Simula processamento de imagem."""

print(f"Processando {caminho} com operação '{operacao}'")

time.sleep(3)

return f"Imagem {caminho} processada"

# Tarefa com agendamento

@app.task

def tarefa_periodica():

"""Tarefa executada periodicamente."""

print("Executando tarefa periódica...")

return "Tarefa periódica concluída"

# Como executar:

# Terminal 1: celery -A tasks worker --loglevel=info

# Terminal 2: python -c "from tasks import tarefa_demorada; tarefa_demorada.delay(5)"

O worker precisa estar rodando para executar as tarefas. O comando celery -A tasks worker inicia o worker.

Enfileirando e monitorando tarefas

Para enfileirar uma tarefa, use o método .delay(). Ele retorna um objeto AsyncResult para acompanhamento. Use .get() para aguardar o resultado (bloqueante). Use .ready() para verificar se a tarefa terminou. Quando usar essas funções? Em aplicações web assíncronas. Também em scripts que precisam monitorar progresso. A voz passiva é aplicada: “os resultados são armazenados no backend”. Exemplo de enfileiramento e monitoramento:

# enfileirar.py
from tasks import tarefa_demorada, enviar_email
from celery.result import AsyncResult
import time

# Enfileirando tarefas (não bloqueia)
print("=== Enfileirando tarefas ===")
resultado1 = tarefa_demorada.delay(3)
resultado2 = enviar_email.delay("user@exemplo.com", "Olá", "Conteúdo do e-mail")
resultado3 = tarefa_demorada.delay(2)

print(f"ID da tarefa 1: {resultado1.id}")
print(f"ID da tarefa 2: {resultado2.id}")
print(f"ID da tarefa 3: {resultado3.id}")

# Monitorando tarefas
print("\n=== Monitorando progresso ===")
tarefas = [resultado1, resultado2, resultado3]

while any(not t.ready() for t in tarefas):
    for t in tarefas:
        status = "pronta" if t.ready() else "pendente"
        print(f"Tarefa {t.id[:8]}...: {status}")
    time.sleep(1)

print("\n=== Resultados ===")
for t in tarefas:
    print(f"Tarefa {t.id[:8]}...: {t.result}")

# Exemplo com Chain (encadeamento de tarefas)
print("\n=== Encadeamento de tarefas (Chain) ===")
from celery import chain

@celery_app.task
def dobrar(x):
    return x * 2

@celery_app.task
def adicionar_um(x):
    return x + 1

# Executa: dobrar -> adicionar_um
cadeia = chain(dobrar.s(5), adicionar_um.s())
resultado = cadeia()
print(f"Resultado do chain: {resultado.get()}")  # (5*2)+1 = 11

# Exemplo com Group (tarefas paralelas)
print("\n=== Grupo de tarefas paralelas (Group) ===")
from celery import group

tarefas_paralelas = group(
    tarefa_demorada.s(2),
    tarefa_demorada.s(3),
    tarefa_demorada.s(1)
)
resultado_grupo = tarefas_paralelas.apply_async()
resultados = resultado_grupo.get()
print(f"Resultados do grupo: {resultados}")

# enfileirar.py

from tasks import tarefa_demorada, enviar_email

from celery.result import AsyncResult

import time

# Enfileirando tarefas (não bloqueia)

print("=== Enfileirando tarefas ===")

resultado1 = tarefa_demorada.delay(3)

resultado2 = enviar_email.delay("[email protected]", "Olá", "Conteúdo do e-mail")

resultado3 = tarefa_demorada.delay(2)

print(f"ID da tarefa 1: {resultado1.id}")

print(f"ID da tarefa 2: {resultado2.id}")

print(f"ID da tarefa 3: {resultado3.id}")

# Monitorando tarefas

print("\n=== Monitorando progresso ===")

tarefas = [resultado1, resultado2, resultado3]

while any(not t.ready() for t in tarefas):

for t in tarefas:

status = "pronta" if t.ready() else "pendente"

print(f"Tarefa {t.id[:8]}...: {status}")

time.sleep(1)

print("\n=== Resultados ===")

for t in tarefas:

print(f"Tarefa {t.id[:8]}...: {t.result}")

# Exemplo com Chain (encadeamento de tarefas)

print("\n=== Encadeamento de tarefas (Chain) ===")

from celery import chain

@celery_app.task

def dobrar(x):

return x * 2

@celery_app.task

def adicionar_um(x):

return x + 1

# Executa: dobrar -> adicionar_um

cadeia = chain(dobrar.s(5), adicionar_um.s())

resultado = cadeia()

print(f"Resultado do chain: {resultado.get()}") # (5*2)+1 = 11

# Exemplo com Group (tarefas paralelas)

print("\n=== Grupo de tarefas paralelas (Group) ===")

from celery import group

tarefas_paralelas = group(

tarefa_demorada.s(2),

tarefa_demorada.s(3),

tarefa_demorada.s(1)

)

resultado_grupo = tarefas_paralelas.apply_async()

resultados = resultado_grupo.get()

print(f"Resultados do grupo: {resultados}")

O método .s() cria uma assinatura de tarefa. Assinaturas permitem composição e encadeamento.

Celery para tarefas periódicas (celery beat)

Celery Beat é o scheduler para tarefas periódicas. Ele enfileira tarefas em intervalos regulares automaticamente. Quando usar Beat? Em backups, limpeza de dados ou relatórios. Por exemplo, enviar relatório todo dia às 8h. A voz passiva é aplicada: “os intervalos são definidos no arquivo de configuração”. Exemplo de configuração de tarefas periódicas:

# celery_config.py
from celery.schedules import crontab

# Configuração do Beat
beat_schedule = {
    # A cada 30 segundos
    'a-cada-30-segundos': {
        'task': 'tasks.tarefa_periodica',
        'schedule': 30.0,  # segundos
        'args': (),
    },
    # Todo dia à meia-noite
    'limpeza-diaria': {
        'task': 'tasks.limpar_cache',
        'schedule': crontab(hour=0, minute=0),
        'args': (),
    },
    # Toda segunda-feira às 10h
    'relatorio-semanal': {
        'task': 'tasks.gerar_relatorio',
        'schedule': crontab(day_of_week=1, hour=10, minute=0),
        'args': ('semanal',),
    },
    # A cada 5 minutos
    'monitoramento': {
        'task': 'tasks.verificar_health',
        'schedule': 300.0,
        'args': (),
    },
}

# Aplicando a configuração
app.conf.beat_schedule = beat_schedule
app.conf.timezone = 'America/Sao_Paulo'

# tasks.py com tarefas periódicas
@app.task
def limpar_cache():
    """Limpa cache do sistema."""
    print("Executando limpeza de cache...")
    # lógica de limpeza
    return "Cache limpo"

@app.task
def gerar_relatorio(tipo):
    """Gera relatório periódico."""
    print(f"Gerando relatório {tipo}...")
    return f"Relatório {tipo} gerado"

@app.task
def verificar_health():
    """Verifica saúde do sistema."""
    print("Verificando health check...")
    return "Sistema saudável"

# Exemplo de tarefa com retry automático
@app.task(bind=True, max_retries=3, default_retry_delay=60)
def tarefa_com_retry(self, url):
    """Tarefa que tenta novamente em caso de falha."""
    import requests
    try:
        response = requests.get(url, timeout=10)
        return response.status_code
    except Exception as e:
        print(f"Falha ao acessar {url}: {e}")
        # Tenta novamente após 60 segundos
        raise self.retry(exc=e)

# Exemplo de tarefa com timeout
@app.task(time_limit=30, soft_time_limit=25)
def tarefa_limitada():
    """Tarefa que para após 30 segundos."""
    import time
    time.sleep(40)  # Excede o timeout
    return "Nunca chegará aqui"

# celery_config.py

from celery.schedules import crontab

# Configuração do Beat

beat_schedule = {

# A cada 30 segundos

'a-cada-30-segundos': {

'task': 'tasks.tarefa_periodica',

'schedule': 30.0, # segundos

'args': (),

# Todo dia à meia-noite

'limpeza-diaria': {

'task': 'tasks.limpar_cache',

'schedule': crontab(hour=0, minute=0),

'args': (),

# Toda segunda-feira às 10h

'relatorio-semanal': {

'task': 'tasks.gerar_relatorio',

'schedule': crontab(day_of_week=1, hour=10, minute=0),

'args': ('semanal',),

# A cada 5 minutos

'monitoramento': {

'task': 'tasks.verificar_health',

'schedule': 300.0,

'args': (),

}

# Aplicando a configuração

app.conf.beat_schedule = beat_schedule

app.conf.timezone = 'America/Sao_Paulo'

# tasks.py com tarefas periódicas

@app.task

def limpar_cache():

"""Limpa cache do sistema."""

print("Executando limpeza de cache...")

# lógica de limpeza

return "Cache limpo"

@app.task

def gerar_relatorio(tipo):

"""Gera relatório periódico."""

print(f"Gerando relatório {tipo}...")

return f"Relatório {tipo} gerado"

@app.task

def verificar_health():

"""Verifica saúde do sistema."""

print("Verificando health check...")

return "Sistema saudável"

# Exemplo de tarefa com retry automático

@app.task(bind=True, max_retries=3, default_retry_delay=60)

def tarefa_com_retry(self, url):

"""Tarefa que tenta novamente em caso de falha."""

import requests

try:

response = requests.get(url, timeout=10)

return response.status_code

except Exception as e:

print(f"Falha ao acessar {url}: {e}")

# Tenta novamente após 60 segundos

raise self.retry(exc=e)

# Exemplo de tarefa com timeout

@app.task(time_limit=30, soft_time_limit=25)

def tarefa_limitada():

"""Tarefa que para após 30 segundos."""

import time

time.sleep(40) # Excede o timeout

return "Nunca chegará aqui"

Para rodar o Beat: celery -A tasks beat --loglevel=info. Para rodar worker: celery -A tasks worker --loglevel=info. A fórmula de throughput da fila: $T = \frac{N_{\text{tarefas}}}{N_{\text{workers}} \times \text{velocidade}}$ Celery é essencial para aplicações web robustas. Comece com tarefas simples e evolua para workflows complexos. Sua aplicação ficará mais rápida e responsiva.