Pyjo.Content - HTTP content base class

from Pyjo.Content import Content

class MyContent(Content):
    def body_contains(self, chunk): ...
    @property
    def body_size(self): ...
    def get_body_chunk(self, offset): ...

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

import Pyjo.Content

class MyContent(Pyjo.Content.object):
    def body_contains(self, chunk): ...
    @property
    def body_size(self): ...
    def get_body_chunk(self, offset): ...

Pyjo.Content is an abstract base class for HTTP content containers, based on RFC 7230 and RFC 7231, like Pyjo.Content.MultiPart and Pyjo.Content.Single.

Events

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

body

content.on('body', lambda content: ...)

Emitted once all headers have been parsed and the body starts.

@content.on('body')
def body(content):
    if content.headers.header('X-No-MultiPart'):
        content.auto_upgrade = False

drain

content.on('drain', lambda content, offset: ...)

Emitted once all data has been written.

content.on('drain', lambda content, offset: content.write_chunk(str(time.time()).encode()))

read

content.on('read', lambda content, chunk: ...)

Emitted when a new chunk of content arrives.

content.on('read', lambda content, chunk: print(f'Streaming: {chunk}'))

sse

content.on('sse', lambda content, event: ...)

Emitted when a new Server-Sent Event (SSE) connection has been established, with None for the event, and for each new event that arrives. Note that this event is EXPERIMENTAL and may change without warning!

@content.on('sse')
def sse(content, event):
    if event:
        print(f'Type: {event["type"]}, Data: {event["text"]}')
    else:
        print('SSE connection established')

See also

Pyjoyment, https://mojolicious.org.

Ported from Mojo::Content of Mojolicious 9.49.

class Pyjo.Content.Pyjo_Content(*, auto_decompress: bool | None = None, auto_relax: bool | None = None, headers: Pyjo_Headers | None = None, max_buffer_size: int | None = None, max_leftover_size: int | None = None, relaxed: bool | None = None, skip_body: bool | None = None)

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

__init__(*, auto_decompress: bool | None = None, auto_relax: bool | None = None, headers: Pyjo_Headers | None = None, max_buffer_size: int | None = None, max_leftover_size: int | None = None, relaxed: bool | None = None, skip_body: bool | None = None) → None

content = Pyjo.Content.Single.new()
content = Pyjo.Content.Single.new(headers=Pyjo.Headers.new())

Construct a new content object.

auto_decompress

boolean = content.auto_decompress
content.auto_decompress = boolean

Decompress content automatically if is_compressed is true.

auto_relax

boolean = content.auto_relax
content.auto_relax = boolean

Try to detect when relaxed parsing is necessary.

body_contains(chunk: bytes) → bool

boolean = content.body_contains(b'foo bar baz')

Check if content contains a specific string. Meant to be overloaded in a subclass.

property body_size: int

size = content.body_size

Content size in bytes. Meant to be overloaded in a subclass.

property boundary: str | None

boundary = content.boundary

Extract multipart boundary from Content-Type header.

property charset: str | None

charset = content.charset

Extract charset from Content-Type header.

clone() → Self | None

clone = content.clone()

Return a new content object cloned from this content if possible, otherwise return None.

generate_body_chunk(offset: int) → bytes | None

chunk = content.generate_body_chunk(0)

Generate dynamic content.

get_body_chunk(offset: int) → bytes | None

chunk = content.get_body_chunk(0)

Get a chunk of content starting from a specific position. Meant to be overloaded in a subclass.

get_header_chunk(offset: int) → bytes

chunk = content.get_header_chunk(13)

Get a chunk of the headers starting from a specific position. Note that this method finalizes the content.

property header_size: int

size = content.header_size

Size of headers in bytes. Note that this method finalizes the content.

property headers: Pyjo_Headers

headers = content.headers
content.headers = Pyjo.Headers.new()

Content headers, defaults to a Pyjo.Headers object.

headers_contain(chunk: bytes) → bool

boolean = content.headers_contain(b'foo bar baz')

