Results API¶

qm.StreamsManager ¶

StreamsManager(
    service: Union[JobResultServiceApi, JobResultApi],
    capabilities: ServerCapabilities,
    wait_until_func: Optional[
        Callable[[Literal["Done"], float], None]
    ],
)

Access to the results of a QmJob

This object is created by calling QmJob.result_handles

Assuming you have an instance of StreamsManager:

    job_results: StreamsManager

This object is iterable:

    for name, handle in job_results:
        print(name)

Can detect if a name exists:

if "somename" in job_results:
    print("somename exists!")
    handle = job_results.get("somename")

fetch_results ¶

fetch_results(
    wait_until_done: bool = True,
    timeout: float = VERY_LONG_TIME,
    stream_names: Optional[
        Mapping[str, Union[int, slice]]
    ] = None,
    item: None = None,
) -> Mapping[
    str,
    Union[
        numpy.typing.NDArray[numpy.generic],
        Optional[Number],
    ],
]

fetch_results(
    wait_until_done: bool = True,
    timeout: float = VERY_LONG_TIME,
    stream_names: Optional[Collection[str]] = None,
    item: Optional[Union[int, slice]] = None,
) -> Mapping[
    str,
    Union[
        numpy.typing.NDArray[numpy.generic],
        Optional[Number],
    ],
]

fetch_results(
    wait_until_done: bool = True,
    timeout: float = VERY_LONG_TIME,
    stream_names: Optional[
        Union[
            Mapping[str, Union[int, slice]], Collection[str]
        ]
    ] = None,
    item: Optional[Union[int, slice]] = None,
) -> Mapping[
    str,
    Union[
        numpy.typing.NDArray[numpy.generic],
        Optional[Number],
    ],
]

Fetch results from the specified streams

PARAMETER	DESCRIPTION
`wait_until_done`	If True, will wait until all results are processed before fetching TYPE: `bool` DEFAULT: `True`
`timeout`	Timeout (in seconds) that will be applied to each of the api requests. This means that the actual overall timeout is at least the one given, but it could be more. TYPE: `float` DEFAULT: `VERY_LONG_TIME`
`stream_names`	A mapping of stream names to indices or slices to fetch, or a collection of stream names to fetch all items from TYPE: `Optional[Union[Mapping[str, Union[int, slice]], Collection[str]]]` DEFAULT: `None`
`item`	An index or slice to fetch from each stream TYPE: `Optional[Union[int, slice]]` DEFAULT: `None`

RETURNS	DESCRIPTION
`Mapping[str, Union[NDArray[generic], Optional[Number]]]`	A mapping of stream names to their fetched results as numpy arrays

get ¶

get(
    name: str,
    /,
    default: Optional[
        Union[AnySingleStreamFetcher, _T]
    ] = None,
) -> Optional[Union[AnySingleStreamFetcher, _T]]

Get a handle to a named result from stream_processing

PARAMETER	DESCRIPTION
`name`	The named result using in stream_processing TYPE: `str`
`default`	The default value to return if the named result is unknown TYPE: `Optional[Union[AnySingleStreamFetcher, _T]]` DEFAULT: `None`

RETURNS	DESCRIPTION
`Optional[Union[AnySingleStreamFetcher, _T]]`	A handle object to the results `MultipleNamedJobResult` or `SingleNamedJobResult` or None if the named results in unknown

is_processing ¶

is_processing() -> bool

Check if the job is still processing results

RETURNS	DESCRIPTION
`bool`	True if results are still being processed, False otherwise

items ¶

items() -> ItemsView[str, AnySingleStreamFetcher]

Returns a view, in which the first item is the name of the result and the second is the result

keys ¶

keys() -> KeysView[str]

Returns a view of the names of the results

values ¶

values() -> ValuesView[AnySingleStreamFetcher]

Returns a view of the results

wait_for_all_values ¶

wait_for_all_values(
    timeout: Optional[float] = None,
) -> bool

Wait until we know all values were processed for all named results

PARAMETER	DESCRIPTION
`timeout`	Timeout for waiting in seconds TYPE: `Optional[float]` DEFAULT: `None`

RETURNS	DESCRIPTION
`bool`	Returns True if all completed successfully; False if any result stream was closed prematurely (e.g., due to job failure or cancellation).

qm.BaseSingleStreamFetcher ¶

BaseSingleStreamFetcher(
    schema: JobResultItemSchema,
    service: Union[JobResultServiceApi, JobResultApi],
    stream_metadata_errors: list[StreamMetadataError],
    stream_metadata: Optional[StreamMetadata],
    capabilities: ServerCapabilities,
    multiple_streams_fetcher: Optional[
        MultipleStreamsFetcher
    ],
)

