reddit_channel

reddit_channel

RedditChannel — Reddit adapter via praw.

Classes

RedditChannel

RedditChannel(
    client_id: str = "",
    *,
    client_secret: str = "",
    username: str = "",
    password: str = "",
    user_agent: str = "",
    bus: Optional[EventBus] = None,
)

Bases: BaseChannel

Reddit messaging channel adapter.

Uses the Reddit API via praw (Python Reddit API Wrapper).

PARAMETER	DESCRIPTION
client_id	Reddit app client ID. Falls back to REDDIT_CLIENT_ID env var. TYPE: str DEFAULT: ''
client_secret	Reddit app client secret. Falls back to REDDIT_CLIENT_SECRET env var. TYPE: str DEFAULT: ''
username	Reddit username. Falls back to REDDIT_USERNAME env var. TYPE: str DEFAULT: ''
password	Reddit password. Falls back to REDDIT_PASSWORD env var. TYPE: str DEFAULT: ''
user_agent	User-agent string for API requests. Falls back to REDDIT_USER_AGENT env var. TYPE: str DEFAULT: ''
bus	Optional event bus for publishing channel events. TYPE: Optional[EventBus] DEFAULT: None

Source code in src/diapason/channels/reddit_channel.py

def __init__(
    self,
    client_id: str = "",
    *,
    client_secret: str = "",
    username: str = "",
    password: str = "",
    user_agent: str = "",
    bus: Optional[EventBus] = None,
) -> None:
    self._client_id = client_id or os.environ.get("REDDIT_CLIENT_ID", "")
    self._client_secret = client_secret or os.environ.get(
        "REDDIT_CLIENT_SECRET", ""
    )
    self._username = username or os.environ.get("REDDIT_USERNAME", "")
    self._password = password or os.environ.get("REDDIT_PASSWORD", "")
    self._user_agent = user_agent or os.environ.get(
        "REDDIT_USER_AGENT", "diapason:v1.0"
    )
    self._bus = bus
    self._handlers: List[ChannelHandler] = []
    self._status = ChannelStatus.DISCONNECTED
    self._reddit: Any = None

Methods:

connect

connect() -> None

Validate credentials and mark as connected.

Source code in src/diapason/channels/reddit_channel.py

def connect(self) -> None:
    """Validate credentials and mark as connected."""
    if not self._client_id or not self._client_secret:
        logger.warning("No Reddit client_id or client_secret configured")
        self._status = ChannelStatus.ERROR
        return
    try:
        import praw  # noqa: F401
    except ImportError:
        raise ImportError(
            "praw not installed. Install with: uv sync --extra channel-reddit"
        )
    self._status = ChannelStatus.CONNECTED

disconnect

disconnect() -> None

Mark as disconnected.

Source code in src/diapason/channels/reddit_channel.py

def disconnect(self) -> None:
    """Mark as disconnected."""
    self._reddit = None
    self._status = ChannelStatus.DISCONNECTED

send

send(
    channel: str,
    content: str,
    *,
    conversation_id: str = "",
    metadata: Dict[str, Any] | None = None,
) -> bool

Send a message or comment on Reddit.

PARAMETER	DESCRIPTION
channel	Subreddit name (without r/ prefix) or a Reddit thing ID to reply to. TYPE: str
content	Text content for the submission or comment. TYPE: str

Source code in src/diapason/channels/reddit_channel.py

def send(
    self,
    channel: str,
    content: str,
    *,
    conversation_id: str = "",
    metadata: Dict[str, Any] | None = None,
) -> bool:
    """Send a message or comment on Reddit.

    Parameters
    ----------
    channel:
        Subreddit name (without ``r/`` prefix) or a Reddit thing ID to
        reply to.
    content:
        Text content for the submission or comment.
    """
    if not self._client_id or not self._client_secret:
        logger.warning("Cannot send: no Reddit credentials configured")
        return False

    try:
        import praw

        reddit = praw.Reddit(
            client_id=self._client_id,
            client_secret=self._client_secret,
            username=self._username,
            password=self._password,
            user_agent=self._user_agent,
        )

        if conversation_id:
            # Reply to an existing comment/submission
            comment = reddit.comment(conversation_id)
            comment.reply(content)
        else:
            # Submit as a new text post to the subreddit
            subreddit = reddit.subreddit(channel)
            title = (metadata or {}).get("title", "Diapason Message")
            subreddit.submit(title=title, selftext=content)

        self._publish_sent(channel, content, conversation_id)
        return True
    except ImportError:
        logger.debug("praw not installed")
        return False
    except Exception:
        logger.debug("Reddit send failed", exc_info=True)
        return False

status

status() -> ChannelStatus

Return the current connection status.

Source code in src/diapason/channels/reddit_channel.py

def status(self) -> ChannelStatus:
    """Return the current connection status."""
    return self._status

list_channels

list_channels() -> List[str]

Return available channel identifiers.

Source code in src/diapason/channels/reddit_channel.py

def list_channels(self) -> List[str]:
    """Return available channel identifiers."""
    return ["reddit"]

on_message

on_message(handler: ChannelHandler) -> None

Register a callback for incoming messages.

Source code in src/diapason/channels/reddit_channel.py

def on_message(self, handler: ChannelHandler) -> None:
    """Register a callback for incoming messages."""
    self._handlers.append(handler)