StillDeep SDK — Python client

Prerequisites

• Python 3.11+
• pip install grpcio grpcio-tools

Setup

mkdir my-pos-client && cd my-pos-client
python -m venv .venv && source .venv/bin/activate  # Windows: .venv\Scripts\activate
pip install grpcio grpcio-tools

cp -r ../waxpos/src/public/StillDeep.Intelligence.Contracts/protos ./protos
mkdir generated

python -m grpc_tools.protoc \
  -I./protos \
  --python_out=./generated \
  --grpc_python_out=./generated \
  ./protos/common.proto ./protos/receipt.proto ./protos/sales.proto ./protos/catalog.proto \
  ./protos/member.proto ./protos/device.proto ./protos/events.proto \
  ./protos/user.proto ./protos/shift.proto

Example: sign in, open the shift, sell one item

Sequence and required fields: see getting-started.md.

import os
import uuid
import grpc

import sys; sys.path.insert(0, './generated')
import common_pb2
import sales_pb2, sales_pb2_grpc
import shift_pb2, shift_pb2_grpc
import user_pb2, user_pb2_grpc

ADDRESS = os.environ.get('STILLDEEP_GRPC', 'localhost:50051')
POST_ID = os.environ['STILLDEEP_POST_ID']               # required: id of this post
SKU = os.environ['STILLDEEP_SKU']                       # product to sell
METHOD = os.environ.get('STILLDEEP_METHOD', 'cash')     # a code from ConfigService.GetPostConfig

def key() -> str:
    return str(uuid.uuid4())

with grpc.insecure_channel(ADDRESS) as channel:
    users = user_pb2_grpc.UserServiceStub(channel)
    shifts = shift_pb2_grpc.ShiftServiceStub(channel)
    sales = sales_pb2_grpc.SalesServiceStub(channel)

    # 1. Authenticate (required).
    auth = users.Authenticate(user_pb2.AuthenticateRequest(
        post_id=POST_ID, mode=user_pb2.AUTH_MODE_PASSWORD,
        login=os.environ['STILLDEEP_LOGIN'], password=os.environ['STILLDEEP_PASSWORD']))

    # 2-3. Reuse the open shift, or open one.
    try:
        shift = shifts.GetCurrentShift(shift_pb2.GetCurrentShiftRequest(post_id=POST_ID))
    except grpc.RpcError as e:
        if e.code() != grpc.StatusCode.NOT_FOUND:
            raise
        shift = shifts.OpenShift(shift_pb2.OpenShiftRequest(
            post_id=POST_ID, user_id=auth.user.user_id,
            opening_cash=common_pb2.Money(cents=0), idempotency_key=key())).shift

    # Every following call carries x-shift-id.
    md = [('x-shift-id', shift.shift_id)]

    # 4. CreateCart (required).
    cart_id = sales.CreateCart(sales_pb2.CreateCartRequest(
        post_id=POST_ID, idempotency_key=key()), metadata=md).cart.cart_id

    # 5. AddItem (required): product_id, sku or barcode; quantity in milli-units.
    sales.AddItem(sales_pb2.AddItemRequest(
        cart_id=cart_id, sku=SKU, quantity=common_pb2.Quantity(milli_units=1000),
        idempotency_key=key()), metadata=md)

    # 6. BeginPayment (optional): handle an extension question before paying.
    checkout = sales.BeginPayment(sales_pb2.BeginPaymentRequest(cart_id=cart_id), metadata=md)
    if checkout.HasField('question'):
        raise RuntimeError(f'Answer required: {checkout.question.code}')

    # 8. ProcessPayment (required): pay the remaining amount.
    paid = sales.ProcessPayment(sales_pb2.ProcessPaymentRequest(
        cart_id=cart_id, method_code=METHOD, amount=checkout.cart.remaining_due,
        idempotency_key=key()), metadata=md)
    if paid.payment.status in (sales_pb2.PAYMENT_STATUS_DECLINED, sales_pb2.PAYMENT_STATUS_FAILED):
        raise RuntimeError(paid.refusal_guide.user_facing_message)

    # 9. FinalizeTransaction (required).
    done = sales.FinalizeTransaction(sales_pb2.FinalizeRequest(
        cart_id=cart_id, idempotency_key=key()), metadata=md)
    print('Receipt', done.transaction.receipt_number, 'change', paid.change_due.cents)

Event stream

EventService is declared in events.proto but not served by the current POSIntelligenceSvc (UNIMPLEMENTED).

import events_pb2, events_pb2_grpc

with grpc.insecure_channel(ADDRESS) as channel:
    stub = events_pb2_grpc.EventServiceStub(channel)
    request = events_pb2.SubscribeRequest(subscriber_id=str(uuid.uuid4()), post_id_filter=POST_ID)
    for ev in stub.Subscribe(request, metadata=md):
        print(ev.WhichOneof('payload'), ev)