duppla_slack 2026.7.6rc1

Tailored Slack Bolt wrapper for duppla's internal services

Slack [Aplicacion interna de duppla]

Este paquete de Python envuelve slack_bolt (y, por debajo, el slack_sdk) con una interfaz tipada y adaptada al uso interno de duppla. Permite enviar mensajes, recibir eventos con Bolt, construir vistas de Block Kit y conectarse a integraciones opcionales (Redis, RabbitMQ, FastAPI) a traves de los diversos repositorios de la organizacion.

[!IMPORTANT] Este paquete no está disponible para su uso externo.

[!NOTE] Para mas información sobre la API de Slack, visite Slack API, la librería slack_bolt y el slack_sdk.

Instalación

uv add duppla_slack
# o
pip install duppla_slack

Extras opcionales segun la integracion que necesites:

uv add "duppla_slack[fastapi]"    # router de FastAPI para recibir eventos
uv add "duppla_slack[redis]"      # rate limiter distribuido + cache
uv add "duppla_slack[rabbitmq]"   # publisher/consumer de eventos
uv add "duppla_slack[all]"        # todo lo anterior

Uso básico

El punto de entrada es SlackService. Puedes construirlo con el token directamente o con un SlackConfig.

from duppla_slack import SlackService, SlackConfig, Message, AIRFLOW_CHANNEL

# Opcion 1: token directo
slack = SlackService(token="xoxb-…", default_channel=AIRFLOW_CHANNEL)

# Opcion 2: configuracion explicita
slack = SlackService(config=SlackConfig(token="xoxb-…", signing_secret="…"))

# Opcion 3: desde variables de entorno
# (SLACK_BOT_TOKEN, SLACK_SIGNING_SECRET, SLACK_APP_TOKEN,
#  SLACK_DEFAULT_CHANNEL, SLACK_REDIS_URL)
slack = SlackService(config=SlackConfig.from_env())

# Envio rapido de texto
slack.send_msg_(text="Hola equipo", channel="general")

# Envio con modelo tipado
slack.send_msg(Message(text="Despliegue completado", channel="C0123456789"))

# Otras operaciones comunes
slack.update_message(channel="C0123456789", timestamp="1700000000.000100", text="Editado")
slack.send_and_pin_msg(Message(text="Anclado", channel="C0123456789"))
slack.upload_file(filename="reporte.csv", content=b"a,b,c", channel="C0123456789")

channel_id = slack.get_channel_id("general")
user_id = slack.get_user_id_email("[e-mail]")
members = slack.get_members()

Cualquier método del slack_sdk.WebClient se reenvía de forma transparente:

info = slack.conversations_info(channel="C0123456789")

Async

Cada helper tiene su variante asíncrona con prefijo a. La app de Bolt subyacente es asíncrona.

import asyncio
from duppla_slack import SlackService, Message

async def main():
    slack = SlackService(token="xoxb-…")
    await slack.asend_msg(Message(text="Hola async", channel="C0123456789"))
    await slack.asend_msg_(text="Hola", channel="general")
    members = await slack.aget_members()

asyncio.run(main())

Recibir eventos con Bolt + FastAPI

Para recibir comandos, acciones, vistas y eventos necesitas un signing_secret. El servicio expone los decoradores de Bolt directamente y un router listo para montar en FastAPI (requiere duppla_slack[fastapi]).

from fastapi import FastAPI
from duppla_slack import SlackService, SlackConfig

slack = SlackService(config=SlackConfig.from_env())  # necesita SLACK_SIGNING_SECRET

@slack.command("/ping")
async def ping(ack, respond):
    await ack()
    await respond("pong")

@slack.action("btn")
async def on_click(ack, body):
    await ack()

@slack.event("app_mention")
async def on_mention(event, say):
    await say("Aqui estoy")

@slack.error
async def on_error(error, body):
    ...

app = FastAPI()
app.include_router(slack.fastapi_router())  # POST /slack/events

