batch_ops#

Batch RPC service.

Classes:

`BatchOps`(session)	Provides for executing operations in batch in CFX.
`BatchOpsService`(channel, metadata)	Provides class wrapping methods in the batch RPC service.

class ansys.cfx.core.services.batch_ops.BatchOps(session)#

Bases: object

Provides for executing operations in batch in CFX.

Examples

>>> import ansys.cfx.core as pycfx
>>> pypost = pycfx.PostProcessing.from_install()
>>> with pycfx.BatchOps(pypost):
...     pypost.file.load_results(file_name=<results_name>)
...     pypost.results.plane["Plane 1"] = {}

The preceding code executes both operations through a single gRPC call upon exiting the with block.

Operations that perform queries in CFX are executed immediately, while others are queued for batch execution. Some queries are executed behind the scenes while queuing an operation for batch execution. Developers must ensure that they do not depend on previously queued operations.

For example, the following code throws a KeyError as pypost.results.plane["Plane 1"] attempts to access the Plane 1 named object which has not been created yet.

>>> pypost = pycfx.PostProcessing.from_install()
>>> with pycfx.BatchOps(pypost):
...     pypost.file.load_results(file_name="StaticMixer_001.res")
...     pypost.results.plane["Plane 1"] = {}
...     pypost.results.plane["Plane 1"].option = "ZX Plane"
Traceback (most recent call last):
  ...
KeyError: "'plane' has no attribute 'Plane 1'.\n"

Classes:

Op(package, service, method, request_body)

Provides for creating a single batch operation.

Methods:

`add_op`(package, service, method, request)	Queue a single batch operation.
`clear_ops`()	Clear all queued batch operations.
`instance`()	Get the `BatchOps` instance.

class Op(package: str, service: str, method: str, request_body: bytes)#

Bases: object

Provides for creating a single batch operation.

Methods:

`__init__`(package, service, method, request_body)	Initialize an instance of the `Op` class.
`update_result`(status, data)	Update results after the batch operation is executed.

__init__(package: str, service: str, method: str, request_body: bytes) → None#: Initialize an instance of the Op class.

update_result(status: <google.protobuf.internal.enum_type_wrapper.EnumTypeWrapper object at 0x7f5c32f09010>, data: str) → None#: Update results after the batch operation is executed.

add_op(package: str, service: str, method: str, request: Message) → Op#

Queue a single batch operation. Only the non-getter operations are queued.

Parameters:

packagestr: gRPC package name.
servicestr: gRPC service name.
methodstr: gRPC method name.
requestAny: gRPC request message.

Returns:

BatchOps.Op: BatchOps.Op object with a queued attribute that is True if the operation has been queued.

clear_ops() → None#: Clear all queued batch operations.

classmethod instance() → _TBatchOps | None#

Get the BatchOps instance.

Returns:

BatchOps: BatchOps instance.

class ansys.cfx.core.services.batch_ops.BatchOpsService(channel: Channel, metadata: list[tuple[str, str]])#

Bases: object

Provides class wrapping methods in the batch RPC service.

Methods:

`__init__`(channel, metadata)	Initialize an instance of the `BatchOpsService` class.
`execute`(request)	Execute RPC of the `BatchOps` service.

__init__(channel: Channel, metadata: list[tuple[str, str]]) → None#: Initialize an instance of the BatchOpsService class.

execute(request: ExecuteRequest) → ExecuteResponse#: Execute RPC of the BatchOps service.