Skip to content

io

StreamHandle

Bases: str

This is either obtained from another method or specified as blob:<uuid> where <uuid> is an UUID of a Blob.

Source code in zendriver/cdp/io.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
class StreamHandle(str):
    """
    This is either obtained from another method or specified as ``blob:<uuid>`` where
    ``<uuid>`` is an UUID of a Blob.
    """

    def to_json(self) -> str:
        return self

    @classmethod
    def from_json(cls, json: str) -> StreamHandle:
        return cls(json)

    def __repr__(self):
        return "StreamHandle({})".format(super().__repr__())

__repr__()

Source code in zendriver/cdp/io.py 
28
29
def __repr__(self):
    return "StreamHandle({})".format(super().__repr__())

from_json(json) classmethod

Source code in zendriver/cdp/io.py 
24
25
26
@classmethod
def from_json(cls, json: str) -> StreamHandle:
    return cls(json)

to_json()

Source code in zendriver/cdp/io.py 
21
22
def to_json(self) -> str:
    return self

close(handle)

Close the stream, discard any temporary backing storage.

Parameters:

Name Type Description Default
handle StreamHandle

Handle of the stream to close.

 required
Source code in zendriver/cdp/io.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
def close(handle: StreamHandle) -> typing.Generator[T_JSON_DICT, T_JSON_DICT, None]:
    """
    Close the stream, discard any temporary backing storage.

    :param handle: Handle of the stream to close.
    """
    params: T_JSON_DICT = dict()
    params["handle"] = handle.to_json()
    cmd_dict: T_JSON_DICT = {
        "method": "IO.close",
        "params": params,
    }
    json = yield cmd_dict

read(handle, offset=None, size=None)

Read a chunk of the stream

Parameters:

Name Type Description Default
handle StreamHandle

Handle of the stream to read.

 required
offset Optional[int]

(Optional) Seek to the specified offset before reading (if not specified, proceed with offset following the last read). Some types of streams may only support sequential reads.

 None
size Optional[int]

(Optional) Maximum number of bytes to read (left upon the agent discretion if not specified).

 None

Returns:

Type Description
Generator[T_JSON_DICT, T_JSON_DICT, Tuple[Optional[bool], str, bool]]

A tuple with the following items: 0. base64Encoded - (Optional) Set if the data is base64-encoded 1. data - Data that were read. 2. eof - Set if the end-of-file condition occurred while reading.
Source code in zendriver/cdp/io.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
def read(
    handle: StreamHandle,
    offset: typing.Optional[int] = None,
    size: typing.Optional[int] = None,
) -> typing.Generator[
    T_JSON_DICT, T_JSON_DICT, typing.Tuple[typing.Optional[bool], str, bool]
]:
    """
    Read a chunk of the stream

    :param handle: Handle of the stream to read.
    :param offset: *(Optional)* Seek to the specified offset before reading (if not specified, proceed with offset following the last read). Some types of streams may only support sequential reads.
    :param size: *(Optional)* Maximum number of bytes to read (left upon the agent discretion if not specified).
    :returns: A tuple with the following items:

        0. **base64Encoded** - *(Optional)* Set if the data is base64-encoded
        1. **data** - Data that were read.
        2. **eof** - Set if the end-of-file condition occurred while reading.
    """
    params: T_JSON_DICT = dict()
    params["handle"] = handle.to_json()
    if offset is not None:
        params["offset"] = offset
    if size is not None:
        params["size"] = size
    cmd_dict: T_JSON_DICT = {
        "method": "IO.read",
        "params": params,
    }
    json = yield cmd_dict
    return (
        (
            bool(json["base64Encoded"])
            if json.get("base64Encoded", None) is not None
            else None
        ),
        str(json["data"]),
        bool(json["eof"]),
    )

resolve_blob(object_id)

Return UUID of Blob object specified by a remote object id.

Parameters:

Name Type Description Default
object_id RemoteObjectId

Object id of a Blob object wrapper.

 required

Returns:

Type Description
Generator[T_JSON_DICT, T_JSON_DICT, str]

UUID of the specified Blob.
Source code in zendriver/cdp/io.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def resolve_blob(
    object_id: runtime.RemoteObjectId,
) -> typing.Generator[T_JSON_DICT, T_JSON_DICT, str]:
    """
    Return UUID of Blob object specified by a remote object id.

    :param object_id: Object id of a Blob object wrapper.
    :returns: UUID of the specified Blob.
    """
    params: T_JSON_DICT = dict()
    params["objectId"] = object_id.to_json()
    cmd_dict: T_JSON_DICT = {
        "method": "IO.resolveBlob",
        "params": params,
    }
    json = yield cmd_dict
    return str(json["uuid"])