or run

npx @tessl/cli init
Log in

Version

Tile

Overview

Evals

Files

docs

index.md
tile.json

tessl/pypi-dagster-twilio

A Dagster integration for twilio

Workspace
tessl
Visibility
Public
Created
Last updated
Describes
pypipkg:pypi/dagster-twilio@0.27.x

To install, run

npx @tessl/cli install tessl/pypi-dagster-twilio@0.27.0

index.mddocs/

Dagster Twilio

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)