Pyjo.UserAgent - Non-blocking I/O HTTP and WebSocket user agent¶

from Pyjo.UserAgent import UserAgent

# Fine grained response handling (raises on connection errors)
ua = UserAgent()
res = ua.get('docs.mojolicious.org').result
if res.is_success:
    print(res.body)
elif res.is_error:
    print(res.message)
elif res.code == 301:
    print(res.headers.location)
else:
    print('Whatever...')

# Or import the module, as in Google Python Style Guide

import Pyjo.IOLoop
import Pyjo.Promise
import Pyjo.URL
import Pyjo.UserAgent

# Fine grained response handling (raises on connection errors)
ua = Pyjo.UserAgent.new()
res = ua.get('docs.mojolicious.org').result
if res.is_success:
    print(res.body)
elif res.is_error:
    print(res.message)
elif res.code == 301:
    print(res.headers.location)
else:
    print('Whatever...')

# Say hello to the Unicode snowman and include an Accept header
print(ua.get('www.☃.net?hello=there', {'Accept': '*/*'}).result.body)

# Extract data from HTML and XML resources with CSS selectors
print(ua.get('www.perl.org').result.dom().at('title').text)

# Scrape the latest headlines from a news site
print(ua.get('blogs.perl.org').result.dom().find('h2 > a').map('text').join('\n'))

# IPv6 PUT request with Content-Type header and content
tx = ua.put('[::1]:3000', {'Content-Type': 'text/plain'}, b'Hi!')

# Quick JSON API request with Basic authentication
url = Pyjo.URL.new('https://example.com/test.json')
url.userinfo = 'sri:☃'
value = ua.get(url).result.json()

# JSON POST (application/json) with TLS certificate authentication
ua.cert, ua.key = 'tls.crt', 'tls.key'
tx = ua.post('https://example.com', json={'top': 'secret'})

# Form POST (application/x-www-form-urlencoded)
tx = ua.post('https://metacpan.org/search', form={'q': 'mojo'})

# Search DuckDuckGo anonymously through Tor
ua.proxy.http = 'socks://127.0.0.1:9050'
print(ua.get('api.3g2upl4pq6kufc4m.onion/?q=mojolicious&format=json').result.json('/Abstract'))

# GET request via UNIX domain socket "/tmp/myapp.sock" (percent encoded slash)
print(ua.get('http+unix://%2Ftmp%2Fmyapp.sock/test').result.body)

# Follow redirects to download Mojolicious from GitHub
ua.max_redirects = 5
ua.get('https://www.github.com/mojolicious/mojo/tarball/main').result.save_to('/home/sri/mojo.tar.gz')

# Non-blocking request
ua.get('mojolicious.org', lambda ua, tx: print(tx.result.dom().at('title').text))
if not Pyjo.IOLoop.is_running():
    Pyjo.IOLoop.start()

# Concurrent non-blocking requests (synchronized with promises)
mojo_promise = ua.get_p('mojolicious.org')
cpan_promise = ua.get_p('cpan.org')


@Pyjo.Promise.all(mojo_promise, cpan_promise).then
def titles(mojo, cpan):
    print(mojo[0].result.dom().at('title').text)
    print(cpan[0].result.dom().at('title').text)


titles.wait()


# WebSocket connection sending and receiving JSON via UNIX domain socket
@ua.websocket('ws+unix://%2Ftmp%2Fmyapp.sock/echo.json')
def echo(ua, tx):
    if not tx.is_websocket:
        return print('WebSocket handshake failed!')

    @tx.on('json')
    def json(tx, hash):
        print(f'WebSocket message via JSON: {hash["msg"]}')
        tx.finish()

    tx.send({'json': {'msg': 'Hello World!'}})


if not Pyjo.IOLoop.is_running():
    Pyjo.IOLoop.start()

Pyjo.UserAgent is a full featured non-blocking I/O HTTP and WebSocket user agent, with IPv6, TLS, SNI, IDNA, HTTP/SOCKS5 proxy, UNIX domain socket, Comet (long polling), Promises/A+, keep-alive, connection pooling, timeout, cookie, multipart, gzip compression and multiple event loop support.

