mxaccessgw/clients/python

Python Client

The Python client package contains generated MXAccess Gateway protobuf bindings, the async mxgateway package, and the mxgw-py test CLI. The package uses the shared proto inputs documented in ../../docs/ClientProtoGeneration.md so gateway and client contracts stay in sync.

Layout

clients/python/
  pyproject.toml
  generate-proto.ps1
  src/mxgateway/
  src/mxgateway/generated/
  src/mxgateway_cli/
  tests/

src/mxgateway/generated contains code produced by grpc_tools.protoc. Do not edit generated files by hand.

Regenerating Protobuf Bindings

Run generation after the shared .proto files or the Python output path changes:

./generate-proto.ps1

The script uses the Python tool path recorded in ../../docs/ToolchainLinks.md.

Build And Test

Run the Python checks from clients/python:

python -m pip install -e ".[dev]"
python -m pytest
python -m pip wheel . --no-deps --wheel-dir "$env:TEMP\mxgateway-python-wheel"

The tests import the generated gateway and worker stubs, run fake async gateway stubs, verify API key metadata, exercise stream cancellation, load shared value and command fixtures, and check deterministic CLI output.

Packaging

Install the package in editable mode for local development:

python -m pip install -e ".[dev]"

Build a wheel from clients/python:

python -m pip wheel . --no-deps --wheel-dir "$env:TEMP\mxgateway-python-wheel"

Install the generated wheel into a target environment:

python -m pip install <wheel-path>

The wheel exposes the mxgw-py console script.

Library Usage

The library is async-first:

from mxgateway import GatewayClient

async with await GatewayClient.connect(
    endpoint="localhost:5000",
    api_key="<gateway-api-key>",
    plaintext=True,
) as client:
    session = await client.open_session(client_session_name="python-client")
    try:
        server_handle = await session.register("python-client")
        item_handle = await session.add_item(server_handle, "Object.Attribute")
        await session.advise(server_handle, item_handle)
    finally:
        await session.close()

GatewayClient.open_session_raw, GatewayClient.invoke_raw, and GatewayClient.stream_events_raw keep the generated protobuf replies and events available for parity tests. Session helpers call the method-specific MXAccess commands and preserve raw replies on typed command exceptions.

*_raw methods (GatewayClient.invoke_raw, Session.invoke_raw) surface gateway protocol failures by raising the typed MxGateway* exceptions, but they deliberately do not run MXAccess-failure detection: an MXAccess HRESULT or MxStatusProxy status failure is left embedded in the returned reply and no MxAccessError is raised. Session.invoke adds that check on top. Parity-test callers using invoke_raw must inspect the reply's protocol_status, hresult, and statuses themselves. The non-raw Session helpers (register, add_item, write, the bulk methods, etc.) run the check and raise MxAccessError.

Value conversion (to_mx_value, used by Session.write/write2 and the bulk helpers) rejects non-finite floats — nan, inf, and -inf raise ValueError rather than being forwarded to MXAccess, which has no defined wire representation for them. Python bytes values are an opaque VT_RECORD pass-through that MXAccess does not interpret.

Canceling a Python task cancels the client-side gRPC call or stream wait. It does not abort an in-flight MXAccess COM call inside the worker process.

Galaxy Repository Browse

The GalaxyRepositoryClient wraps the read-only GalaxyRepository gRPC service. It lets callers test connectivity to the AVEVA System Platform Galaxy Repository (ZB SQL database), read the last deploy timestamp, and enumerate the deployed object hierarchy plus each object's dynamic attributes:

from mxgateway import GalaxyRepositoryClient

async with await GalaxyRepositoryClient.connect(
    endpoint="localhost:5000",
    api_key="<gateway-api-key>",
    plaintext=True,
) as galaxy:
    if not await galaxy.test_connection():
        raise RuntimeError("gateway cannot reach the Galaxy Repository DB")

    last_deploy = await galaxy.get_last_deploy_time()
    print(f"last deploy: {last_deploy}")

    for obj in await galaxy.discover_hierarchy():
        print(obj.tag_name, obj.contained_name)
        for attr in obj.attributes:
            print("  ", attr.attribute_name, "->", attr.full_tag_reference)