Otros decoradores disponibles: @slack.view("cb"), @slack.shortcut(...), @slack.message(...), @slack.middleware.

Parseo tipado de interacciones

El módulo duppla_slack.interactions (re-exportado en el nivel superior) evita el acceso a diccionarios anidados (body["view"]["state"]["values"][block][action][...]).

parse_view_state(view) devuelve un ViewState, un Mapping plano {action_id: valor} que auto-detecta el tipo de cada elemento de Block Kit (selects, pickers, checkboxes, file inputs, …).

from duppla_slack import SlackService, SlackConfig, parse_view_state
from duppla_slack.interactions import ViewSubmission, BlockAction

slack = SlackService(config=SlackConfig.from_env())

@slack.view("reporte")
async def on_submit(ack, body):
    await ack()
    sub = ViewSubmission.model_validate(body)
    sub.callback_id        # "reporte"
    sub.private_metadata   # metadata del modal
    detalle = sub.state.value("detalle")   # ViewState.value(action_id, default)
    todos = sub.values                     # dict {action_id: valor} (== sub.state.as_dict())
    # equivalente sin envoltura: parse_view_state(body["view"]).as_dict()

@slack.action("confirm")
async def on_click(ack, body):
    await ack()
    action = BlockAction.model_validate(body)
    action.user_id         # quien pulsó
    action.channel_id
    action.message_ts
    action.action_id       # "confirm"
    action.value           # value del botón o de la opción seleccionada
    action.metadata        # metadata.event_payload del mensaje origen
    action.view_state      # ViewState si la acción ocurrió dentro de un modal

Otros métodos de ViewState: .as_dict() (todos los valores) y .block_id_of(action_id). También accede por clave (sub.state["detalle"]).

Sobres tipados disponibles: BlockAction, ViewSubmission, ViewClosed, Shortcut, SlackEvent. Para dispatch manual (sin decoradores), parse_payload(raw) despacha según raw["type"] al sobre correcto.

from duppla_slack import parse_payload

evt = parse_payload(raw)   # -> BlockAction | ViewSubmission | ViewClosed | Shortcut | SlackEvent

Block Kit

El módulo duppla_slack.block_builder ofrece factorías tipadas para construir bloques y vistas.

from duppla_slack import block_builder as bb

modal = bb.ModalView(
    title=bb.PlainText("Reporte"),
    submit=bb.PlainText("Enviar"),
    blocks=[
        bb.Header(text=bb.PlainText("Nuevo reporte")),
        bb.Divider(),
        bb.Section(text=bb.Markdown("Describe el *incidente*:")),
        bb.InputBlock(
            label=bb.PlainText("Detalle"),
            element=bb.Input(action_id="detalle", multiline=True),
        ),
        bb.ActionsBlock(elements=[
            bb.Button(text=bb.PlainText("Confirmar"), action_id="confirm"),
        ]),
    ],
)

slack.open_modal(trigger_id="…", view=modal)

Builders de alto nivel

El mismo módulo ofrece factorías que arman vistas completas y reducen el boilerplate.

from duppla_slack import block_builder as bb

# Modal de formulario con inputs; input_block toma el block_id del action_id del elemento
modal = bb.form_modal(
    title="Nuevo reporte",
    callback_id="reporte",
    blocks=[
        bb.input_block("Detalle", bb.Input(action_id="detalle", multiline=True)),
        bb.input_block(
            "Prioridad",
            bb.Select("static_select", action_id="prioridad",
                      options=bb.options_from({"low": "Baja", "high": "Alta"})),
        ),
    ],
)
slack.open_modal(trigger_id="…", view=modal)

Otras factorías de vistas: notice_modal(title, text), loading_modal() (respuesta inmediata a views.open antes de una actualización lenta), error_modal(text) y prefill(view, {action_id: valor}) para poblar los valores iniciales al reabrir un modal en modo edición.

Helpers de bloques y de mrkdwn:

bb.fields_section([("Estado", "Abierto"), ("Prioridad", "Alta")])  # section con fields
bb.options_from({"low": "Baja", "high": "Alta"})                   # opciones {value: label}