All connections will be reset automatically if a new process has been forked, this allows multiple processes to share the same Pyjo.UserAgent object safely.

Non-blocking name resolution, SOCKS5 and TLS support come with Pyjo.IOLoop and the standard library. Individual features can be disabled with the PYJO_NO_NNR, PYJO_NO_SOCKS and PYJO_NO_TLS environment variables.

The end of a with block closes the connections and the server that the user agent made.

Events¶

Pyjo.UserAgent inherits all events from Pyjo.EventEmitter and can emit the following new ones.

prepare¶

ua.on('prepare', lambda ua, tx: ...)

Emitted whenever a new transaction is being prepared, before relative URLs are rewritten and cookies added. This includes automatically prepared proxy CONNECT requests and followed redirects.

@ua.on('prepare')
def prepare(ua, tx):
    if tx.req.url.host == 'mojolicious.org':
        tx.req.url = Pyjo.URL.new('/mock-mojolicious')

start¶

ua.on('start', lambda ua, tx: ...)

Emitted whenever a new transaction is about to start. This includes automatically prepared proxy CONNECT requests and followed redirects.

@ua.on('start')
def start(ua, tx):
    tx.req.headers.header('X-Bender', 'Bite my shiny metal ass!')

Debugging¶

You can set the PYJO_CLIENT_DEBUG environment variable to get some advanced diagnostics information printed to sys.stderr.

PYJO_CLIENT_DEBUG=1

Classes¶

Pyjo.UserAgent inherits all methods from Pyjo.EventEmitter and implements the following new ones.

build_tx(*args: Any, **generator: Any) → Pyjo_Transaction_HTTP[source]¶

tx = ua.build_tx('GET', 'example.com')
tx = ua.build_tx('PUT', 'http://example.com', {'Accept': '*/*'}, b'Content!')
tx = ua.build_tx('PUT', 'http://example.com', {'Accept': '*/*'}, form={'a': 'b'})
tx = ua.build_tx('PUT', 'http://example.com', {'Accept': '*/*'}, json={'a': 'b'})

Generate Pyjo.Transaction.HTTP object with Pyjo.UserAgent.Transactor.Pyjo_UserAgent_Transactor.tx().

# Request with custom cookie
tx = ua.build_tx('GET', 'https://example.com/account')
tx.req.cookies({'name': 'user', 'value': 'sri'})
tx = ua.start(tx)

# Deactivate gzip compression
tx = ua.build_tx('GET', 'example.com')
tx.req.headers.remove('Accept-Encoding')
tx = ua.start(tx)

# Interrupt response by raising an error
tx = ua.build_tx('GET', 'http://example.com')

@tx.res.on('progress')
def progress(res):
    if (server := res.headers.server) and 'IIS' in server:
        res.error = {'message': 'Oh noes, it is IIS!'}

tx = ua.start(tx)

build_websocket_tx(*args: Any) → Pyjo_Transaction_HTTP[source]¶

tx = ua.build_websocket_tx('ws://example.com')
tx = ua.build_websocket_tx('ws://example.com', {'DNT': 1}, ['v1.proto'])

Generate Pyjo.Transaction.HTTP object with Pyjo.UserAgent.Transactor.Pyjo_UserAgent_Transactor.websocket().

# Custom WebSocket handshake with cookie
tx = ua.build_websocket_tx('wss://example.com/echo')
tx.req.cookies({'name': 'user', 'value': 'sri'})

def echo(ua, tx):
    if not tx.is_websocket:
        return print('WebSocket handshake failed!')

    @tx.on('message')
    def message(tx, msg):
        print(f'WebSocket message: {msg}')
        tx.finish()

    tx.send('Hi!')

ua.start(tx, echo)
if not Pyjo.IOLoop.is_running():
    Pyjo.IOLoop.start()

ca¶

ca = ua.ca
ua.ca = '/etc/tls/ca.crt'

