tessl/pypi-raven

Legacy Python client for Sentry error monitoring service with framework integration and exception tracking capabilities.

—

Pending

Overview

Eval results

Files

Transport Layer

Name: tessl/pypi-raven
Author: tessl

Raven provides multiple transport implementations for sending events to Sentry servers, supporting different execution environments and performance requirements from synchronous blocking to fully asynchronous.

Capabilities

Base Transport Classes

Foundation classes for all transport implementations.

from raven.transport.base import Transport, AsyncTransport

class Transport:
    def __init__(self, parsed_url, timeout=None):
        """
        Base transport class.

        Parameters:
        - parsed_url: Parsed Sentry DSN URL
        - timeout (int): Request timeout in seconds
        """

    def send(self, url, data, headers):
        """
        Send event data to Sentry.

        Parameters:
        - url (str): Sentry endpoint URL
        - data (str): Encoded event data
        - headers (dict): HTTP headers

        Returns:
        str: Event ID
        """

class AsyncTransport(Transport):
    """Base class for asynchronous transports."""
    
    def async_send(self, url, data, headers, success_cb, failure_cb):
        """
        Send event data asynchronously.

        Parameters:
        - url (str): Sentry endpoint URL
        - data (str): Encoded event data
        - headers (dict): HTTP headers
        - success_cb (callable): Success callback
        - failure_cb (callable): Failure callback
        """

HTTP Transport

Standard synchronous HTTP transport using Python's built-in urllib.

from raven.transport.http import HTTPTransport

class HTTPTransport(Transport):
    scheme = ['sync+http', 'sync+https']
    
    def __init__(self, timeout=5, verify_ssl=True, ca_certs=None):
        """
        Synchronous HTTP transport.

        Parameters:
        - timeout (int): Request timeout in seconds
        - verify_ssl (bool): Verify SSL certificates
        - ca_certs (str): Path to CA certificate bundle
        """

Threaded HTTP Transport

HTTP transport with background thread for non-blocking sends.

from raven.transport.threaded import ThreadedHTTPTransport

class ThreadedHTTPTransport(AsyncTransport, HTTPTransport):
    scheme = ['http', 'https', 'threaded+http', 'threaded+https']
    
    def __init__(self, timeout=5, verify_ssl=True, ca_certs=None):
        """
        Threaded HTTP transport for non-blocking sends.
        Inherits constructor from HTTPTransport.

        Parameters:
        - timeout (int): Request timeout in seconds
        - verify_ssl (bool): Verify SSL certificates
        - ca_certs (str): Path to CA certificate bundle
        """

Requests-Based Transport

HTTP transport using the requests library for enhanced features.

from raven.transport.requests import RequestsHTTPTransport, ThreadedRequestsHTTPTransport

class RequestsHTTPTransport(Transport):
    def __init__(self, parsed_url, timeout=5, verify_ssl=True, ca_certs=None):
        """
        HTTP transport using requests library.

        Parameters:
        - parsed_url: Parsed Sentry DSN URL
        - timeout (int): Request timeout in seconds
        - verify_ssl (bool): Verify SSL certificates
        - ca_certs (str): Path to CA certificate bundle
        """

class ThreadedRequestsHTTPTransport(AsyncTransport):
    """Threaded version of requests-based transport."""

Gevent Transport

Asynchronous transport for gevent-based applications.

from raven.transport.gevent import GeventedHTTPTransport

class GeventedHTTPTransport(AsyncTransport):
    def __init__(self, parsed_url, timeout=5, pool_size=10):
        """
        Gevent-based async HTTP transport.

        Parameters:
        - parsed_url: Parsed Sentry DSN URL
        - timeout (int): Request timeout in seconds
        - pool_size (int): Gevent pool size
        """

Twisted Transport

Asynchronous transport for Twisted-based applications.

from raven.transport.twisted import TwistedHTTPTransport

class TwistedHTTPTransport(AsyncTransport):
    def __init__(self, parsed_url, timeout=5):
        """
        Twisted-based async HTTP transport.

        Parameters:
        - parsed_url: Parsed Sentry DSN URL  
        - timeout (int): Request timeout in seconds
        """

Tornado Transport

Asynchronous transport for Tornado applications.

from raven.transport.tornado import TornadoHTTPTransport

class TornadoHTTPTransport(AsyncTransport):
    def __init__(self, parsed_url, timeout=5):
        """
        Tornado async HTTP transport.

        Parameters:
        - parsed_url: Parsed Sentry DSN URL
        - timeout (int): Request timeout in seconds
        """

Eventlet Transport

Asynchronous transport for eventlet-based applications.

from raven.transport.eventlet import EventletHTTPTransport

class EventletHTTPTransport(AsyncTransport):
    def __init__(self, parsed_url, timeout=5):
        """
        Eventlet-based async HTTP transport.

        Parameters:
        - parsed_url: Parsed Sentry DSN URL
        - timeout (int): Request timeout in seconds
        """

