tessl/pypi-dagster-twilio

A Dagster integration for twilio

—

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Securityby

Pending

The risk profile of this skill

Overview

Eval results

Files

Dagster Twilio

Name: tessl/pypi-dagster-twilio
Author: tessl

A Dagster integration for Twilio that enables users to connect to Twilio services within Dagster data pipelines. This package provides both modern Pydantic-based resources and legacy resource functions for integrating Twilio REST API functionality into Dagster workflows.

Package Information

Package Name: dagster-twilio
Language: Python
Installation: pip install dagster-twilio

Core Imports

from dagster_twilio import TwilioResource, twilio_resource

For version information:

from dagster_twilio import __version__

Basic Usage

Modern Resource Approach (Recommended)

from dagster import asset
from dagster_twilio import TwilioResource

@asset
def send_notification(twilio_resource: TwilioResource):
    # Create Twilio client
    client = twilio_resource.create_client()
    
    # Send SMS message
    message = client.messages.create(
        body="Pipeline completed successfully!",
        from_="+1234567890",
        to="+0987654321"
    )
    
    return message.sid

# Configure resource
twilio_resource = TwilioResource(
    account_sid="your_account_sid",
    auth_token="your_auth_token" 
)

Legacy Resource Approach

from dagster import op, job
from dagster_twilio import twilio_resource

@op(required_resource_keys={"twilio_resource"})
def send_sms_op(context):
    client = context.resources.twilio_resource
    
    message = client.messages.create(
        body="Job completed!",
        from_="+1234567890", 
        to="+0987654321"
    )
    
    return message.sid

@job(resource_defs={"twilio_resource": twilio_resource})
def notification_job():
    send_sms_op()

Capabilities

TwilioResource

Modern Pydantic-based resource class for Twilio integration that provides secure configuration management and creates Twilio REST API clients.

class TwilioResource(ConfigurableResource):
    """This resource is for connecting to Twilio."""
    
    account_sid: str = Field(
        description="Twilio Account SID, created with your Twilio account. This can be found on your Twilio dashboard, see https://www.twilio.com/blog/twilio-access-tokens-python"
    )
    auth_token: str = Field(
        description="Twilio Authentication Token, created with your Twilio account. This can be found on your Twilio dashboard, see https://www.twilio.com/blog/twilio-access-tokens-python"  
    )
    
    def create_client(self) -> Client:
        """Create and return a Twilio REST API client."""
        
    @classmethod
    def _is_dagster_maintained(cls) -> bool:
        """Indicates this is a Dagster-maintained resource."""

Configuration Fields:

account_sid (str): Twilio Account SID from your Twilio dashboard. See Twilio Access Tokens documentation for details.
auth_token (str): Twilio Authentication Token from your Twilio dashboard. See Twilio Access Tokens documentation for details.

Methods:

create_client(): Creates and returns a twilio.rest.Client instance configured with the provided credentials
configure_at_launch(): Inherited method that allows runtime configuration of the resource

twilio_resource

Legacy resource function that provides Twilio integration using the traditional Dagster resource pattern.

@dagster_maintained_resource
@resource(
    config_schema=TwilioResource.to_config_schema(),
    description="This resource is for connecting to Twilio"
)
def twilio_resource(context: InitResourceContext) -> Client:
    """
    Legacy resource function for Twilio integration.
    
    Parameters:
    - context: InitResourceContext with Twilio configuration
    
    Returns:
    twilio.rest.Client: Configured Twilio REST API client
    """

Configuration Schema: Uses the same configuration schema as TwilioResource:

account_sid (str): Twilio Account SID
auth_token (str): Twilio Authentication Token

Twilio Client Operations

Both resource approaches return a twilio.rest.Client instance that provides access to all Twilio REST API functionality, including:

SMS Messages: Send and manage SMS messages via client.messages.create()
Voice Calls: Make voice calls via client.calls.create()
WhatsApp: Send WhatsApp messages
Email: Send emails via SendGrid integration
Verify: Phone number verification services
Lookup: Phone number information lookup

Common Twilio client operations:

# Send SMS
message = client.messages.create(
    body="Your message here",
    from_="+1234567890",  # Your Twilio phone number
    to="+0987654321"      # Recipient phone number
)

# Make voice call
call = client.calls.create(
    url="http://demo.twilio.com/docs/voice.xml",  # TwiML URL
    to="+0987654321",
    from_="+1234567890"
)

# Lookup phone number information
phone_number = client.lookups.phone_numbers("+15558675310").fetch()

Types

# From dagster
from dagster import ConfigurableResource, InitResourceContext

# From pydantic
from pydantic import Field

# From twilio
from twilio.rest import Client
from twilio.base.exceptions import TwilioRestException

class TwilioResource(ConfigurableResource):
    account_sid: str = Field(description="...")
    auth_token: str = Field(description="...")
    
    def create_client(self) -> Client: ...
    
    @classmethod 
    def _is_dagster_maintained(cls) -> bool: ...

# Package version
__version__: str  # Currently "0.27.9"

Error Handling

When working with Twilio operations, handle TwilioRestException for API-related errors:

from twilio.base.exceptions import TwilioRestException

try:
    message = client.messages.create(
        body="Test message",
        from_="+15005550006",  # Valid test number
        to="+15005550001"      # Invalid test number
    )
except TwilioRestException as e:
    context.log.error(f"Twilio API error: {e}")
    raise

Configuration Best Practices

Environment Variables

Store Twilio credentials securely using environment variables:

import os
from dagster_twilio import TwilioResource

twilio_resource = TwilioResource(
    account_sid=os.getenv("TWILIO_ACCOUNT_SID"),
    auth_token=os.getenv("TWILIO_AUTH_TOKEN")
)

Dagster Configuration

Configure via Dagster's configuration system:

# dagster.yaml
resources:
  twilio_resource:
    config:
      account_sid: 
        env: TWILIO_ACCOUNT_SID
      auth_token:
        env: TWILIO_AUTH_TOKEN

Test Credentials

For testing, use Twilio's test credentials:

Test Account SID: Starts with AC followed by test identifier
Test Auth Token: Test authentication token
Test phone numbers: +15005550006 (valid), +15005550001 (invalid)

Workspace: tessl
Visibility: Public
Created: 7 months ago
Last updated: 3 months ago
Describes: pkg:pypi/dagster-twilio@0.27.x
Publish Source: CLI
Badge