Path to TLS certificate authority file used to verify the peer certificate, defaults to the value of the PYJO_CA_FILE environment variable.

cert¶

cert = ua.cert
ua.cert = '/etc/tls/client.crt'

Path to TLS certificate file, defaults to the value of the PYJO_CERT_FILE environment variable.

close() → None[source]¶

ua.close()

Close the connections of the user agent, and the server that it made for relative URLs, as the end of a with block does. The user agent makes new ones when it is used again.

connect_timeout: float¶

timeout = ua.connect_timeout
ua.connect_timeout = 5

Maximum amount of time in seconds establishing a connection may take before getting canceled, defaults to the value of the PYJO_CONNECT_TIMEOUT environment variable or 10.

cookie_jar¶

cookie_jar = ua.cookie_jar
ua.cookie_jar = Pyjo.UserAgent.CookieJar.new()

Cookie jar to use for requests performed by this user agent, defaults to a Pyjo.UserAgent.CookieJar object.

# Ignore all cookies
ua.cookie_jar.ignore = lambda cookie: True

# Add custom cookie to the jar
ua.cookie_jar.add(
    Pyjo.Cookie.Response.new(
        name='foo', value='bar', domain='docs.mojolicious.org', path='/Mojolicious'
    )
)

delete¶

tx = ua.delete('example.com')
tx = ua.delete('http://example.com', {'Accept': '*/*'}, b'Content!')
tx = ua.delete('http://example.com', {'Accept': '*/*'}, form={'a': 'b'})
tx = ua.delete('http://example.com', {'Accept': '*/*'}, json={'a': 'b'})

Perform blocking DELETE request and return resulting Pyjo.Transaction.HTTP object, takes the same arguments as Pyjo.UserAgent.Transactor.Pyjo_UserAgent_Transactor.tx() (except for the DELETE method, which is implied). You can also append a callback, or pass it as cb, to perform requests non-blocking.

ua.delete('http://example.com', lambda ua, tx: print(tx.result.body), json={'a': 'b'})
ua.delete('http://example.com', json={'a': 'b'}, cb=lambda ua, tx: print(tx.result.body))
if not Pyjo.IOLoop.is_running():
    Pyjo.IOLoop.start()

delete_p¶

promise = ua.delete_p('http://example.com')

Same as delete, but performs all requests non-blocking and returns a Pyjo.Promise object instead of accepting a callback.

@ua.delete_p('http://example.com', json={'a': 'b'}).then
def ok(tx):
    print(tx.result.body)

ok.catch(lambda err: warn(f'Connection error: {err}')).wait()

get¶

tx = ua.get('example.com')
tx = ua.get('http://example.com', {'Accept': '*/*'}, b'Content!')
tx = ua.get('http://example.com', {'Accept': '*/*'}, form={'a': 'b'})
tx = ua.get('http://example.com', {'Accept': '*/*'}, json={'a': 'b'})

Perform blocking GET request and return resulting Pyjo.Transaction.HTTP object, takes the same arguments as Pyjo.UserAgent.Transactor.Pyjo_UserAgent_Transactor.tx() (except for the GET method, which is implied). You can also append a callback, or pass it as cb, to perform requests non-blocking.

ua.get('http://example.com', lambda ua, tx: print(tx.result.body), json={'a': 'b'})
ua.get('http://example.com', json={'a': 'b'}, cb=lambda ua, tx: print(tx.result.body))
if not Pyjo.IOLoop.is_running():
    Pyjo.IOLoop.start()

get_p¶

promise = ua.get_p('http://example.com')

Same as get, but performs all requests non-blocking and returns a Pyjo.Promise object instead of accepting a callback.

@ua.get_p('http://example.com', json={'a': 'b'}).then
def ok(tx):
    print(tx.result.body)

ok.catch(lambda err: warn(f'Connection error: {err}')).wait()

head¶

