FM / transport

/opt/cloudlinux/venv/lib/python3.11/site-packages/raven/transport/

/opt/cloudlinux/venv/lib/python3.11/site-packages/raven/transport

Name	Size	Mode	Actions
__pycache__/	-	0755	rm
base.py	1219	0644	edit dl rm
eventlet.py	1535	0644	edit dl rm
exceptions.py	517	0644	edit dl rm
gevent.py	1658	0644	edit dl rm
http.py	1819	0644	edit dl rm
registry.py	2606	0644	edit dl rm
requests.py	1024	0644	edit dl rm
threaded.py	5102	0644	edit dl rm
threaded_requests.py	1094	0644	edit dl rm
tornado.py	1833	0644	edit dl rm
twisted.py	2322	0644	edit dl rm
__init__.py	784	0644	edit dl rm

Edit: /opt/cloudlinux/venv/lib/python3.11/site-packages/raven/transport/threaded_requests.py (1094B)

"""
raven.transport.threaded_requests
~~~~~~~~~~~~~~~~~~~~~~~~

:copyright: (c) 2010-2012 by the Sentry Team, see AUTHORS for more details.
:license: BSD, see LICENSE for more details.
"""
from __future__ import absolute_import

from raven.transport.base import AsyncTransport
from raven.transport import RequestsHTTPTransport
from raven.transport.threaded import AsyncWorker

class ThreadedRequestsHTTPTransport(AsyncTransport, RequestsHTTPTransport):

scheme = ['threaded+requests+http', 'threaded+requests+https']

def get_worker(self):
        if not hasattr(self, '_worker'):
            self._worker = AsyncWorker()
        return self._worker

def send_sync(self, url, data, headers, success_cb, failure_cb):
        try:
            super(ThreadedRequestsHTTPTransport, self).send(url, data, headers)
        except Exception as e:
            failure_cb(e)
        else:
            success_cb()

def async_send(self, url, data, headers, success_cb, failure_cb):
        self.get_worker().queue(
            self.send_sync, url, data, headers, success_cb, failure_cb)

""" raven.transport.threaded_requests ~~~~~~~~~~~~~~~~~~~~~~~~ :copyright: (c) 2010-2012 by the Sentry Team, see AUTHORS for more details. :license: BSD, see LICENSE for more details. """ from __future__ import absolute_import from raven.transport.base import AsyncTransport from raven.transport import RequestsHTTPTransport from raven.transport.threaded import AsyncWorker class ThreadedRequestsHTTPTransport(AsyncTransport, RequestsHTTPTransport): scheme = ['threaded+requests+http', 'threaded+requests+https'] def get_worker(self): if not hasattr(self, '_worker'): self._worker = AsyncWorker() return self._worker def send_sync(self, url, data, headers, success_cb, failure_cb): try: super(ThreadedRequestsHTTPTransport, self).send(url, data, headers) except Exception as e: failure_cb(e) else: success_cb() def async_send(self, url, data, headers, success_cb, failure_cb): self.get_worker().queue( self.send_sync, url, data, headers, success_cb, failure_cb)