Examples¶
The examples/ directory contains runnable
scripts. They read credentials from environment variables (IIKO_API_KEY, IIKO_APP_ID, IIKO_CLIENT_SECRET, or
IIKO_API_LOGIN) and are run against a fake iikoCloud in the test suite, so they always match the current API.
Quickstart¶
Organizations, their terminal groups and whether the terminals are online.
"""List organizations and their terminal groups, and check that the terminals are online.
python -m examples.quickstart
"""
from __future__ import annotations
import asyncio
from iikocloudapi import IikoCloud
try:
from examples._env import client_from_env
except ImportError: # pragma: no cover - when run as a script from the examples directory
from _env import client_from_env # type: ignore[no-redef]
async def main(iiko: IikoCloud) -> None:
organizations = await iiko.organizations()
organization_ids = [org.id for org in organizations.organizations]
groups = await iiko.terminal_groups(organization_ids=organization_ids)
terminal_groups = [group for org in groups.terminal_groups for group in org.items]
alive = await iiko.terminal_groups.is_alive(
organization_ids=organization_ids,
terminal_group_ids=[group.id for group in terminal_groups],
)
online = {status.terminal_group_id: status.is_alive for status in alive.is_alive_status}
for org in organizations.organizations:
print(f"{org.name} ({org.id})")
for group in terminal_groups:
if group.organization_id == org.id:
state = "online" if online.get(group.id) else "offline"
print(f" {group.name}: {state}, time zone {group.time_zone}")
if __name__ == "__main__":
async def run() -> None:
async with client_from_env() as iiko:
await main(iiko)
asyncio.run(run())
Table order, pay first¶
The guest pays online, and the order reaches the restaurant already paid with an External payment.
"""Pay first: a table order that reaches the restaurant already paid by an external acquirer.
The guest pays online (for example through T-Bank); once the payment is confirmed, the order is created with an
``External`` payment, so the cashier does not need to take money. The acquirer's payment ID is passed in
``paymentAdditionalData.customData``.
IIKO_ORGANIZATION_ID=... IIKO_TERMINAL_GROUP_ID=... IIKO_TABLE_ID=... IIKO_PRODUCT_ID=... \\
python -m examples.table_order_pay_first
"""
from __future__ import annotations
import asyncio
from decimal import Decimal
from iikocloudapi import IikoCloud
from iikocloudapi.models import (
ExternalPayment,
ExternalPaymentAdditionalData,
PaymentType,
ProductOrderItem,
TableOrder,
)
try:
from examples._env import client_from_env, env
except ImportError: # pragma: no cover
from _env import client_from_env, env # type: ignore[no-redef]
async def external_payment_type(iiko: IikoCloud, organization_id: str) -> PaymentType:
"""The payment type configured in iikoOffice for payments processed outside of iiko."""
response = await iiko.payment_types(organization_ids=[organization_id])
for payment_type in response.payment_types:
if payment_type.payment_type_kind == "External" and not payment_type.is_deleted:
return payment_type
msg = "No payment type of kind External; create one in iikoOffice"
raise LookupError(msg)
async def main(
iiko: IikoCloud,
*,
organization_id: str,
terminal_group_id: str,
table_id: str,
product_id: str,
price: Decimal,
acquirer_payment_id: str,
) -> None:
payment_type = await external_payment_type(iiko, organization_id)
if payment_type.id is None:
msg = "The payment type has no ID"
raise LookupError(msg)
created = await iiko.order.create(
organization_id=organization_id,
terminal_group_id=terminal_group_id,
order=TableOrder(
table_ids=[table_id],
items=[ProductOrderItem(product_id=product_id, amount=Decimal(1), price=price)],
payments=[
ExternalPayment(
payment_type_id=payment_type.id,
sum=price,
is_processed_externally=True,
payment_additional_data=ExternalPaymentAdditionalData(custom_data=acquirer_payment_id),
)
],
),
)
# Creating an order is a command: wait until the terminal has accepted it.
await iiko.wait(created, organization_id=organization_id)
found = await iiko.order.by_id(organization_ids=[organization_id], order_ids=[created.order_info.id])
order = found.orders[0].order
if order is not None:
print(f"Order #{order.number}: {order.status}, {order.sum} paid {order.processed_payments_sum}")
if __name__ == "__main__":
async def run() -> None:
async with client_from_env() as iiko:
await main(
iiko,
organization_id=env("IIKO_ORGANIZATION_ID"),
terminal_group_id=env("IIKO_TERMINAL_GROUP_ID"),
table_id=env("IIKO_TABLE_ID"),
product_id=env("IIKO_PRODUCT_ID"),
price=Decimal("450.00"),
acquirer_payment_id="tbank-payment-id",
)
asyncio.run(run())
Table order, cook first¶
Open an order (or add to the one already open on the table), pay with funds held by the acquirer, close it.
"""Cook first: open a table order, add dishes while the guest is eating, then pay with held funds and close it.
1. The guest scans the QR code on the table. If the table already has an open order, dishes are added to it
instead of creating a second order.
2. When the guest asks for the bill, the acquirer confirms the funds held on the card.
3. The external payment is attached to the order and the order is closed.
Every step that changes the order is a command, so each one is followed by ``iiko.wait()``.
IIKO_ORGANIZATION_ID=... IIKO_TERMINAL_GROUP_ID=... IIKO_TABLE_ID=... IIKO_PRODUCT_ID=... \\
IIKO_PAYMENT_TYPE_ID=... python -m examples.table_order_cook_first
"""
from __future__ import annotations
import asyncio
from decimal import Decimal
from iikocloudapi import IikoCloud
from iikocloudapi.models import (
ExternalPayment,
ExternalPaymentAdditionalData,
OrderItem,
OrderStatus,
ProductOrderItem,
TableOrder,
)
try:
from examples._env import client_from_env, env
except ImportError: # pragma: no cover
from _env import client_from_env, env # type: ignore[no-redef]
async def open_order_on_table(iiko: IikoCloud, organization_id: str, table_id: str) -> str | None:
"""ID of an order that is still open on the table, if any."""
found = await iiko.order.by_table(
organization_ids=[organization_id],
table_ids=[table_id],
statuses=[OrderStatus.NEW, OrderStatus.BILL],
)
return found.orders[0].id if found.orders else None
async def add_dishes(
iiko: IikoCloud, *, organization_id: str, terminal_group_id: str, table_id: str, items: list[OrderItem]
) -> str:
order_id = await open_order_on_table(iiko, organization_id, table_id)
if order_id is None:
created = await iiko.order.create(
organization_id=organization_id,
terminal_group_id=terminal_group_id,
order=TableOrder(table_ids=[table_id], items=items),
)
await iiko.wait(created, organization_id=organization_id)
return created.order_info.id
added = await iiko.order.add_items(organization_id=organization_id, order_id=order_id, items=items)
await iiko.wait(added, organization_id=organization_id)
return order_id
async def pay_and_close(
iiko: IikoCloud, *, organization_id: str, order_id: str, payment_type_id: str, acquirer_payment_id: str
) -> None:
found = await iiko.order.by_id(organization_ids=[organization_id], order_ids=[order_id])
order = found.orders[0].order
if order is None:
msg = f"Order {order_id} is not available yet"
raise LookupError(msg)
paid = await iiko.order.change_payments(
organization_id=organization_id,
order_id=order_id,
payments=[
ExternalPayment(
payment_type_id=payment_type_id,
sum=order.sum,
is_processed_externally=True,
payment_additional_data=ExternalPaymentAdditionalData(custom_data=acquirer_payment_id),
)
],
)
await iiko.wait(paid, organization_id=organization_id)
closed = await iiko.order.close(organization_id=organization_id, order_id=order_id)
await iiko.wait(closed, organization_id=organization_id)
print(f"Order #{order.number} paid ({order.sum}) and closed")
async def main(
iiko: IikoCloud,
*,
organization_id: str,
terminal_group_id: str,
table_id: str,
product_id: str,
payment_type_id: str,
) -> None:
dishes: list[OrderItem] = [ProductOrderItem(product_id=product_id, amount=Decimal(2), price=Decimal("320.00"))]
order_id = await add_dishes(
iiko, organization_id=organization_id, terminal_group_id=terminal_group_id, table_id=table_id, items=dishes
)
# ... later, the acquirer confirms the funds held on the guest's card ...
await pay_and_close(
iiko,
organization_id=organization_id,
order_id=order_id,
payment_type_id=payment_type_id,
acquirer_payment_id="tbank-payment-id",
)
if __name__ == "__main__":
async def run() -> None:
async with client_from_env() as iiko:
await main(
iiko,
organization_id=env("IIKO_ORGANIZATION_ID"),
terminal_group_id=env("IIKO_TERMINAL_GROUP_ID"),
table_id=env("IIKO_TABLE_ID"),
product_id=env("IIKO_PRODUCT_ID"),
payment_type_id=env("IIKO_PAYMENT_TYPE_ID"),
)
asyncio.run(run())
Delivery¶
A delivery order paid in cash on delivery.
"""Create a delivery order paid in cash on delivery and follow its status.
The address format depends on the organization's ``addressFormatType`` setting: ``AddressCity`` takes a free-form
line, ``AddressLegacy`` needs a street from ``iiko.streets.by_city()``.
IIKO_ORGANIZATION_ID=... IIKO_TERMINAL_GROUP_ID=... IIKO_PRODUCT_ID=... IIKO_PAYMENT_TYPE_ID=... \\
python -m examples.delivery
"""
from __future__ import annotations
import asyncio
from decimal import Decimal
from iikocloudapi import CommandFailed, IikoCloud
from iikocloudapi.models import (
AddressCity,
CashPayment,
DeliveryOrder,
DeliveryPoint,
DeliveryServiceType,
ProductOrderItem,
RegularCustomer,
)
try:
from examples._env import client_from_env, env
except ImportError: # pragma: no cover
from _env import client_from_env, env # type: ignore[no-redef]
async def main(
iiko: IikoCloud,
*,
organization_id: str,
terminal_group_id: str,
product_id: str,
cash_payment_type_id: str,
) -> None:
price = Decimal("390.00")
order = DeliveryOrder(
phone="+79990000000",
order_service_type=DeliveryServiceType.DELIVERY_BY_COURIER,
customer=RegularCustomer(name="Anna"),
delivery_point=DeliveryPoint(
address=AddressCity(line1="Moscow, Tverskaya st. 1", flat="12", floor="3"),
comment="Entrance from the yard",
),
items=[ProductOrderItem(product_id=product_id, amount=Decimal(2), price=price)],
payments=[CashPayment(payment_type_id=cash_payment_type_id, sum=price * 2)],
comment="Please call 10 minutes before arrival",
)
created = await iiko.deliveries.create(
organization_id=organization_id, terminal_group_id=terminal_group_id, order=order
)
try:
await iiko.wait(created, organization_id=organization_id)
except CommandFailed as error:
print(f"The terminal rejected the order: {error}")
return
found = await iiko.deliveries.by_id(organization_id=organization_id, order_ids=[created.order_info.id])
delivery = found.orders[0].order
if delivery is not None:
print(f"Delivery #{delivery.number}: {delivery.status}, complete before {delivery.complete_before}")
if __name__ == "__main__":
async def run() -> None:
async with client_from_env() as iiko:
await main(
iiko,
organization_id=env("IIKO_ORGANIZATION_ID"),
terminal_group_id=env("IIKO_TERMINAL_GROUP_ID"),
product_id=env("IIKO_PRODUCT_ID"),
cash_payment_type_id=env("IIKO_PAYMENT_TYPE_ID"),
)
asyncio.run(run())
Menu sync¶
An external menu (v3) with stop list marks.
"""Download an external menu (v3) and mark the dishes that are in the stop list.
IIKO_ORGANIZATION_ID=... python -m examples.menu_sync
"""
from __future__ import annotations
import asyncio
from iikocloudapi import IikoCloud
try:
from examples._env import client_from_env, env
except ImportError: # pragma: no cover
from _env import client_from_env, env # type: ignore[no-redef]
async def out_of_stock(iiko: IikoCloud, organization_id: str) -> set[str]:
"""IDs of products that cannot be sold right now on any terminal group of the organization."""
response = await iiko.stop_lists(organization_ids=[organization_id])
return {
item.product_id
for organization in response.terminal_group_stop_lists
for terminal_group in organization.items
for item in terminal_group.items
if item.balance <= 0
}
async def main(iiko: IikoCloud, *, organization_id: str) -> None:
menus = await iiko.menu()
if not menus.external_menus:
print("No external menus are configured in iikoWeb")
return
external_menu = menus.external_menus[0]
price_category = menus.price_categories[0] if menus.price_categories else None
menu = await iiko.menu.by_id(
external_menu_id=external_menu.id,
organization_id=organization_id,
price_category_id=price_category.id if price_category else None,
)
stopped = await out_of_stock(iiko, organization_id)
print(f"Menu {menu.name!r}")
for group in menu.items_groups or []:
print(f" {group.name}")
for item in group.items or []:
prices = [p.price for p in item.size_prices or [] if p.price is not None]
price = f"{min(prices)}" if prices else "no price"
mark = " (stop list)" if item.product_id in stopped else ""
print(f" {item.name}: {price}{mark}")
if __name__ == "__main__":
async def run() -> None:
async with client_from_env() as iiko:
await main(iiko, organization_id=env("IIKO_ORGANIZATION_ID"))
asyncio.run(run())
Webhook receiver¶
A FastAPI application that receives and dispatches webhook events, and the code that registers it.
"""Receive iikoCloud webhooks with FastAPI.
Register the receiver once (``register()`` below), then run it:
IIKO_WEBHOOK_TOKEN=... uvicorn examples.webhook_receiver:app --port 8000
iikoCloud expects a ``200`` answer quickly, so do the heavy work in the background.
"""
from __future__ import annotations
import asyncio
import os
from fastapi import FastAPI, Request, Response
from iikocloudapi import IikoCloud, parse_webhook, verify_webhook_token
from iikocloudapi.models import (
DeliveryOrderWebHooksFilter,
DeliveryStatus,
OrderStatus,
TableOrderWebHookFilter,
WebHooksFilter,
WebHookShortFilter,
)
from iikocloudapi.webhooks import (
DeliveryOrderUpdateEvent,
StopListUpdateEvent,
TableOrderUpdateEvent,
UnknownWebhookEvent,
)
try:
from examples._env import client_from_env, env
except ImportError: # pragma: no cover
from _env import client_from_env, env # type: ignore[no-redef]
WEBHOOK_TOKEN = os.environ.get("IIKO_WEBHOOK_TOKEN", "change-me")
app = FastAPI()
@app.post("/iiko/webhooks")
async def receive(request: Request) -> Response:
if not verify_webhook_token(request.headers.get("Authorization"), WEBHOOK_TOKEN):
return Response(status_code=401)
for event in parse_webhook(await request.body()):
match event:
case DeliveryOrderUpdateEvent(event_info=info) if info and info.order:
print(f"delivery {info.id}: {info.order.status}")
case TableOrderUpdateEvent(event_info=info) if info and info.order:
print(f"table order {info.id}: {info.order.status}")
case StopListUpdateEvent(organization_id=organization_id):
print(f"stop list changed in {organization_id}")
case UnknownWebhookEvent(event_type=event_type):
print(f"event {event_type} is not known to this version of iikocloudapi")
case _:
pass
return Response(status_code=200)
async def register(iiko: IikoCloud, *, organization_id: str, url: str) -> None:
"""Point the organization's webhooks to ``url`` and choose which events to receive."""
await iiko.webhooks.update_settings(
organization_id=organization_id,
web_hooks_uri=url,
auth_token=WEBHOOK_TOKEN,
web_hooks_filter=WebHooksFilter(
delivery_order_filter=DeliveryOrderWebHooksFilter(
order_statuses=[DeliveryStatus.ON_WAY, DeliveryStatus.DELIVERED, DeliveryStatus.CANCELLED],
errors=True,
),
table_order_filter=TableOrderWebHookFilter(order_statuses=[OrderStatus.CLOSED], errors=True),
stop_list_update_filter=WebHookShortFilter(updates=True),
),
)
if __name__ == "__main__":
async def run() -> None:
async with client_from_env() as iiko:
await register(iiko, organization_id=env("IIKO_ORGANIZATION_ID"), url=env("IIKO_WEBHOOK_URL"))
print("Webhook settings updated")
asyncio.run(run())