tx = ua.head('example.com')
tx = ua.head('http://example.com', {'Accept': '*/*'}, b'Content!')
tx = ua.head('http://example.com', {'Accept': '*/*'}, form={'a': 'b'})
tx = ua.head('http://example.com', {'Accept': '*/*'}, json={'a': 'b'})

Perform blocking HEAD request and return resulting Pyjo.Transaction.HTTP object, takes the same arguments as Pyjo.UserAgent.Transactor.Pyjo_UserAgent_Transactor.tx() (except for the HEAD method, which is implied). You can also append a callback, or pass it as cb, to perform requests non-blocking.

ua.head('http://example.com', lambda ua, tx: print(tx.result.body), json={'a': 'b'})
ua.head('http://example.com', json={'a': 'b'}, cb=lambda ua, tx: print(tx.result.body))
if not Pyjo.IOLoop.is_running():
    Pyjo.IOLoop.start()

head_p¶

promise = ua.head_p('http://example.com')

Same as head, but performs all requests non-blocking and returns a Pyjo.Promise object instead of accepting a callback.

@ua.head_p('http://example.com', json={'a': 'b'}).then
def ok(tx):
    print(tx.result.body)

ok.catch(lambda err: warn(f'Connection error: {err}')).wait()

inactivity_timeout: float¶

timeout = ua.inactivity_timeout
ua.inactivity_timeout = 15

Maximum amount of time in seconds a connection can be inactive before getting closed, defaults to the value of the PYJO_INACTIVITY_TIMEOUT environment variable or 40. Setting the value to 0 will allow connections to be inactive indefinitely.

insecure¶

boolean = ua.insecure
ua.insecure = boolean

Do not require a valid TLS certificate to access HTTPS/WSS sites, defaults to the value of the PYJO_INSECURE environment variable.

# Disable TLS certificate verification for testing
ua.insecure = True
print(ua.get('https://127.0.0.1:3000').result.code)

property ioloop: Pyjo_IOLoop[source]¶

loop = ua.ioloop
ua.ioloop = Pyjo.IOLoop.new()

Event loop object to use for blocking I/O operations, defaults to a Pyjo.IOLoop object.

key¶

key = ua.key
ua.key = '/etc/tls/client.crt'

Path to TLS key file, defaults to the value of the PYJO_KEY_FILE environment variable.

max_connections¶

max = ua.max_connections
ua.max_connections = 5

Maximum number of keep-alive connections that the user agent will retain before it starts closing the oldest ones, defaults to 5. Setting the value to 0 will prevent any connections from being kept alive.

max_redirects: int¶

max = ua.max_redirects
ua.max_redirects = 3

Maximum number of redirects the user agent will follow before it fails, defaults to the value of the PYJO_MAX_REDIRECTS environment variable or 0.

max_response_size¶

max = ua.max_response_size
ua.max_response_size = 16777216

Maximum response size in bytes, defaults to the value of Pyjo.Message.Response.Pyjo_Message_Response.max_message_size. Setting the value to 0 will allow responses of indefinite size. Note that increasing this value can also drastically increase memory usage, should you for example attempt to parse an excessively large response body with the methods dom or json of Pyjo.Message.

options¶

tx = ua.options('example.com')
tx = ua.options('http://example.com', {'Accept': '*/*'}, b'Content!')
tx = ua.options('http://example.com', {'Accept': '*/*'}, form={'a': 'b'})
tx = ua.options('http://example.com', {'Accept': '*/*'}, json={'a': 'b'})

Perform blocking OPTIONS request and return resulting Pyjo.Transaction.HTTP object, takes the same arguments as Pyjo.UserAgent.Transactor.Pyjo_UserAgent_Transactor.tx() (except for the OPTIONS method, which is implied). You can also append a callback, or pass it as cb, to perform requests non-blocking.

ua.options('http://example.com', lambda ua, tx: print(tx.result.body), json={'a': 'b'})
ua.options('http://example.com', json={'a': 'b'}, cb=lambda ua, tx: print(tx.result.body))
if not Pyjo.IOLoop.is_running():
    Pyjo.IOLoop.start()

options_p¶