job_id `property` ¶

job_id: str

The job id this result came from

name `property` ¶

name: str

The name of the result this handle is connected to

stream_metadata `property` ¶

stream_metadata: Optional[StreamMetadata]

Provides the StreamMetadata of this stream.

Metadata currently includes the values and shapes of the automatically identified loops in the program.

count_so_far ¶

count_so_far() -> int

also len(handle)

RETURNS	DESCRIPTION
`int`	The number of values this result has so far

fetch `abstractmethod` ¶

fetch(
    item: Union[int, slice],
    *,
    check_for_errors: bool = True,
    flat_struct: bool = False,
    timeout: Optional[float] = None
) -> ReturnedT

Fetch a single result from the current result stream saved in server memory. The result stream is populated by the save().

PARAMETER	DESCRIPTION
`item`	ignored TYPE: `Union[int, slice]`
`check_for_errors`	If true, the function would also check whether run-time errors happened during the program execution and would write to the logger an error message. TYPE: `bool` DEFAULT: `True`
`flat_struct`	results will have a flat structure - dimensions will be part of the shape and not of the type TYPE: `bool` DEFAULT: `False`
`timeout`	Timeout for waiting in seconds TYPE: `Optional[float]` DEFAULT: `None`

RETURNS	DESCRIPTION
`ReturnedT`	the current result

Example

res.fetch() # return the item in the top position

fetch_all ¶

fetch_all(
    *,
    check_for_errors: bool = True,
    flat_struct: bool = False
) -> ReturnedT

Fetch a result from the current result stream saved in server memory. The result stream is populated by the save() and save_all() statements. Note that if save_all() statements are used, calling this function twice may give different results.

PARAMETER	DESCRIPTION
`check_for_errors`	If true, the function would also check whether run-time errors happened during the program execution and would write to the logger an error message. TYPE: `bool` DEFAULT: `True`
`flat_struct`	results will have a flat structure - dimensions will be part of the shape and not of the type TYPE: `bool` DEFAULT: `False`

RETURNS	DESCRIPTION
`ReturnedT`	all results of current result stream

has_dataloss ¶

has_dataloss() -> bool

Returns: true if there was data loss during job execution

strict_fetch ¶

strict_fetch(
    item: Union[int, slice],
    *,
    check_for_errors: bool = True,
    flat_struct: bool = False,
    timeout: Optional[float] = None
) -> numpy.typing.NDArray[numpy.generic]

Fetch a result from the current result stream saved in server memory. The result stream is populated by the save() and save_all() statements. Note that if save_all() statements are used, calling this function twice with the same item index may give different results.

PARAMETER	DESCRIPTION
`item`	The index of the result in the saved results stream. TYPE: `Union[int, slice]`
`check_for_errors`	If true, the function would also check whether run-time errors happened during the program execution and would write to the logger an error message. TYPE: `bool` DEFAULT: `True`
`flat_struct`	results will have a flat structure - dimensions will be part of the shape and not of the type TYPE: `bool` DEFAULT: `False`
`timeout`	Timeout for waiting in seconds TYPE: `Optional[float]` DEFAULT: `None`

RETURNS	DESCRIPTION
`NDArray[generic]`	a single result if item is integer or multiple results if item is a Python slice object.

RAISES	DESCRIPTION
`Exception`	If item is not an integer or a slice object.
`StreamProcessingDataLossError`	If data loss is detected in the data for the job.

Example

res.fetch(0)         #return the item in the top position
res.fetch(1)         #return the item in position number 2
res.fetch(slice(1,6))# return items from position 1 to position 6 (exclusive)
                     # same as res.fetch_all()[1:6]

wait_for_all_values ¶

wait_for_all_values(
    timeout: float = VERY_LONG_TIME,
) -> bool

Wait until we know all values were processed for this named result

PARAMETER	DESCRIPTION
`timeout`	Timeout for waiting in seconds TYPE: `float` DEFAULT: `VERY_LONG_TIME`

RETURNS	DESCRIPTION
`bool`	True if the job finished successfully and False if the job was closed before it was done.
`bool`	If the job is still running when reaching the timeout, a TimeoutError is raised.

wait_for_values ¶

wait_for_values(
    count: int = 1, timeout: float = VERY_LONG_TIME
) -> None

Wait until we know at least count values were processed for this named result

PARAMETER	DESCRIPTION
`count`	The number of items to wait for TYPE: `int` DEFAULT: `1`
`timeout`	Timeout for waiting in seconds TYPE: `float` DEFAULT: `VERY_LONG_TIME`

qm.SingleStreamSingleResultFetcher ¶