Check if headers contain a specific string. Note that this method finalizes the content.

property is_chunked: bool

boolean = content.is_chunked

Check if Transfer-Encoding header indicates chunked transfer encoding.

property is_compressed: bool

boolean = content.is_compressed

Check Content-Encoding header for gzip value.

property is_dynamic: bool

boolean = content.is_dynamic

Check if content will be dynamically generated, which prevents clone() from working.

property is_finished: bool

boolean = content.is_finished

Check if parser is finished.

property is_limit_exceeded: bool

boolean = content.is_limit_exceeded

Check if buffer has exceeded max_buffer_size.

property is_multipart: bool

boolean = content.is_multipart

False, this is not a Pyjo.Content.MultiPart object.

property is_parsing_body: bool

boolean = content.is_parsing_body

Check if body parsing started yet.

property is_sse: bool

boolean = content.is_sse

Check if Content-Type header indicates Server-Sent Events (SSE). Note that this method is EXPERIMENTAL and may change without warning!

leftovers() → bytes

data = content.leftovers()

Get leftover data from content parser.

property max_buffer_size: int

size = content.max_buffer_size
content.max_buffer_size = 1024

Maximum size in bytes of buffer for content parser, defaults to the value of the PYJO_MAX_BUFFER_SIZE environment variable or 262144 (256KiB).

property max_leftover_size: int

size = content.max_leftover_size
content.max_leftover_size = 1024

Maximum size in bytes of buffer for pipelined HTTP requests, defaults to the value of the PYJO_MAX_LEFTOVER_SIZE environment variable or 262144 (256KiB).

parse(chunk: bytes | None = None) → Pyjo_Content

content = content.parse(b'Content-Length: 12\x0d\x0a\x0d\x0aHello World!')

Parse content chunk.

parse_body(chunk: bytes | None = None) → Pyjo_Content

content = content.parse_body(b'Hi!')

Parse body chunk and skip headers.

property progress: int

size = content.progress

Size of content already received from message in bytes.

relaxed

boolean = content.relaxed
content.relaxed = boolean

Activate relaxed parsing for responses that are terminated with a connection close.

skip_body

boolean = content.skip_body
content.skip_body = boolean

Skip body parsing and finish after headers.

write(chunk: bytes | None = None, cb: DrainCallback | None = None) → Self

content = content.write()
content = content.write(b'')
content = content.write(chunk)
content = content.write(chunk, lambda content, offset: ...)

Write dynamic content non-blocking, the optional drain callback will be executed once all data has been written. Calling this method without a chunk of data will finalize the headers and allow for dynamic content to be written later. You can write an empty chunk of data at any time to end the stream.

# Make sure previous chunk of data has been written before continuing
content.write(
    b'He',
    lambda content, offset: content.write(b'llo!', lambda content, offset: content.write(b'')),
)

write_chunk(chunk: bytes | None = None, cb: DrainCallback | None = None) → Self

content = content.write_chunk()
content = content.write_chunk(b'')
content = content.write_chunk(chunk)
content = content.write_chunk(chunk, lambda content, offset: ...)

Write dynamic content non-blocking with chunked transfer encoding, the optional drain callback will be executed once all data has been written. Calling this method without a chunk of data will finalize the headers and allow for dynamic content to be written later. You can write an empty chunk of data at any time to end the stream.

# Make sure previous chunk of data has been written before continuing
content.write_chunk(
    b'He',
    lambda content, offset: content.write_chunk(b'llo!', lambda content, offset: content.write_chunk(b'')),
)

write_sse(event: Mapping[str, Any] | None = None, cb: DrainCallback | None = None) → Self

content = content.write_sse()
content = content.write_sse(event)
content = content.write_sse(event, lambda content, offset: ...)

Write Server-Sent Event (SSE) non-blocking, the optional drain callback will be executed once all data has been written. Calling this method without an event will finalize the response headers and allow for events to be written later. Note that this method is EXPERIMENTAL and may change without warning!