promise = ua.options_p('http://example.com')

Same as options, but performs all requests non-blocking and returns a Pyjo.Promise object instead of accepting a callback.

@ua.options_p('http://example.com', json={'a': 'b'}).then
def ok(tx):
    print(tx.result.body)

ok.catch(lambda err: warn(f'Connection error: {err}')).wait()

patch¶

tx = ua.patch('example.com')
tx = ua.patch('http://example.com', {'Accept': '*/*'}, b'Content!')
tx = ua.patch('http://example.com', {'Accept': '*/*'}, form={'a': 'b'})
tx = ua.patch('http://example.com', {'Accept': '*/*'}, json={'a': 'b'})

Perform blocking PATCH request and return resulting Pyjo.Transaction.HTTP object, takes the same arguments as Pyjo.UserAgent.Transactor.Pyjo_UserAgent_Transactor.tx() (except for the PATCH method, which is implied). You can also append a callback, or pass it as cb, to perform requests non-blocking.

ua.patch('http://example.com', lambda ua, tx: print(tx.result.body), json={'a': 'b'})
ua.patch('http://example.com', json={'a': 'b'}, cb=lambda ua, tx: print(tx.result.body))
if not Pyjo.IOLoop.is_running():
    Pyjo.IOLoop.start()

patch_p¶

promise = ua.patch_p('http://example.com')

Same as patch, but performs all requests non-blocking and returns a Pyjo.Promise object instead of accepting a callback.

@ua.patch_p('http://example.com', json={'a': 'b'}).then
def ok(tx):
    print(tx.result.body)

ok.catch(lambda err: warn(f'Connection error: {err}')).wait()

post¶

tx = ua.post('example.com')
tx = ua.post('http://example.com', {'Accept': '*/*'}, b'Content!')
tx = ua.post('http://example.com', {'Accept': '*/*'}, form={'a': 'b'})
tx = ua.post('http://example.com', {'Accept': '*/*'}, json={'a': 'b'})

Perform blocking POST request and return resulting Pyjo.Transaction.HTTP object, takes the same arguments as Pyjo.UserAgent.Transactor.Pyjo_UserAgent_Transactor.tx() (except for the POST method, which is implied). You can also append a callback, or pass it as cb, to perform requests non-blocking.

ua.post('http://example.com', lambda ua, tx: print(tx.result.body), json={'a': 'b'})
ua.post('http://example.com', json={'a': 'b'}, cb=lambda ua, tx: print(tx.result.body))
if not Pyjo.IOLoop.is_running():
    Pyjo.IOLoop.start()

post_p¶

promise = ua.post_p('http://example.com')

Same as post, but performs all requests non-blocking and returns a Pyjo.Promise object instead of accepting a callback.

@ua.post_p('http://example.com', json={'a': 'b'}).then
def ok(tx):
    print(tx.result.body)

ok.catch(lambda err: warn(f'Connection error: {err}')).wait()

proxy¶

proxy = ua.proxy
ua.proxy = Pyjo.UserAgent.Proxy.new()

Proxy manager, defaults to a Pyjo.UserAgent.Proxy object.

# Detect proxy servers from environment
ua.proxy.detect()

# Manually configure HTTP proxy (using CONNECT for HTTPS/WebSockets)
ua.proxy.http = ua.proxy.https = 'http://127.0.0.1:8080'

# Manually configure Tor (SOCKS5)
ua.proxy.http = ua.proxy.https = 'socks://127.0.0.1:9050'

# Manually configure UNIX domain socket (using CONNECT for HTTPS/WebSockets)
ua.proxy.http = ua.proxy.https = 'http+unix://%2Ftmp%2Fproxy.sock'

put¶

tx = ua.put('example.com')
tx = ua.put('http://example.com', {'Accept': '*/*'}, b'Content!')
tx = ua.put('http://example.com', {'Accept': '*/*'}, form={'a': 'b'})
tx = ua.put('http://example.com', {'Accept': '*/*'}, json={'a': 'b'})