The methods return native Python types (bool, datetime | None, and a list[GalaxyObject] of generated proto messages) so callers can index into the hierarchy without learning the underlying stub class. The service requires the metadata:read scope on the API key.

discover_hierarchy buffers every object (with its full attribute list) into a single in-memory list. For a large Galaxy use iter_hierarchy instead — it is an async generator that fetches one page at a time and yields objects as they arrive, so peak memory stays bounded by a single page rather than the whole hierarchy:

async with await GalaxyRepositoryClient.connect(
    endpoint="localhost:5000",
    api_key="<gateway-api-key>",
    plaintext=True,
) as galaxy:
    async for obj in galaxy.iter_hierarchy():
        print(obj.tag_name, obj.contained_name)

Pages are fetched lazily: the next page is only requested once the caller has consumed every object from the current page.

Watching deploy events

GalaxyRepositoryClient.watch_deploy_events opens a server-streaming subscription that emits the current cached deploy state immediately and then one DeployEvent per new Galaxy deploy. sequence is monotonic per gateway start; gaps mean events were dropped from the per-subscriber buffer. Pass last_seen_deploy_time to suppress the bootstrap event when the caller already has the current state cached:

from datetime import datetime, timezone
from mxgateway import DeployEvent, GalaxyRepositoryClient

async with await GalaxyRepositoryClient.connect(
    endpoint="localhost:5000",
    api_key="<gateway-api-key>",
    plaintext=True,
) as galaxy:
    last_seen: datetime | None = None
    async for event in galaxy.watch_deploy_events(last_seen_deploy_time=last_seen):
        assert isinstance(event, DeployEvent)
        print(
            f"#{event.sequence} deploy={event.time_of_last_deploy.ToDatetime(tzinfo=timezone.utc)} "
            f"objects={event.object_count} attributes={event.attribute_count}"
        )
        if event.time_of_last_deploy_present:
            last_seen = event.time_of_last_deploy.ToDatetime(tzinfo=timezone.utc)

The method returns an async iterator yielding the generated DeployEvent proto. Breaking out of the loop, calling aclose() on the iterator, or cancelling the surrounding task closes the underlying gRPC stream cleanly. The streaming RPC requires the same metadata:read scope as the other Galaxy methods. The CLI does not currently expose a streaming watch-deploy-events subcommand — use the library API directly when subscribing to deploy events from Python.

Authentication And TLS

ClientOptions.api_key adds this metadata to unary calls and streams:

authorization: Bearer <api-key>

The client supports plaintext channels for local development, TLS with system roots, TLS with a custom ca_file, and an optional test server name override. API keys are redacted from option repr output and CLI error output.

CLI

The CLI emits deterministic JSON for automation:

mxgw-py version --json
mxgw-py open-session --endpoint localhost:5000 --plaintext --json
mxgw-py register --session-id <id> --client-name python-client --json
mxgw-py add-item --session-id <id> --server-handle 1 --item Object.Attribute --json
mxgw-py advise --session-id <id> --server-handle 1 --item-handle 2 --json
mxgw-py stream-events --session-id <id> --max-events 1 --json
mxgw-py write --session-id <id> --server-handle 1 --item-handle 2 --type int32 --value 123 --json

Use --api-key or --api-key-env MXGATEWAY_API_KEY to attach API key metadata. smoke opens a session, registers, adds an item, advises, streams a bounded event count, and closes the session in a finally block.

Use TLS options for a secured gateway:

mxgw-py smoke --endpoint mxgateway.example.local:5001 --tls --ca-file C:\certs\mxgateway-ca.pem --server-name-override mxgateway.example.local --api-key-env MXGATEWAY_API_KEY --item Object.Attribute --json

Integration Checks

Run live checks only when a gateway and MXAccess-backed worker are available:

$env:MXGATEWAY_INTEGRATION = '1'
$env:MXGATEWAY_ENDPOINT = 'localhost:5000'
$env:MXGATEWAY_API_KEY = '<gateway-api-key>'
$env:MXGATEWAY_TEST_ITEM = 'Object.Attribute'
mxgw-py smoke --endpoint $env:MXGATEWAY_ENDPOINT --plaintext --api-key-env MXGATEWAY_API_KEY --item $env:MXGATEWAY_TEST_ITEM --json

Related Documentation

• Client Packaging
• Client Proto Generation
• Python Client Detailed Design