Workflow Results¶

You can use the Results API to perform actions such as fetching, viewing, updating, locking and deleting results from the database, along with many other actions.

Usage¶

You can instantiate the Results object to interact with the API.

Python

from workflow.http.context import HTTPContext

ResultsContext = HTTPContext(backends=["results"])

results = ResultsContext.results

Once this is done, all of the methods listed below will be available on the results object. For example, you can call a method like this:

Python

results.get_by_count('header-localization', 10)

Methods¶

Bases: Client

HTTP Client for interacting with the Results backend.

Parameters:

Name	Type	Description	Default
`Client`	`client`	The base class for interacting with the backend.	required

Returns:

Name	Type	Description
`Results`		A client for interacting with the Buckets backend.

count ¶

Python

count(pipeline: str, query: Dict[str, Any]) -> int

Get count of results filtered by the query.

Parameters:

Name	Type	Description	Default
`pipeline`	`str`	Name of pipeline.	required
`query`	`Dict[str, Any]`	MongoDB query.	required

Returns:

Name	Type	Description
`int`	`int`	The number of results that match the query.

delete_ids ¶

Python

delete_ids(pipeline: str, ids: List[str]) -> bool

Delete results from the results backend.

Parameters:

Name	Type	Description	Default
`pipeline`	`str`	Name of pipeline.	required
`ids`	`List[str]`	The IDs of the works to delete.	required

Returns:

Name	Type	Description
`bool`	`bool`	Whether the results were deleted successfully.

deposit ¶

Python

deposit(works: List[Dict[str, Any]]) -> Dict[str, bool]

Deposit works into the results backend.

Parameters:

Name	Type	Description	Default
`works`	`List[Dict[str, Any]]`	A list of payloads from Work Objects.	required

Note

This method is not intended for direct use by the user. Work status must be 'success' or 'failure'.

Raises:

Type	Description
`AssertionError`	If work status is not 'success' or 'failure'.

Returns:

Type	Description
`Dict[str, bool]`	Dict[str, bool]: Dictionary of deposit results for each pipeline.

Examples:

from workflow.http.results import Results from workflow.definitions.work import Work results = Results() work = Work.withdraw(pipeline="test) work.status = "success" status = results.deposit([work.payload])

get_by_count ¶

Python

get_by_count(pipeline: str, count: int = 10) -> List[Dict[str, Any]]

Get results by count from the specified pipeline.

Parameters:

Name	Type	Description	Default
`pipeline`	`str`	The name of the pipeline.	required
`count`	`int`	The number of results to get. Defaults to 10.	`10`

Returns:

Type	Description
`List[Dict[str, Any]]`	List[Dict[str, Any]]: Retrieved results.

get_by_event ¶

Python

get_by_event(pipeline: str, event_number: int) -> List[Dict[str, Any]]

Get work filtered by event number from the results backend.

Parameters:

Name	Type	Description	Default
`pipeline`	`str`	The name of the pipeline.	required
`event_number`	`int`	The event number of the event.	required

Returns:

Type	Description
`List[Dict[str, Any]]`	List[Dict[str, Any]]: Retrieved results.

get_by_id ¶

Python

get_by_id(pipeline: str, ids: List[str]) -> List[Dict[str, Any]]

Get results by ID from the specified pipeline.

Parameters:

Name	Type	Description	Default
`pipeline`	`str`	The name of the pipeline.	required
`ids`	`List[str]`	The IDs of the results to get.	required

Returns:

Type	Description
`List[Dict[str, Any]]`	List[Dict[str, Any]]: Retrieved results.

get_locked ¶

Python

get_locked(pipeline: str, skip: int = 0, count: int = 10) -> List[Dict[str, Any]]

Get work that is locked from the results backend.

Parameters:

Name	Type	Description	Default
`pipeline`	`str`	The name of the pipeline.	required
`skip`	`int`	The number of results to skip. Defaults to 0.	`0`
`count`	`int`	The maximum number of results. Defaults to 10.	`10`

Note

Only results with valid results, plots and products are returned.

Returns:

Type	Description
`List[Dict[str, Any]]`	List[Dict[str, Any]]: A list of the locked results.

get_locked_count ¶

Python

get_locked_count(pipeline: str) -> int

Retrieves the count of locked results.

Parameters:

Name	Type	Description	Default
`pipeline`	`str`	The name of the pipeline.	required

Returns:

Name	Type	Description
`int`	`int`	The count of locked results.

info ¶

Python

info() -> Dict[str, Any]

Get the version of the results backend.

Returns:

Type	Description
`Dict[str, Any]`	Dict[str, Any]: The version of the results backend.

lock ¶

Python

lock(pipeline: str, ids: List[str]) -> bool

Lock results.

Note

A locked result cannot be deleted or updated.

Parameters:

Name	Type	Description	Default
`pipeline`	`str`	The name of the pipeline.	required
`ids`	`List[str]`	The IDs of the results to lock.	required

Returns:

Name	Type	Description
`Boolean`	`bool`	True if successful, False if not.

status ¶

Python

status() -> Dict[str, int]

Get the status of the results backend.

Returns:

Type	Description
`Dict[str, int]`	Dict[str, int]: The status of the results backend.

unlock ¶

Python

unlock(pipeline: str, ids: List[str], force: bool = False) -> bool

Unlock results.

Parameters:

Name	Type	Description	Default
`pipeline`	`str`	The pipeline name.	required
`ids`	`List[str]`	The IDs of the results to unlock.	required
`force`	`bool`	Force the unlock. Defaults to False.	`False`

Returns:

Name	Type	Description
`Bool`	`bool`	True if successful, False if not.

update ¶

Python

update(works: List[Dict[str, Any]]) -> bool

Update works in the results backend.

Parameters:

Name	Type	Description	Default
`works`	`List[Dict[str, Any]]`	A list of payloads from Work Objects.	required

Note

This method is not intended for direct use by the user. Work status must be 'success' or 'failure'.

Returns:

Name	Type	Description
`bool`	`bool`	Whether the works were updated successfully.

view ¶

Python

view(pipeline: str, query: Dict[str, Any], projection: Dict[str, bool] = {}, skip: int = 0, limit: int = 100) -> List[Dict[str, Any]]

View works in the results backend.

Parameters:

Name	Type	Description	Default
`pipeline`	`str`	Name of pipeline.	required
`query`	`Dict[str, Any]`	MongoDB query.	required
`projection`	`Dict[str, bool]`	MongoDB projection. Defaults to {}.	`{}`
`skip`	`int`	Number of works to skip. Defaults to 0.	`0`
`limit`	`int`	Number of works to limit to. -1 for no limit. Defaults to 100.	`100`

Returns:

Type	Description
`List[Dict[str, Any]]`	List[Dict[str, Any]]: List of work payloads.