Perform blocking PUT request and return resulting Pyjo.Transaction.HTTP object, takes the same arguments as Pyjo.UserAgent.Transactor.Pyjo_UserAgent_Transactor.tx() (except for the PUT method, which is implied). You can also append a callback, or pass it as cb, to perform requests non-blocking.

ua.put('http://example.com', lambda ua, tx: print(tx.result.body), json={'a': 'b'})
ua.put('http://example.com', json={'a': 'b'}, cb=lambda ua, tx: print(tx.result.body))
if not Pyjo.IOLoop.is_running():
    Pyjo.IOLoop.start()

put_p¶

promise = ua.put_p('http://example.com')

Same as put, but performs all requests non-blocking and returns a Pyjo.Promise object instead of accepting a callback.

@ua.put_p('http://example.com', json={'a': 'b'}).then
def ok(tx):
    print(tx.result.body)

ok.catch(lambda err: warn(f'Connection error: {err}')).wait()

query¶

tx = ua.query('example.com')
tx = ua.query('http://example.com', {'Accept': '*/*'}, b'Content!')
tx = ua.query('http://example.com', {'Accept': '*/*'}, form={'a': 'b'})
tx = ua.query('http://example.com', {'Accept': '*/*'}, json={'a': 'b'})

Perform blocking QUERY request and return resulting Pyjo.Transaction.HTTP object, takes the same arguments as Pyjo.UserAgent.Transactor.Pyjo_UserAgent_Transactor.tx() (except for the QUERY method, which is implied). You can also append a callback, or pass it as cb, to perform requests non-blocking.

ua.query('http://example.com', lambda ua, tx: print(tx.result.body), json={'a': 'b'})
ua.query('http://example.com', json={'a': 'b'}, cb=lambda ua, tx: print(tx.result.body))
if not Pyjo.IOLoop.is_running():
    Pyjo.IOLoop.start()

query_p¶

promise = ua.query_p('http://example.com')

Same as query, but performs all requests non-blocking and returns a Pyjo.Promise object instead of accepting a callback.

@ua.query_p('http://example.com', json={'a': 'b'}).then
def ok(tx):
    print(tx.result.body)

ok.catch(lambda err: warn(f'Connection error: {err}')).wait()

request_timeout: float¶

timeout = ua.request_timeout
ua.request_timeout = 5

Maximum amount of time in seconds establishing a connection, sending the request and receiving a whole response may take before getting canceled, defaults to the value of the PYJO_REQUEST_TIMEOUT environment variable or 0. Setting the value to 0 will allow the user agent to wait indefinitely. The timeout will reset for every followed redirect.

# Total limit of 5 seconds, of which 3 seconds may be spent connecting
ua.max_redirects, ua.connect_timeout, ua.request_timeout = 0, 3, 5

property server: Pyjo_UserAgent_Server[source]¶

server = ua.server
ua.server = Pyjo.UserAgent.Server.new()

Application server relative URLs will be processed with, defaults to a Pyjo.UserAgent.Server object.

# Mock web service
ua.server.app = Pyjoyment.new()
ua.server.app.routes.get('/time', lambda c: c.render(json={'now': time.time()}))
now = ua.get('/time').result.json()['now']

# Change log level
ua.server.app.log.level = 'fatal'

# Port currently used for processing relative URLs blocking
print(ua.server.url().port)

# Port currently used for processing relative URLs non-blocking
print(ua.server.nb_url().port)

source_address¶

address = ua.source_address
ua.source_address = ('127.0.0.1', 0)

Local address and port to connect from when opening new connections, as socket.create_connection() takes it.

start(tx: Pyjo_Transaction_HTTP) → Any[source]¶

start(tx: Pyjo_Transaction_HTTP, cb: Callback) → str | None

start(tx: Pyjo_Transaction_HTTP, cb: None) → Any

tx = ua.start(Pyjo.Transaction.HTTP.new())

Perform blocking request for a custom Pyjo.Transaction.HTTP object, which can be prepared manually or with build_tx(). You can also append a callback to perform requests non-blocking.

