# 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 ```text 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: ```powershell ./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`: ```powershell 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: ```powershell python -m pip install -e ".[dev]" ``` Build a wheel from `clients/python`: ```powershell python -m pip wheel . --no-deps --wheel-dir "$env:TEMP\mxgateway-python-wheel" ``` Install the generated wheel into a target environment: ```powershell python -m pip install ``` The wheel exposes the `mxgw-py` console script. ## Library Usage The library is async-first: ```python from mxgateway import GatewayClient async with await GatewayClient.connect( endpoint="localhost:5000", 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. 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: ```python from mxgateway import GalaxyRepositoryClient async with await GalaxyRepositoryClient.connect( endpoint="localhost:5000", 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: ```python async with await GalaxyRepositoryClient.connect( endpoint="localhost:5000", 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: ```python from datetime import datetime, timezone from mxgateway import DeployEvent, GalaxyRepositoryClient async with await GalaxyRepositoryClient.connect( endpoint="localhost:5000", 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: ```text authorization: Bearer ``` 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: ```powershell mxgw-py version --json mxgw-py open-session --endpoint localhost:5000 --plaintext --json mxgw-py register --session-id --client-name python-client --json mxgw-py add-item --session-id --server-handle 1 --item Object.Attribute --json mxgw-py advise --session-id --server-handle 1 --item-handle 2 --json mxgw-py stream-events --session-id --max-events 1 --json mxgw-py write --session-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: ```powershell 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: ```powershell $env:MXGATEWAY_INTEGRATION = '1' $env:MXGATEWAY_ENDPOINT = 'localhost:5000' $env:MXGATEWAY_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](../../docs/ClientPackaging.md) - [Client Proto Generation](../../docs/ClientProtoGeneration.md) - [Python Client Detailed Design](./PythonClientDesign.md)