Google Cloud Monitoring API client library for collecting, analyzing, and alerting on metrics, events, and metadata from cloud and on-premise sources.
—
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
Comprehensive notification channel management for delivering alert notifications via email, SMS, Slack, PagerDuty, and other supported channels with verification and configuration management in Google Cloud Monitoring.
Manage the complete lifecycle of notification channels including creation, updates, retrieval, deletion, and verification.
class NotificationChannelServiceClient:
def list_notification_channels(
self,
request=None,
*,
name: str = None,
retry=None,
timeout=None,
metadata=()
) -> pagers.ListNotificationChannelsPager:
"""
Lists the notification channels that have been created.
Args:
request: The request object or dict equivalent
name: Required. Project name in format 'projects/[PROJECT_ID]'
retry: Retry configuration
timeout: Request timeout in seconds
metadata: Additional metadata
Returns:
Pager for iterating over NotificationChannel objects
"""
def get_notification_channel(
self,
request=None,
*,
name: str = None,
retry=None,
timeout=None,
metadata=()
) -> notification_pb2.NotificationChannel:
"""
Gets a single notification channel.
Args:
request: The request object or dict equivalent
name: Required. Channel name in format 'projects/[PROJECT_ID]/notificationChannels/[CHANNEL_ID]'
retry: Retry configuration
timeout: Request timeout in seconds
metadata: Additional metadata
Returns:
NotificationChannel object
"""
def create_notification_channel(
self,
request=None,
*,
name: str = None,
notification_channel=None,
retry=None,
timeout=None,
metadata=()
) -> notification_pb2.NotificationChannel:
"""
Creates a new notification channel.
Args:
request: The request object or dict equivalent
name: Required. Project name
notification_channel: Required. The notification channel to create
retry: Retry configuration
timeout: Request timeout in seconds
metadata: Additional metadata
Returns:
Created NotificationChannel object
"""
def update_notification_channel(
self,
request=None,
*,
update_mask=None,
notification_channel=None,
retry=None,
timeout=None,
metadata=()
) -> notification_pb2.NotificationChannel:
"""
Updates a notification channel.
Args:
request: The request object or dict equivalent
update_mask: Field mask for selective updates
notification_channel: Required. Updated notification channel
retry: Retry configuration
timeout: Request timeout in seconds
metadata: Additional metadata
Returns:
Updated NotificationChannel object
"""
def delete_notification_channel(
self,
request=None,
*,
name: str = None,
force: bool = None,
retry=None,
timeout=None,
metadata=()
) -> None:
"""
Deletes a notification channel.
Args:
request: The request object or dict equivalent
name: Required. Channel name to delete
force: Force deletion even if channel is referenced
retry: Retry configuration
timeout: Request timeout in seconds
metadata: Additional metadata
"""Manage notification channel verification including sending and retrieving verification codes.
class NotificationChannelServiceClient:
def send_notification_channel_verification_code(
self,
request=None,
*,
name: str = None,
retry=None,
timeout=None,
metadata=()
) -> None:
"""
Causes a verification code to be delivered to the channel.
Args:
request: The request object or dict equivalent
name: Required. Channel name
retry: Retry configuration
timeout: Request timeout in seconds
metadata: Additional metadata
"""
def get_notification_channel_verification_code(
self,
request=None,
*,
name: str = None,
retry=None,
timeout=None,
metadata=()
) -> notification_service.GetNotificationChannelVerificationCodeResponse:
"""
Requests a verification code for an already verified channel.
Args:
request: The request object or dict equivalent
name: Required. Channel name
retry: Retry configuration
timeout: Request timeout in seconds
metadata: Additional metadata
Returns:
GetNotificationChannelVerificationCodeResponse object
"""
def verify_notification_channel(
self,
request=None,
*,
name: str = None,
code: str = None,
retry=None,
timeout=None,
metadata=()
) -> notification_pb2.NotificationChannel:
"""
Verifies a NotificationChannel by proving receipt of the code.
Args:
request: The request object or dict equivalent
name: Required. Channel name
code: Required. Verification code
retry: Retry configuration
timeout: Request timeout in seconds
metadata: Additional metadata
Returns:
Verified NotificationChannel object
"""Retrieve information about supported notification channel types and their configurations.
class NotificationChannelServiceClient:
def list_notification_channel_descriptors(
self,
request=None,
*,
name: str = None,
retry=None,
timeout=None,
metadata=()
) -> pagers.ListNotificationChannelDescriptorsPager:
"""
Lists the descriptors for supported notification channels.
Args:
request: The request object or dict equivalent
name: Required. Project name
retry: Retry configuration
timeout: Request timeout in seconds
metadata: Additional metadata
Returns:
Pager for iterating over NotificationChannelDescriptor objects
"""
def get_notification_channel_descriptor(
self,
request=None,
*,
name: str = None,
retry=None,
timeout=None,
metadata=()
) -> notification_pb2.NotificationChannelDescriptor:
"""
Gets a single notification channel descriptor.
Args:
request: The request object or dict equivalent
name: Required. Descriptor name
retry: Retry configuration
timeout: Request timeout in seconds
metadata: Additional metadata
Returns:
NotificationChannelDescriptor object
"""Represents a notification channel configuration.
class NotificationChannel:
type: str # Channel type (email, sms, slack, etc.)
name: str # Resource name
display_name: str # Human-readable name
description: str # Channel description
labels: Dict[str, str] # Channel-specific configuration
user_labels: Dict[str, str] # User-defined labels
verification_status: NotificationChannel.VerificationStatus # Verification state
enabled: bool # Whether channel is enabled
creation_record: MutationRecord # Creation metadata
mutation_records: List[MutationRecord] # Modification history
class NotificationChannel.VerificationStatus(enum.Enum):
VERIFICATION_STATUS_UNSPECIFIED = 0
UNVERIFIED = 1
VERIFIED = 2Describes a notification channel type and its required configuration.
class NotificationChannelDescriptor:
name: str # Resource name
type: str # Channel type identifier
display_name: str # Human-readable type name
description: str # Type description
labels: List[LabelDescriptor] # Required and optional labels
supported_tiers: List[ServiceTier] # Supported service tiers
launch_stage: LaunchStage # Launch stage of the channel typeclass CreateNotificationChannelRequest:
name: str # Required. Project name
notification_channel: NotificationChannel # Required. Channel to create
class DeleteNotificationChannelRequest:
name: str # Required. Channel name to delete
force: bool # Force deletion even if referenced
class GetNotificationChannelRequest:
name: str # Required. Channel name to retrieve
class ListNotificationChannelsRequest:
name: str # Required. Project name
filter: str # Filter expression
order_by: str # Ordering specification
page_size: int # Maximum results per page
page_token: str # Page token
class ListNotificationChannelsResponse:
notification_channels: List[NotificationChannel] # Channels
next_page_token: str # Next page token
class UpdateNotificationChannelRequest:
update_mask: FieldMask # Fields to update
notification_channel: NotificationChannel # Required. Updated channel
class SendNotificationChannelVerificationCodeRequest:
name: str # Required. Channel name
class GetNotificationChannelVerificationCodeRequest:
name: str # Required. Channel name
expire_time: Timestamp # Code expiration time
class GetNotificationChannelVerificationCodeResponse:
code: str # Verification code
expire_time: Timestamp # Code expiration time
class VerifyNotificationChannelRequest:
name: str # Required. Channel name
code: str # Required. Verification code
class GetNotificationChannelDescriptorRequest:
name: str # Required. Channel descriptor name
class ListNotificationChannelDescriptorsRequest:
name: str # Required. Project name
page_size: int # Maximum results per page
page_token: str # Page token
class ListNotificationChannelDescriptorsResponse:
channel_descriptors: List[NotificationChannelDescriptor] # Channel descriptors
next_page_token: str # Next page tokenfrom google.cloud.monitoring import NotificationChannelServiceClient
from google.api import notification_pb2
client = NotificationChannelServiceClient()
project_name = f"projects/{project_id}"
# Create email notification channel
channel = notification_pb2.NotificationChannel()
channel.type = "email"
channel.display_name = "Engineering Team Email"
channel.description = "Email notifications for engineering team"
# Set email address in labels
channel.labels["email_address"] = "engineering@example.com"
# Add user labels for organization
channel.user_labels["team"] = "engineering"
channel.user_labels["priority"] = "high"
# Create the channel
created_channel = client.create_notification_channel(
name=project_name,
notification_channel=channel
)
print(f"Created notification channel: {created_channel.name}")
print(f"Verification status: {created_channel.verification_status}")# Create Slack notification channel
slack_channel = notification_pb2.NotificationChannel()
slack_channel.type = "slack"
slack_channel.display_name = "Engineering Slack Channel"
slack_channel.description = "Slack notifications for critical alerts"
# Set Slack webhook URL
slack_channel.labels["url"] = "https://hooks.slack.com/services/YOUR/SLACK/WEBHOOK"
created_slack = client.create_notification_channel(
name=project_name,
notification_channel=slack_channel
)
print(f"Created Slack channel: {created_slack.name}")channel_name = f"projects/{project_id}/notificationChannels/{channel_id}"
# Send verification code
client.send_notification_channel_verification_code(name=channel_name)
print("Verification code sent")
# After receiving the code, verify the channel
verification_code = "123456" # Code received via email/SMS
verified_channel = client.verify_notification_channel(
name=channel_name,
code=verification_code
)
print(f"Channel verified: {verified_channel.verification_status}")client = NotificationChannelServiceClient()
project_name = f"projects/{project_id}"
# List all notification channels
print("All notification channels:")
for channel in client.list_notification_channels(name=project_name):
print(f"- {channel.display_name} ({channel.type})")
print(f" Status: {channel.verification_status}")
print(f" Enabled: {channel.enabled}")
# List only email channels
email_filter = 'type="email"'
print("\nEmail channels only:")
for channel in client.list_notification_channels(name=project_name, filter=email_filter):
print(f"- {channel.display_name}: {channel.labels.get('email_address')}")from google.protobuf import field_mask_pb2
# Get existing channel
channel_name = f"projects/{project_id}/notificationChannels/{channel_id}"
channel = client.get_notification_channel(name=channel_name)
# Update channel properties
channel.display_name = "Updated Engineering Team Email"
channel.enabled = True
channel.user_labels["environment"] = "production"
# Create field mask for selective update
update_mask = field_mask_pb2.FieldMask()
update_mask.paths.extend(["display_name", "enabled", "user_labels"])
# Update the channel
updated_channel = client.update_notification_channel(
notification_channel=channel,
update_mask=update_mask
)
print(f"Updated channel: {updated_channel.display_name}")# List all supported notification channel types
print("Supported notification channel types:")
for descriptor in client.list_notification_channel_descriptors(name=project_name):
print(f"- {descriptor.type}: {descriptor.display_name}")
print(f" Description: {descriptor.description}")
print(f" Launch stage: {descriptor.launch_stage}")
# Show required labels
print(" Required labels:")
for label in descriptor.labels:
print(f" - {label.key}: {label.description}")
# Get specific channel type descriptor
email_descriptor_name = f"projects/{project_id}/notificationChannelDescriptors/email"
email_descriptor = client.get_notification_channel_descriptor(name=email_descriptor_name)
print(f"\nEmail channel descriptor: {email_descriptor.display_name}")
for label in email_descriptor.labels:
print(f" {label.key}: {label.description} (required: {label.value_type})")# Create SMS notification channel
sms_channel = notification_pb2.NotificationChannel()
sms_channel.type = "sms"
sms_channel.display_name = "On-call SMS Alerts"
sms_channel.description = "SMS notifications for critical incidents"
# Set phone number
sms_channel.labels["number"] = "+1234567890"
created_sms = client.create_notification_channel(
name=project_name,
notification_channel=sms_channel
)
print(f"Created SMS channel: {created_sms.name}")
# SMS channels typically require verification
client.send_notification_channel_verification_code(name=created_sms.name)
print("SMS verification code sent")channel_name = f"projects/{project_id}/notificationChannels/{channel_id}"
try:
# Regular deletion
client.delete_notification_channel(name=channel_name)
print("Channel deleted successfully")
except Exception as e:
print(f"Could not delete channel: {e}")
# Force deletion even if referenced by alert policies
client.delete_notification_channel(name=channel_name, force=True)
print("Channel force deleted")import asyncio
from google.cloud.monitoring import NotificationChannelServiceAsyncClient
async def manage_notifications():
client = NotificationChannelServiceAsyncClient()
project_name = f"projects/{project_id}"
# List channels asynchronously
async for channel in await client.list_notification_channels(name=project_name):
print(f"Async channel: {channel.display_name}")
# Create channel asynchronously
new_channel = notification_pb2.NotificationChannel()
new_channel.type = "email"
new_channel.display_name = "Async Created Email"
new_channel.labels["email_address"] = "async@example.com"
created = await client.create_notification_channel(
name=project_name,
notification_channel=new_channel
)
print(f"Async created channel: {created.name}")
asyncio.run(manage_notifications())class NotificationChannelServiceClient:
@staticmethod
def notification_channel_path(project: str, notification_channel: str) -> str:
"""Returns a fully-qualified notification_channel string."""
@staticmethod
def notification_channel_descriptor_path(project: str, channel_descriptor: str) -> str:
"""Returns a fully-qualified notification_channel_descriptor string."""
@staticmethod
def parse_notification_channel_path(path: str) -> Dict[str, str]:
"""Parses a notification_channel path into its component segments."""Notification channel operations can raise specific exceptions:
from google.api_core import exceptions
from google.cloud.monitoring import NotificationChannelServiceClient
client = NotificationChannelServiceClient()
try:
channel = client.get_notification_channel(name="invalid/path")
except exceptions.NotFound:
print("Notification channel not found")
except exceptions.InvalidArgument as e:
print(f"Invalid channel configuration: {e}")
except exceptions.PermissionDenied:
print("Insufficient permissions")
except exceptions.FailedPrecondition as e:
print(f"Channel verification required: {e}")Install with Tessl CLI
npx tessl i tessl/pypi-google-cloud-monitoring