Pyroski

Asynchronous Python client for Eroski Digital Services via the Tempos21 Mobile Gateway (xpas1.tempos21.com).

It provides clean, scoped access to digital purchase receipts, itemized line items, PDF receipt downloads, digital saving coupons (vales de ahorro), customer loyalty cards, store occupancy (aforos), and fuel prices.

Features

• Long-Lived Bearer Token: Logs in once using the mobile login flow and persists an OAuth2 Bearer token across runs (~68 years validity).
• Auto-Healing Token Refresh: Automatically detects 401 Unauthorized responses and re-authenticates transparently.
• Modern Scoped Interface: Logical resource scoping via api.tickets, api.coupons, api.customer, and api.stores.
• Digital Receipt Export: Fetch itemized receipt details, VAT breakdowns, or download original receipts in PDF format.
• Batch Coupon Activation: Activate all eligible savings coupons concurrently with throttled semaphores.
• Headless 2FA Support: Custom SMS handlers for headless or automated environments (e.g. bots, webhooks, Home Assistant).

Installation

Install using uv (recommended) or pip:

# Using uv
uv add pyroski

# Or using pip
pip install pyroski

Configuration & Initialization

Credentials can be passed explicitly in code or loaded automatically from environment variables:

export EROSKI_EMAIL="your_email@example.com"
export EROSKI_PASSWORD="your_password"
export EROSKI_TOKEN_PATH=".eroski_token.json"  # optional, default: .eroski_token.json

Initialization Options

The Api constructor supports several configuration options:

Api(
    email: str | None = None,
    password: str | None = None,
    token_path: str | Path | None = None,
    sms_handler: Callable[[], Awaitable[str] | str] | None = None,
    timeout: float = 30.0,
    api_version: str = "v20",
)

Examples

1. Standard Initialization (Environment Variables):

from pyroski import Api

# Reads EROSKI_EMAIL and EROSKI_PASSWORD from environment
api = Api()

2. Custom Token Location & Timeout:

api = Api(
    email="user@example.com",
    password="secretpassword",
    token_path="/var/data/eroski_token.json",
    timeout=45.0,
)

3. Headless / Automated SMS 2FA Handler:

async def fetch_sms_from_telegram_or_db() -> str:
    # Await code from external notification, webhook, or queue
    return await external_2fa_service.get_latest_code()


api = Api(
    email="user@example.com",
    password="secretpassword",
    sms_handler=fetch_sms_from_telegram_or_db,
)

Quickstart

import asyncio
from pyroski import Api


async def main():
    async with Api() as api:
        # 1. Customer profile
        customer = await api.customer.get()
        print(f"Customer: {customer.name} {customer.surname} | Card: {customer.club_card}")

        # 2. Latest receipts
        recent_tickets = await api.tickets.latest()
        for t in recent_tickets:
            print(f"Ticket #{t.ticket_number}: {t.total_amount}€ at {t.center_name}")

        # 3. Activate all coupons
        activated = await api.coupons.activate_all()
        print(f"Activated {activated} coupon(s)")


if __name__ == "__main__":
    asyncio.run(main())

API Endpoints

The API is organized into resource-scoped services:

1. Tickets (api.tickets)

Manage and retrieve digital purchase receipts:

# Download latest receipt as PDF
tickets = await api.tickets.latest()
if tickets:
    pdf_path = await api.tickets.save_pdf(tickets[0].ticket_id, "./receipt.pdf")
    print(f"Saved PDF receipt to {pdf_path}")

2. Coupons (api.coupons)

Retrieve and activate digital saving coupons (vales):

# List and activate all coupons
coupons = await api.coupons.all()
print(f"Found {len(coupons)} coupon(s)")

count = await api.coupons.activate_all(max_concurrent=3)
print(f"Activated {count} coupon(s)")

3. Customer Profile (api.customer)

Access customer loyalty account and membership data:

customer = await api.customer.get()
print(f"Club Card: {customer.club_card} | Favorite Store CECO: {customer.favourite_shop}")

4. Stores (api.stores)

Access live store metadata, affluence, and fuel pricing:

# Check store affluence
occupancy = await api.stores.influx("159")
print(f"Current affluence: {occupancy}")

Error Handling

Pyroski defines specific exception types for authentication and API errors:

• PyroskiError: Base exception for all library errors.
• AuthenticationError: Raised on invalid credentials or failed OAuth2 token exchange.
• TwoFactorAuthError: Raised when SMS 2FA is challenged or verification fails.