SingleStreamSingleResultFetcher(
    schema: JobResultItemSchema,
    service: Union[JobResultServiceApi, JobResultApi],
    stream_metadata_errors: list[StreamMetadataError],
    stream_metadata: Optional[StreamMetadata],
    capabilities: ServerCapabilities,
    multiple_streams_fetcher: Optional[
        MultipleStreamsFetcher
    ],
)

A handle to a result of a pipeline terminating with save

fetch ¶

fetch(
    item: Union[int, slice],
    *,
    check_for_errors: bool = True,
    flat_struct: bool = False,
    timeout: Optional[float] = None
) -> Optional[numpy.typing.NDArray[numpy.generic]]

Fetch a single result from the current result stream saved in server memory. The result stream is populated by the save().

PARAMETER	DESCRIPTION
`item`	ignored TYPE: `Union[int, slice]`
`check_for_errors`	If true, the function would also check whether run-time errors happened during the program execution and would write to the logger an error message. TYPE: `bool` DEFAULT: `True`
`flat_struct`	results will have a flat structure - dimensions will be part of the shape and not of the type TYPE: `bool` DEFAULT: `False`
`timeout`	Timeout for waiting in seconds TYPE: `Optional[float]` DEFAULT: `None`

RETURNS	DESCRIPTION
`Optional[NDArray[generic]]`	the current result

Example

res.fetch() # return the item in the top position

fetch_all ¶

fetch_all(
    *,
    check_for_errors: bool = True,
    flat_struct: bool = False
) -> Optional[numpy.typing.NDArray[numpy.generic]]

Fetch a result from the current result stream saved in server memory. The result stream is populated by the save() and save_all() statements. Note that if save_all() statements are used, calling this function twice may give different results.

PARAMETER	DESCRIPTION
`check_for_errors`	If true, the function would also check whether run-time errors happened during the program execution and would write to the logger an error message. TYPE: `bool` DEFAULT: `True`
`flat_struct`	results will have a flat structure - dimensions will be part of the shape and not of the type TYPE: `bool` DEFAULT: `False`

RETURNS	DESCRIPTION
`Optional[NDArray[generic]]`	all result of current result stream

qm.SingleStreamMultipleResultFetcher ¶

SingleStreamMultipleResultFetcher(
    schema: JobResultItemSchema,
    service: Union[JobResultServiceApi, JobResultApi],
    stream_metadata_errors: list[StreamMetadataError],
    stream_metadata: Optional[StreamMetadata],
    capabilities: ServerCapabilities,
    multiple_streams_fetcher: Optional[
        MultipleStreamsFetcher
    ],
)

A handle to a result of a pipeline terminating with save_all

fetch ¶

fetch(
    item: Union[int, slice],
    *,
    check_for_errors: bool = True,
    flat_struct: bool = False,
    timeout: Optional[float] = None
) -> numpy.typing.NDArray[numpy.generic]

Fetch a result from the current result stream saved in server memory. The result stream is populated by the save() and save_all() statements. Note that if save_all() statements are used, calling this function twice with the same item index may give different results.

PARAMETER	DESCRIPTION
`item`	The index of the result in the saved results stream. TYPE: `Union[int, slice]`
`check_for_errors`	If true, the function would also check whether run-time errors happened during the program execution and would write to the logger an error message. TYPE: `bool` DEFAULT: `True`
`flat_struct`	results will have a flat structure - dimensions will be part of the shape and not of the type TYPE: `bool` DEFAULT: `False`
`timeout`	Timeout for waiting in seconds TYPE: `Optional[float]` DEFAULT: `None`

RETURNS	DESCRIPTION
`NDArray[generic]`	a single result if item is integer or multiple results if item is Python slice object.

Example

res.fetch(0)         # return the item in the top position
res.fetch(1)         # return the item in position number 2
res.fetch(slice(1,6))# return items from position 1 to position 6 (exclusive)
                     # same as res.fetch_all()[1:6]

Results API¶

qm.StreamsManager ¶

fetch_results ¶

get ¶

is_processing ¶

items ¶

keys ¶

values ¶

wait_for_all_values ¶

qm.BaseSingleStreamFetcher ¶

job_id property ¶

name property ¶

stream_metadata property ¶

count_so_far ¶

fetch abstractmethod ¶

fetch_all ¶

has_dataloss ¶

strict_fetch ¶

wait_for_all_values ¶

wait_for_values ¶

qm.SingleStreamSingleResultFetcher ¶

fetch ¶

fetch_all ¶

qm.SingleStreamMultipleResultFetcher ¶

fetch ¶

job_id `property` ¶

name `property` ¶

stream_metadata `property` ¶

fetch `abstractmethod` ¶