tx = ua.build_tx('GET', 'http://example.com')
ua.start(tx, lambda ua, tx: print(tx.result.body))
if not Pyjo.IOLoop.is_running():
    Pyjo.IOLoop.start()

A blocking request cannot wait for its response in an event loop of asyncio that runs already, as in a coroutine or in an action of an application on Pyjo.Reactor.Asyncio, as asyncio does not nest event loops: it warns with RuntimeWarning and returns the transaction without the response. Await the promise of start_p(), or of a method like get_p, there.

start_p(tx: Pyjo_Transaction_HTTP) → Pyjo_Promise[source]¶

promise = ua.start_p(Pyjo.Transaction.HTTP.new())

Same as start(), but performs all requests non-blocking and returns a Pyjo.Promise object instead of accepting a callback.

tx = ua.build_tx('GET', 'http://example.com')

@ua.start_p(tx).then
def ok(tx):
    print(tx.result.body)

ok.catch(lambda err: warn(f'Connection error: {err}')).wait()

tls_context¶

context = ua.tls_context
ua.tls_context = ssl.create_default_context()

The ssl.SSLContext to use for TLS connections instead of the one that ca, cert and key make.

transactor¶

t = ua.transactor
ua.transactor = Pyjo.UserAgent.Transactor.new()

Transaction builder, defaults to a Pyjo.UserAgent.Transactor object.

# Change name of user agent
ua.transactor.name = 'MyUA 1.0'

# Disable compression
ua.transactor.compressed = False

websocket(*args: Any, cb: Callback | None = None) → Any[source]¶

ua.websocket('ws://example.com', lambda ua, tx: ...)
ua.websocket('ws://example.com', {'DNT': 1}, ['v1.proto'], lambda ua, tx: ...)
ua.websocket('ws://example.com', cb=lambda ua, tx: ...)

Open a non-blocking WebSocket connection with transparent handshake, takes the same arguments as Pyjo.UserAgent.Transactor.Pyjo_UserAgent_Transactor.websocket(). The callback will receive either a Pyjo.Transaction.WebSocket or Pyjo.Transaction.HTTP object, depending on if the handshake was successful. Without it the method is a decorator, and the decorated name is the id of the connection.

@ua.websocket('wss://example.com/echo', ['v1.proto'])
def echo(ua, tx):
    if not tx.is_websocket:
        return print('WebSocket handshake failed!')
    if not tx.protocol:
        return print('Subprotocol negotiation failed!')
    tx.on('finish', lambda tx, code, reason: print(f'WebSocket closed with status {code}.'))

    @tx.on('message')
    def message(tx, msg):
        print(f'WebSocket message: {msg}')
        tx.finish()

    tx.send('Hi!')

if not Pyjo.IOLoop.is_running():
    Pyjo.IOLoop.start()

You can activate permessage-deflate compression by setting the Sec-WebSocket-Extensions header, this can result in much better performance, but also increases memory usage by up to 300KiB per connection.

ua.websocket('ws://example.com/foo', {'Sec-WebSocket-Extensions': 'permessage-deflate'}, lambda ua, tx: ...)

websocket_p(*args: Any) → Pyjo_Promise[source]¶

promise = ua.websocket_p('ws://example.com')

Same as websocket(), but returns a Pyjo.Promise object instead of accepting a callback.

@ua.websocket_p('wss://example.com/echo').then
def ok(tx):
    promise = Pyjo.Promise.new()
    tx.on('finish', lambda tx, code, reason: promise.resolve())

    @tx.on('message')
    def message(tx, msg):
        print(f'WebSocket message: {msg}')
        tx.finish()

    tx.send('Hi!')
    return promise

ok.catch(lambda err: warn(f'WebSocket error: {err}')).wait()

Pyjo.UserAgent - Non-blocking I/O HTTP and WebSocket user agent¶

Events¶

prepare¶

start¶

Debugging¶

See also¶

Classes¶

Pyjoyment

Navigation

Related Topics