Transport Registry

Management system for transport class registration and selection.

from raven.transport.registry import TransportRegistry, default_transports

class TransportRegistry:
    def __init__(self):
        """Transport registry for managing available transports."""

    def register(self, scheme, transport_class):
        """
        Register transport class for URL scheme.

        Parameters:
        - scheme (str): URL scheme (http, https, etc.)
        - transport_class (type): Transport class
        """

    def get_transport(self, parsed_url, **options):
        """
        Get transport instance for URL.

        Parameters:
        - parsed_url: Parsed Sentry DSN URL
        - **options: Transport configuration options

        Returns:
        Transport: Transport instance
        """

# Default transport registry with all available transports
default_transports = TransportRegistry()

Usage Examples

Custom Transport Configuration

from raven import Client
from raven.transport.threaded import ThreadedHTTPTransport

# Use specific transport with custom settings
transport = ThreadedHTTPTransport(
    timeout=10,
    shutdown_timeout=2
)

client = Client(
    dsn='https://your-dsn@sentry.io/project-id',
    transport=transport
)

Gevent Application

import gevent
from raven import Client
from raven.transport.gevent import GeventedHTTPTransport

# Configure gevent-compatible client
client = Client(
    dsn='https://your-dsn@sentry.io/project-id',
    transport=GeventedHTTPTransport
)

def worker():
    try:
        # This won't block other greenlets
        risky_operation()
    except Exception:
        client.captureException()

# Spawn multiple workers
jobs = [gevent.spawn(worker) for _ in range(10)]
gevent.joinall(jobs)

Twisted Application

from twisted.internet import reactor
from raven import Client
from raven.transport.twisted import TwistedHTTPTransport

client = Client(
    dsn='https://your-dsn@sentry.io/project-id',
    transport=TwistedHTTPTransport
)

def handle_error(failure):
    client.captureException()
    
def async_operation():
    d = some_async_call()
    d.addErrback(handle_error)
    return d

reactor.run()

Tornado Application

import tornado.web
import tornado.ioloop
from raven import Client
from raven.transport.tornado import TornadoHTTPTransport

class MainHandler(tornado.web.RequestHandler):
    def initialize(self):
        self.client = Client(
            dsn='https://your-dsn@sentry.io/project-id',
            transport=TornadoHTTPTransport
        )
    
    async def get(self):
        try:
            result = await async_database_call()
            self.write(result)
        except Exception:
            self.client.captureException()
            self.set_status(500)
            self.write('Error occurred')

app = tornado.web.Application([
    (r'/', MainHandler),
])

if __name__ == '__main__':
    app.listen(8888)
    tornado.ioloop.IOLoop.current().start()

Custom Transport Implementation

from raven.transport.base import Transport
import requests

class CustomHTTPTransport(Transport):
    def __init__(self, parsed_url, timeout=5, retries=3):
        super().__init__(parsed_url, timeout)
        self.retries = retries
        
    def send(self, data, headers):
        url = f"{self.parsed_url.scheme}://{self.parsed_url.netloc}/api/{self.parsed_url.path.strip('/')}/store/"
        
        for attempt in range(self.retries):
            try:
                response = requests.post(
                    url,
                    data=data,
                    headers=headers,
                    timeout=self.timeout
                )
                response.raise_for_status()
                return response.headers.get('X-Sentry-ID')
                
            except requests.RequestException as e:
                if attempt == self.retries - 1:
                    raise
                time.sleep(2 ** attempt)  # Exponential backoff

# Register custom transport
from raven.transport.registry import default_transports
default_transports.register('https', CustomHTTPTransport)

client = Client('https://your-dsn@sentry.io/project-id')

Transport Selection Based on Environment

import os
from raven import Client

def get_transport_class():
    if os.getenv('GEVENT_SUPPORT'):
        from raven.transport.gevent import GeventedHTTPTransport
        return GeventedHTTPTransport
    elif os.getenv('TWISTED_SUPPORT'):
        from raven.transport.twisted import TwistedHTTPTransport
        return TwistedHTTPTransport
    elif os.getenv('ASYNC_SUPPORT'):
        from raven.transport.threaded import ThreadedHTTPTransport
        return ThreadedHTTPTransport
    else:
        from raven.transport.http import HTTPTransport
        return HTTPTransport

client = Client(
    dsn=os.getenv('SENTRY_DSN'),
    transport=get_transport_class()
)

Error Handling in Transports

from raven import Client
from raven.transport.threaded import ThreadedHTTPTransport
import logging

# Configure logging to see transport errors
logging.basicConfig(level=logging.DEBUG)

client = Client(
    dsn='https://your-dsn@sentry.io/project-id',
    transport=ThreadedHTTPTransport,
    # Raise errors instead of swallowing them
    raise_send_errors=True
)

try:
    client.captureMessage('Test message')
except Exception as e:
    print(f"Transport error: {e}")

Install with Tessl CLI