mxaccessgw/clients/python/PythonClientDesign.md

# Python Client Detailed Design

## Purpose

Provide an async Python client package for MXAccess Gateway, plus a test CLI and
unit tests. The Python client should be useful for automation, diagnostics, and
test harnesses.

Follow the [Python Style Guide](../../docs/style-guides/PythonStyleGuide.md) for
handwritten code and the [Protobuf Style Guide](../../docs/style-guides/ProtobufStyleGuide.md)
for generated contract inputs.

## Package Layout

Recommended layout:

```text
clients/python/
  pyproject.toml
  src/zb_mom_ww_mxgateway/
    __init__.py
    client.py
    session.py
    options.py
    auth.py
    values.py
    errors.py
    generated/
  src/zb_mom_ww_mxgateway_cli/
    __main__.py
    commands.py
  tests/
```

Expected dependencies:

- `grpcio`
- `grpcio-tools`
- `protobuf`
- `click` or `typer`
- `pytest`
- `pytest-asyncio`

## Library API

Use async-first API. A sync wrapper can be added later if needed.

Suggested API:

```python
client = await GatewayClient.connect(
    endpoint="localhost:5000",
    api_key=api_key,
    plaintext=True,
)

session = await client.open_session()
server = await session.register("python-client")
item = await session.add_item(server, "TestChildObject.TestInt")
await session.advise(server, item)

async for event in session.stream_events():
    ...

await session.close()
await client.close()
```

Classes:

```python
class GatewayClient:
    @classmethod
    async def connect(cls, options: ClientOptions) -> "GatewayClient": ...
    async def open_session(self, options: OpenSessionOptions | None = None) -> "Session": ...
    async def invoke(self, request: MxCommandRequest) -> MxCommandReply: ...
    async def close(self) -> None: ...

class Session:
    session_id: str
    async def register(self, client_name: str) -> int: ...
    async def add_item(self, server_handle: int, item: str) -> int: ...
    async def add_item2(self, server_handle: int, item: str, context: str) -> int: ...
    async def advise(self, server_handle: int, item_handle: int) -> None: ...
    async def add_item_bulk(self, server_handle: int, tag_addresses: Sequence[str]) -> list[SubscribeResult]: ...
    async def advise_item_bulk(self, server_handle: int, item_handles: Sequence[int]) -> list[SubscribeResult]: ...
    async def remove_item_bulk(self, server_handle: int, item_handles: Sequence[int]) -> list[SubscribeResult]: ...
    async def unadvise_item_bulk(self, server_handle: int, item_handles: Sequence[int]) -> list[SubscribeResult]: ...
    async def subscribe_bulk(self, server_handle: int, tag_addresses: Sequence[str]) -> list[SubscribeResult]: ...
    async def unsubscribe_bulk(self, server_handle: int, item_handles: Sequence[int]) -> list[SubscribeResult]: ...
    async def write(self, server_handle: int, item_handle: int, value: MxValueInput, user_id: int = 0) -> None: ...
    async def stream_events(self) -> AsyncIterator[MxEvent]: ...
    async def close(self) -> None: ...
```

## Authentication

Use gRPC metadata:

```python
metadata = (("authorization", f"Bearer {api_key}"),)
```

Provide a metadata helper that all unary and streaming calls use. Redact API
keys in exceptions and CLI output.

## TLS

Support:

- insecure channel for local development,
- TLS channel with default roots,
- custom root certificate file.

## Streaming

Expose `stream_events` as an async iterator. Canceling the task should cancel
the gRPC stream.

Do not hide stream errors. Convert common auth/session errors into typed
exceptions.

## Error Handling

Define typed exceptions:

```python
MxGatewayError
MxGatewayTransportError
MxGatewayAuthenticationError
MxGatewayAuthorizationError
MxGatewaySessionError
MxGatewayWorkerError
MxGatewayCommandError
MxAccessError
```

`MxGatewayCommandError` should include the raw protobuf reply when available.

## Test CLI

Entry point:

```text
mxgw-py
```

Recommended commands:

```text
mxgw-py version
mxgw-py smoke --endpoint localhost:5000 --api-key-env MXGATEWAY_API_KEY --plaintext --item TestChildObject.TestInt
mxgw-py stream-events --session-id <id> --json
mxgw-py write --session-id <id> --server-handle 1 --item-handle 1 --type int32 --value 123
```

Use `click` or `typer`. JSON output should be stable for test automation.

## Unit Tests

Use `pytest` and `pytest-asyncio`.

Use fake generated stubs or an in-process test gRPC server where practical.

Required tests:

- API key metadata injection,
- API key redaction,
- insecure and TLS channel option construction,
- request construction for method helpers,
- value conversion from Python values,
- status/error mapping,
- async event iteration,
- stream cancellation,
- CLI parsing,
- JSON output.

## Integration Tests

Skip unless:

```text
MXGATEWAY_INTEGRATION=1
```

Use bounded smoke flow and always attempt `close_session` in `finally`.

## Packaging

Use `pyproject.toml`. Publishable package name should be stable, for example:

```text
mxaccess-gateway-client
```

Generated protobuf code should be regenerated through a documented command, not
edited by hand.

## Related Documentation

- [Client Libraries Detailed Design](../../docs/ClientLibrariesDesign.md)
- [Client Proto Generation](../../docs/ClientProtoGeneration.md)
- [Client Packaging](../../docs/ClientPackaging.md)
- [Python Style Guide](../../docs/style-guides/PythonStyleGuide.md)