bb.mention("U123")                 # <@U123>
bb.mentions(["U123", "U456"])      # <@U123> <@U456>
bb.channel_link("C123")            # <#C123>
bb.link("https://duppla.com", "duppla")
bb.emoji("tada")                   # :tada:

Manejo de errores

Las excepciones son tipadas y derivan de DupplaSlackError. SlackApiError subclasea la de slack_sdk y expone .error_code, con subclases codificadas.

from duppla_slack.errors import (
    DupplaSlackError, SlackApiError, NotInChannelError,
    ChannelNotFoundError, RateLimitedError, AuthenticationError,
    MissingScopeError, ConfigurationError, MissingDependencyError,
)

try:
    slack.send_msg_(text="Hola", channel="C0123456789")
except NotInChannelError:
    ...  # el servicio intenta auto-unirse y reintentar en not_in_channel
except SlackApiError as exc:
    print(exc.error_code)

El servicio se une automáticamente al canal y reintenta ante not_in_channel. MissingDependencyError se lanza si usas una integración opcional sin su extra instalado.

Conveniencias del servicio

SlackService incluye atajos para reemplazar helpers repetidos entre proyectos.

# Volcar cualquier contenido a un canal: modelos/dicts/listas se formatean como JSON
# y, si el texto es largo, se sube como snippet (evita el límite de 4000 caracteres)
slack.log({"deploy": "ok", "version": "1.2.3"}, channel="C0123456789")

# Descargar un archivo privado de Slack (autentica con el bot token)
data = slack.download_file(url_private)          # y await slack.adownload_file(...)

# Responder a un response_url de una interacción (slash/action)
slack.respond(response_url, text="Listo", response_type="ephemeral")  # y arespond(...)

# Leer un mensaje puntual por timestamp (None si no existe)
msg = slack.get_message(channel="C0123456789", ts="1700000000.000100")

# Canales: crear si no existe, archivar, invitar/expulsar async
cid = slack.ensure_channel("incidentes")         # id existente o recién creado
slack.archive_channel(cid)
await slack.ainvite_channel(cid, "U123")
await slack.akick_channel(cid, "U123")

# Persona: un sender con identidad/canal fijos
ticket = slack.as_(username="Ticketera", icon_emoji=":ticket:", channel="C0123456789")
ticket.send(text="Nuevo ticket")                 # y await ticket.asend(...)

# Directorio de miembros con caché en proceso (TTL en segundos)
directory = slack.member_directory(ttl=300)      # {user_id: SlackProfile}
names = slack.member_names(ttl=300)              # {user_id: real_name}

Con el extra fastapi, install_error_handlers(app) registra un handler que mapea SlackApiError a una respuesta JSON (y los rate limits a 429).

from fastapi import FastAPI
from duppla_slack.integrations.fastapi_adapter import (
    install_error_handlers, slack_exception_handler,
)

app = FastAPI()
install_error_handlers(app)   # o app.add_exception_handler(SlackApiError, slack_exception_handler)

Integraciones opcionales

Redis (duppla_slack[redis])

Rate limiter distribuido (auto-conectado cuando defines redis_url) y cache de lookups.

from duppla_slack import SlackService, SlackConfig
from duppla_slack.integrations.redis_ext import CachedSlackService

# Auto-conecta el rate limiter distribuido al definir redis_url
slack = SlackService(config=SlackConfig(token="xoxb-…", redis_url="redis://localhost:6379/0"))

# Servicio con cache de lookups
cached = CachedSlackService.from_url(token="xoxb-…", redis_url="redis://localhost:6379/0")

Puedes inyectar tu propio cliente Redis (por ejemplo pooled/TLS) con from_client, y usar un serializer conectable (por defecto json):

import redis
from duppla_slack.integrations.redis_ext import (
    RedisCache, RedisRateLimiter, RedisLock, RedisSlidingWindow,
    AsyncRedisIdempotencyStore,
)

