### Python Factory Pattern for Notifier Management Source: https://context7.com/loonghao/notify-bridge/llms.txt Explains how the factory pattern is used within `notify-bridge` to manage notifier instances. It shows how to retrieve a list of all registered notifiers and how to get a specific notifier class by its name. ```python from notify_bridge import NotifyBridge bridge = NotifyBridge() # List all registered notifiers notifiers = bridge.notifiers print(f"Available notifiers: {notifiers}") # Output: ['feishu', 'wecom', 'github', 'notify'] # Get notifier class notifier_class = bridge.get_notifier_class("feishu") print(f"Notifier class: {notifier_class}") ``` -------------------------------- ### Send Simple Text Message to Feishu Source: https://context7.com/loonghao/notify-bridge/llms.txt A straightforward example of sending a plain text notification to Feishu using the NotifyBridge. It utilizes the 'feishu' notifier with a provided webhook URL and text content. ```python from notify_bridge import NotifyBridge bridge = NotifyBridge() # Simple text notification response = bridge.send( "feishu", webhook_url="https://open.feishu.cn/open-apis/bot/v2/hook/YOUR-WEBHOOK-KEY", content="This is a simple text message from notify-bridge.", msg_type="text" ) print(f"Status: {response.success}") print(f"Response data: {response.data}") ``` -------------------------------- ### Secure Configuration with Environment Variables in Python Source: https://context7.com/loonghao/notify-bridge/llms.txt Illustrates how to securely manage sensitive credentials and configuration settings using environment variables with the `dotenv` library and `notify-bridge`. It shows loading variables from a `.env` file and using `os.getenv` for services like Feishu and GitHub. ```python import os from dotenv import load_dotenv from notify_bridge import NotifyBridge # Load environment variables from .env file load_dotenv() # .env file contents: # FEISHU_WEBHOOK_URL=https://open.feishu.cn/open-apis/bot/v2/hook/xxx # WECOM_WEBHOOK_URL=https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxx # GITHUB_TOKEN=ghp_xxxxxxxxxxxx # NOTIFY_BASE_URL=https://notify.example.com bridge = NotifyBridge() # Use environment variables for sensitive data response = bridge.send( "feishu", webhook_url=os.getenv("FEISHU_WEBHOOK_URL"), content="Secure notification using environment variables", msg_type="text" ) # GitHub with token from environment response = bridge.send( "github", owner="your-username", repo="your-repo", token=os.getenv("GITHUB_TOKEN"), title="Automated Issue", message="Created from secure configuration", labels=["automated"], msg_type="text" ) print(f"Secure notifications sent: {response.success}") ``` -------------------------------- ### Create and Inspect Feishu Notifier Source: https://context7.com/loonghao/notify-bridge/llms.txt Demonstrates how to create a Feishu notifier instance using the bridge and then access its properties like name and supported notification types. It also shows how to retrieve and display all registered notifier classes. ```python feishu_notifier = bridge.create_notifier("feishu") print(f"Notifier name: {feishu_notifier.name}") print(f"Supported types: {feishu_notifier.supported_types}") registered = bridge.get_registered_notifiers() for name, notifier_cls in registered.items(): print(f"{name}: {notifier_cls}") ``` -------------------------------- ### Python Context Managers for Resource Management Source: https://context7.com/loonghao/notify-bridge/llms.txt Demonstrates the use of context managers (`with` statement) in Python for reliable resource management with `NotifyBridge`. It covers both synchronous and asynchronous usage, ensuring that resources like HTTP clients are properly closed after use. ```python from notify_bridge import NotifyBridge import asyncio # Synchronous context manager with NotifyBridge() as bridge: response = bridge.send( "feishu", webhook_url="https://open.feishu.cn/open-apis/bot/v2/hook/YOUR-KEY", content="Message sent with automatic cleanup", msg_type="text" ) print(f"Sent: {response.success}") # HTTP client automatically closed after exiting context # Asynchronous context manager async def send_with_context(): async with NotifyBridge() as bridge: response = await bridge.send_async( "feishu", webhook_url="https://open.feishu.cn/open-apis/bot/v2/hook/YOUR-KEY", content="Async message with automatic cleanup", msg_type="text" ) return response response = asyncio.run(send_with_context()) print(f"Async sent: {response.success}") ``` -------------------------------- ### Initialize and Send Synchronous Notification with NotifyBridge Source: https://context7.com/loonghao/notify-bridge/llms.txt Demonstrates initializing the NotifyBridge and sending a synchronous text notification to Feishu. It requires the 'notify-bridge' library and a Feishu webhook URL. Returns a NotificationResponse object. ```python from notify_bridge import NotifyBridge # Initialize the bridge bridge = NotifyBridge() # Send a synchronous notification to Feishu response = bridge.send( "feishu", webhook_url="https://open.feishu.cn/open-apis/bot/v2/hook/YOUR-WEBHOOK-KEY", content="Hello from notify-bridge! This is a test message.", msg_type="text" ) # Response: NotificationResponse(success=True, name='feishu', message='Notification sent successfully', data={...}) print(f"Success: {response.success}, Message: {response.message}") ``` -------------------------------- ### Send Rich Post Message to Feishu Source: https://context7.com/loonghao/notify-bridge/llms.txt Demonstrates sending a rich post message to Feishu, allowing for formatted text, links, and structured content. Requires 'notify-bridge' and correctly formatted 'post_content' dictionary. ```python from notify_bridge import NotifyBridge bridge = NotifyBridge() # Create rich post content with formatting post_content = { "zh_cn": [ [ {"tag": "text", "text": "Project Update\n\n"}, ], [ {"tag": "text", "text": "Status: "}, {"tag": "text", "text": "Completed", "text_type": "bold"}, {"tag": "text", "text": "\n"}, ], [ {"tag": "text", "text": "Progress: "}, {"tag": "text", "text": "100%", "text_type": "bold"}, ], [ {"tag": "text", "text": "\n\nView details at "}, {"tag": "a", "text": "Dashboard", "href": "https://example.com/dashboard"} ] ] } response = bridge.send( "feishu", webhook_url="https://open.feishu.cn/open-apis/bot/v2/hook/YOUR-KEY", post_content=post_content, msg_type="post" ) print(f"Post message sent: {response.success}") ``` -------------------------------- ### Create GitHub Issues with Markdown Content (Python) Source: https://context7.com/loonghao/notify-bridge/llms.txt Creates GitHub issues using markdown formatting for rich content. This function allows for structured issue descriptions, requirements lists, and embedding code snippets within the issue body. Supports 'markdown' message type. ```python from notify_bridge import NotifyBridge bridge = NotifyBridge() # Create issue with markdown content markdown_issue = """ ## Description User authentication module needs refactoring to support OAuth 2.0. ## Requirements - [ ] Implement OAuth 2.0 flow - [ ] Add support for Google login - [ ] Add support for GitHub login - [ ] Update documentation ## Technical Details ```python # Example code structure class OAuthHandler: def authenticate(self, provider: str) -> Token: pass ``` ## Timeline Expected completion: 2 weeks """ response = bridge.send( "github", owner="your-username", repo="your-repository", token="ghp_YOUR_GITHUB_TOKEN", title="Feature: OAuth 2.0 Authentication", message=markdown_issue, labels=["enhancement", "authentication"], msg_type="markdown" ) print(f"Markdown issue created: {response.success}") ``` -------------------------------- ### Send News Message with Articles (Python) Source: https://context7.com/loonghao/notify-bridge/llms.txt Sends a news message to a specified webhook URL with a list of articles. Requires the 'wecom' integration and provides details for each article like title, description, URL, and image. ```python response = bridge.send( "wecom", webhook_url="https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=YOUR-KEY", msg_type="news", mentioned_list=["@all"], articles=[ { "title": "New Feature Released: Dark Mode", "description": f"We've launched dark mode support. Released on {datetime.now().strftime('%Y-%m-%d')}", "url": "https://example.com/blog/dark-mode", "picurl": "https://example.com/images/dark-mode-preview.png" }, { "title": "Performance Improvements", "description": "Application now loads 50% faster with our latest optimizations.", "url": "https://example.com/blog/performance", "picurl": "https://example.com/images/performance.png" } ] ) print(f"News message sent: {response.success}") ``` -------------------------------- ### Create GitHub Issues with Labels and Assignees (Python) Source: https://context7.com/loonghao/notify-bridge/llms.txt Creates a GitHub issue programmatically using the Notify Bridge. It requires repository details, an authentication token, and allows specifying the issue title, message content, labels, and assignees. Supports 'text' message type. ```python from notify_bridge import NotifyBridge bridge = NotifyBridge() # Create a GitHub issue response = bridge.send( "github", owner="your-username", repo="your-repository", token="ghp_YOUR_GITHUB_TOKEN", title="Bug: Login page not responsive on mobile", message="The login page doesn't display correctly on mobile devices.\n\n**Steps to reproduce:**\n1. Open login page on mobile\n2. Observe layout issues\n\n**Expected:** Responsive layout\n**Actual:** Elements overlap", labels=["bug", "mobile", "priority-high"], assignees=["developer1", "developer2"], msg_type="text" ) print(f"Issue created: {response.success}") print(f"Issue URL: {response.data.get('html_url')}") ``` -------------------------------- ### Create and Use Custom Slack Notifier in Python Source: https://context7.com/loonghao/notify-bridge/llms.txt This snippet shows how to define a custom notifier for Slack by inheriting from BaseNotifier, implementing `assemble_data`, and then registering and using this custom notifier with the NotifyBridge. It demonstrates sending a basic text message. ```python from notify_bridge import BaseNotifier, MessageType from typing import Dict, Any, ClassVar # Assuming SlackSchema is defined elsewhere class SlackSchema: # Placeholder for actual schema def __init__(self, content, username, icon_emoji, channel=None, msg_type=None): self.content = content self.username = username self.icon_emoji = icon_emoji self.channel = channel self.msg_type = msg_type class SlackNotifier(BaseNotifier): """Slack notifier implementation.""" name = "slack" schema_class = SlackSchema supported_types: ClassVar[set[MessageType]] = {MessageType.TEXT, MessageType.MARKDOWN} def assemble_data(self, data: SlackSchema) -> Dict[str, Any]: """Build Slack API payload.""" payload = { "text": data.content, "username": data.username, "icon_emoji": data.icon_emoji } if data.channel: payload["channel"] = data.channel if data.msg_type == MessageType.MARKDOWN: payload["mrkdwn"] = True return payload # Register and use custom notifier bridge = NotifyBridge() bridge.register_notifier("slack", SlackNotifier) response = bridge.send( "slack", webhook_url="https://hooks.slack.com/services/YOUR/WEBHOOK/URL", content="Hello from custom Slack notifier!", username="MyBot", icon_emoji=":rocket:", msg_type="text" ) print(f"Custom notifier sent: {response.success}") ``` -------------------------------- ### Python Exception Handling for Notifications Source: https://context7.com/loonghao/notify-bridge/llms.txt Demonstrates robust error handling for notification failures using the notify-bridge library. It catches specific exceptions like `ValidationError`, `NotificationError`, `NoSuchNotifierError`, and `ConfigurationError`, providing detailed error information for debugging. ```python from notify_bridge import ( NotifyBridge, NotificationError, ValidationError, NoSuchNotifierError, ConfigurationError ) bridge = NotifyBridge() try: # Attempt to send notification response = bridge.send( "feishu", webhook_url="https://invalid-url.com", content="Test message", msg_type="text" ) except ValidationError as e: print(f"Validation failed: {e}") print(f"Invalid data: {e.data}") print(f"Validation errors: {e.errors}") except NotificationError as e: print(f"Notification failed: {e}") print(f"Notifier: {e.notifier_name}") print(f"Response: {e.response}") if e.original_exception: print(f"Original error: {e.original_exception}") except NoSuchNotifierError as e: print(f"Notifier not found: {e.notifier_name}") print(f"Available notifiers: {e.available_notifiers}") except ConfigurationError as e: print(f"Configuration error: {e}") print(f"Config key: {e.config_key}") print(f"Config value: {e.config_value}") except Exception as e: print(f"Unexpected error: {e}") ``` -------------------------------- ### Synchronous Notification Sending with Error Handling Source: https://context7.com/loonghao/notify-bridge/llms.txt Shows how to send a synchronous notification using NotifyBridge and includes robust error handling for potential NotificationError and ValidationError exceptions. Requires 'notify-bridge' and 'pydantic'. ```python from notify_bridge import NotifyBridge, NotificationError from pydantic import ValidationError bridge = NotifyBridge() try: # Send text message with error handling response = bridge.send( "feishu", webhook_url="https://open.feishu.cn/open-apis/bot/v2/hook/YOUR-KEY", content="Important notification message", msg_type="text" ) if response.success: print(f"Notification sent successfully: {response.data}") else: print(f"Failed to send: {response.message}") except NotificationError as e: print(f"Notification error: {e}") print(f"Notifier: {e.notifier_name}") except ValidationError as e: print(f"Validation error: {e}") ``` -------------------------------- ### Asynchronous Notification Sending with NotifyBridge Source: https://context7.com/loonghao/notify-bridge/llms.txt Illustrates sending multiple notifications concurrently using the asynchronous `send_async` method of NotifyBridge. This is suitable for async applications and requires 'notify-bridge' and 'asyncio'. ```python import asyncio from notify_bridge import NotifyBridge async def send_multiple_notifications(): bridge = NotifyBridge() # Send multiple notifications concurrently tasks = [ bridge.send_async( "feishu", webhook_url="https://open.feishu.cn/open-apis/bot/v2/hook/YOUR-KEY", content=f"Async notification {i}", msg_type="text" ) for i in range(3) ] responses = await asyncio.gather(*tasks) for i, response in enumerate(responses): print(f"Notification {i+1}: {response.success} - {response.message}") return responses # Run the async function responses = asyncio.run(send_multiple_notifications()) ``` -------------------------------- ### Send Feishu Image Messages Source: https://context7.com/loonghao/notify-bridge/llms.txt Sends an image message to Feishu using a local file path. Requires the 'notify_bridge' library and a valid webhook URL. The image path is converted to a string. ```python from notify_bridge import NotifyBridge from pathlib import Path bridge = NotifyBridge() # Send image from local file image_path = Path("/path/to/your/image.png") if image_path.exists(): response = bridge.send( "feishu", webhook_url="https://open.feishu.cn/open-apis/bot/v2/hook/YOUR-KEY", image_path=str(image_path), msg_type="image" ) print(f"Image sent: {response.success}") else: print(f"Image file not found at {image_path}") ``` -------------------------------- ### Send WeCom Image Messages Source: https://context7.com/loonghao/notify-bridge/llms.txt Sends an image message to WeCom by providing a local image file path. Requires the 'notify_bridge' library and a valid WeCom webhook URL. The image path is converted to a string. ```python from notify_bridge import NotifyBridge from pathlib import Path bridge = NotifyBridge() # Send image message image_path = Path("/path/to/chart.png") if image_path.exists(): response = bridge.send( "wecom", webhook_url="https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=YOUR-KEY", msg_type="image", image_path=str(image_path) ) print(f"Image sent: {response.success}") print(f"Response: {response.data}") else: print(f"Image not found at {image_path}") ``` -------------------------------- ### Send Generic Notifications (Python) Source: https://context7.com/loonghao/notify-bridge/llms.txt Sends notifications to a generic notification service. This can include custom base URLs, titles, messages, tags, icons, and colors. It also supports sending notifications with an authentication bearer token. ```python from notify_bridge import NotifyBridge bridge = NotifyBridge() # Send to generic notify service response = bridge.send( "notify", base_url="https://notify-demo.deno.dev", title="System Alert", message="Database backup completed successfully at 02:00 AM.", tags=["backup", "database", "success"], icon="https://example.com/icons/success.png", color="#00ff00", msg_type="text" ) print(f"Notification sent: {response.success}") # Send with authentication token response = bridge.send( "notify", base_url="https://your-notify-server.com", token="your-bearer-token", title="Deployment Complete", message="Application v2.0.0 deployed to production.", tags=["deployment", "production"], msg_type="text" ) ``` -------------------------------- ### Send Feishu Interactive Card Messages Source: https://context7.com/loonghao/notify-bridge/llms.txt Sends an interactive card message to Feishu with structured content, including headers, text elements, and action buttons. Requires 'notify_bridge' and a Feishu webhook URL. Card structure is defined using dictionaries. ```python from notify_bridge import NotifyBridge bridge = NotifyBridge() # Create interactive card with header and elements response = bridge.send( "feishu", webhook_url="https://open.feishu.cn/open-apis/bot/v2/hook/YOUR-KEY", msg_type="interactive", card_header={ "title": "Deployment Notification", "template": "blue" # Color options: blue, red, green, orange, etc. }, card_elements=[ { "tag": "div", "text": { "tag": "plain_text", "content": "Application deployed successfully to production." } }, { "tag": "div", "text": { "tag": "lark_md", "content": "**Version:** 2.1.0\n**Environment:** Production\n**Time:** 2024-01-15 14:30:00" } }, { "tag": "action", "actions": [ { "tag": "button", "text": {"tag": "plain_text", "content": "View Logs"}, "url": "https://example.com/logs", "type": "primary" } ] } ] ) print(f"Interactive card sent: {response.success}") ``` -------------------------------- ### Send WeCom Markdown Messages Source: https://context7.com/loonghao/notify-bridge/llms.txt Sends markdown formatted messages to WeCom. Supports automatic formatting for standard markdown or raw markdown (msg_type='markdown_v2') to preserve special characters. Requires 'notify_bridge' and a WeCom webhook URL. ```python from notify_bridge import NotifyBridge bridge = NotifyBridge() # Send markdown message (formatted automatically) markdown_content = """ # Project Status Report **Date:** 2024-01-15 ## Progress • Backend API: Completed • Frontend UI: In Progress • Testing: Pending ## Issues > Critical bug found in authentication module [View Details](https://example.com/report) """ response = bridge.send( "wecom", webhook_url="https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=YOUR-KEY", content=markdown_content, msg_type="markdown", mentioned_list=["@all"] ) print(f"Markdown sent: {response.success}") # Send raw markdown without formatting (preserves underscores, etc.) response = bridge.send( "wecom", webhook_url="https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=YOUR-KEY", content="Raw markdown with `code_variable` and __underscores__", msg_type="markdown_v2" # No automatic formatting applied ) ``` -------------------------------- ### Send WeCom Text Messages with Mentions Source: https://context7.com/loonghao/notify-bridge/llms.txt Sends plain text messages to WeCom, supporting mentions of all users ('@all') or specific users by their mobile numbers. Requires 'notify_bridge' and a WeCom webhook URL. ```python from notify_bridge import NotifyBridge bridge = NotifyBridge() # Send text with @mentions response = bridge.send( "wecom", webhook_url="https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=YOUR-KEY", content="Project meeting in 10 minutes!", msg_type="text", mentioned_list=["@all"] # Mention all members, or use specific user IDs ) print(f"Text message sent: {response.success}") # Mention specific users by mobile number response = bridge.send( "wecom", webhook_url="https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=YOUR-KEY", content="Please review the pull request.", msg_type="text", mentioned_mobile_list=["13800138000", "13900139000"] ) ``` -------------------------------- ### Define Custom Slack Notifier Schema (Python) Source: https://context7.com/loonghao/notify-bridge/llms.txt Defines a custom Pydantic schema for Slack notifications by extending the WebhookSchema. This schema includes fields specific to Slack, such as webhook URL, message content, channel, bot username, and icon emoji. ```python from notify_bridge import BaseNotifier, NotificationSchema, NotificationResponse from notify_bridge.schema import WebhookSchema, MessageType from pydantic import Field from typing import ClassVar, Dict, Any # Define custom schema class SlackSchema(WebhookSchema): """Schema for Slack notifications.""" webhook_url: str = Field(..., description="Slack webhook URL") content: str = Field(..., description="Message content") channel: str = Field(None, description="Channel override") username: str = Field("notify-bridge", description="Bot username") icon_emoji: str = Field(":robot_face:", description="Bot icon emoji") # Note: The actual implementation of BaseNotifier and its usage would follow. ``` === COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.