client = redis.Redis.from_url("redis://localhost:6379/0")
cache = RedisCache.from_client(client)              # serializer=... opcional
limiter = RedisRateLimiter.from_client(client)

# Lock distribuido (context manager, con owner-token)
with RedisLock(client, "rebuild-index"):
    ...  # solo un proceso a la vez

# Primitiva general de sliding window
allowed, retry_after = RedisSlidingWindow(client).hit("key", limit=5, period=60)

# Seam de idempotencia (async): True la primera vez, False si ya se vio
import redis.asyncio as aioredis
store = AsyncRedisIdempotencyStore.from_client(aioredis.Redis.from_url("redis://localhost:6379/0"))
first_seen = await store.check_and_set("event-id")

Existen las variantes async AsyncRedisCache, AsyncRedisLock y AsyncRedisSlidingWindow.

RabbitMQ (duppla_slack[rabbitmq])

Publica eventos entrantes como middleware de Bolt y consume una cola para disparar envíos.

from duppla_slack.integrations.rabbitmq import (
    RabbitMQSettings, EventPublisher, SendConsumer, SendCommand,
)

settings = RabbitMQSettings()  # lee de entorno

# Publica cada evento entrante en RabbitMQ
publisher = EventPublisher(settings)
slack.middleware(publisher)

# Consume una cola y envia por Slack
consumer = SendConsumer(settings, slack)
# await consumer.start()  -> procesa SendCommand(...) de la cola; await consumer.stop() para cerrar

El serializer es conectable (por defecto json; puedes pasar uno basado en orjson). SendConsumer acepta un AckMode con política dead-letter-first (nunca descarta en silencio):

from duppla_slack.integrations.rabbitmq import AckMode

# AUTO_ALL (default): ack en éxito; en fallo hace reject hacia el dead-letter (o descarta si no hay DLX).
# AUTO_RETRY: reintenta una vez (según message.redelivered) y luego dead-letter/descarta.
# NONE: consume con no_ack. (MANUAL no aplica a SendConsumer: no expone un handler que confirme el mensaje.)
consumer = SendConsumer(settings, slack, ack_mode=AckMode.AUTO_RETRY)

Los campos DLX viven en RabbitMQSettings: dead_letter_exchange y dead_letter_routing_key. Para deduplicar redeliveries hay un seam de idempotencia opcional: pasa un IdempotencyStore (Protocol con check_and_set) junto a una función fingerprint(cmd) -> str.

consumer = SendConsumer(
    settings, slack,
    idempotency=store,                       # p. ej. AsyncRedisIdempotencyStore
    fingerprint=lambda cmd: cmd.payload.get("channel", "") + cmd.payload.get("ts", ""),
)

Migración / cambios importantes

• Ahora el paquete envuelve slack_bolt (no solo el slack_sdk) y la app subyacente es asíncrona.
• get_channel(channel_id=…) y get_channel(channel_name=…) ahora ambos devuelven el objeto de canal ({"id", "name", "is_archived", …}) o None. Antes, la ruta por id devolvía la respuesta completa de la API (con ok/channel).
• SlackApiError ahora se construye en la librería con respuestas estructuradas reales y es subclase de slack_sdk.errors.SlackApiError.
• Los alias en español siguen siendo de primera clase: crear_canal/create_channel, invitar_canal/invite_channel, uninvitar_canal/kick_channel, channel_exists.

Desarrollo

El proyecto usa uv.

uv sync --all-extras
uv run ruff check .
uv run ruff format --check .
uv run pyright duppla_slack
uv run pytest

Despliegue

Alternativa 1: GitHub Actions

Ejecuta la acción Deploy to PyPI (on demand) en la pestaña Actions. Calcula una versión CalVer, hace uv build y uv publish.

Alternativa 2: CLI

uv build
uv publish

Recuerda que necesitas un API token; pregunta al dueño del repositorio o a los administradores de la organización.

Recuerda

El objetivo de este repositorio es tener cosas simples, funcionales y que sean de utilidad para la organización.