Recommendation Engine Python Client

Official Python client library for the GrooveShop Recommendation Engine API.

Features

•  Full type hints with comprehensive type definitions
•  Modern Python 3.14+ with latest features
•  Async/await support with httpx
•  Automatic error handling with custom exceptions
•  Support for all API endpoints
•  Authentication support (API key)
•  Multi-tenancy support
•  Batch operations support
•  Context manager support for resource cleanup
•  Fully tested with pytest (85%+ coverage)
•  Type-checked with mypy
•  Linted with ruff

Installation

pip install recommendation-engine-client

Or with uv (recommended):

uv add recommendation-engine-client

Quick Start

from recommendation_engine_client import RecommendationClient

# Initialize the client
async with RecommendationClient(
    base_url="http://localhost:8080",
    api_key="your-api-key",  # Optional
    timeout=30.0,  # Optional, default: 30.0
) as client:
    # Get recommendations for a user
    recommendations = await client.get_user_recommendations(
        "user_123",
        {"algorithm": "hybrid", "count": 10}
    )

    print(recommendations["recommendations"])

Usage

Client Initialization

from recommendation_engine_client import RecommendationClient

# Basic initialization
client = RecommendationClient(base_url="https://api.example.com")

# With API key authentication
client = RecommendationClient(
    base_url="https://api.example.com",
    api_key="your-secret-api-key",
)

# With custom configuration
client = RecommendationClient(
    base_url="https://api.example.com",
    api_key="your-secret-api-key",
    timeout=60.0,
    headers={
        "X-Custom-Header": "value",
    },
)

# Using as context manager (recommended)
async with RecommendationClient(base_url="https://api.example.com") as client:
    # Use client
    pass

Entity Operations

Create an Entity

entity = await client.create_entity({
    "entity_id": "product_1",
    "entity_type": "product",
    "attributes": {
        "name": "Wireless Headphones",
        "category": "Electronics",
        "price": 99.99,
        "brand": "TechPro",
        "in_stock": True,
        "tags": ["wireless", "audio", "bluetooth"],
    },
    "tenant_id": "tenant_a",  # Optional
})

Get an Entity

entity = await client.get_entity("product_1", tenant_id="tenant_a")
print(entity["attributes"])

Update an Entity

updated_entity = await client.update_entity(
    "product_1",
    {
        "attributes": {
            "price": 89.99,
            "in_stock": False,
        },
        "tenant_id": "tenant_a",
    },
)

Delete an Entity

await client.delete_entity("product_1", tenant_id="tenant_a")

Bulk Import Entities

result = await client.bulk_import_entities({
    "entities": [
        {
            "entity_id": "product_1",
            "entity_type": "product",
            "attributes": {
                "name": "Product 1",
                "price": 29.99,
            },
        },
        {
            "entity_id": "product_2",
            "entity_type": "product",
            "attributes": {
                "name": "Product 2",
                "price": 39.99,
            },
        },
    ],
    "tenant_id": "tenant_a",
})

print(f"Imported {result['successful']}/{result['total_records']} entities")

Interaction Operations

Create an Interaction

interaction = await client.create_interaction({
    "user_id": "user_123",
    "entity_id": "product_1",
    "entity_type": "product",
    "interaction_type": "purchase",
    "metadata": {
        "source": "web",
        "device": "desktop",
    },
    "tenant_id": "tenant_a",
})

Get User Interactions

interactions = await client.get_user_interactions(
    "user_123",
    limit=50,
    offset=0,
    tenant_id="tenant_a",
)

for interaction in interactions:
    print(f"{interaction['user_id']} -> {interaction['entity_id']}: {interaction['interaction_type']}")

Bulk Import Interactions

result = await client.bulk_import_interactions({
    "interactions": [
        {
            "user_id": "user_1",
            "entity_id": "product_1",
            "entity_type": "product",
            "interaction_type": "view",
        },
        {
            "user_id": "user_1",
            "entity_id": "product_2",
            "entity_type": "product",
            "interaction_type": "purchase",
        },
    ],
    "tenant_id": "tenant_a",
})

Recommendation Operations

Get User Recommendations

# Hybrid recommendations (default)
recommendations = await client.get_user_recommendations(
    "user_123",
    {
        "algorithm": "hybrid",
        "count": 10,
        "tenant_id": "tenant_a",
    },
)

# Collaborative filtering
collab_recs = await client.get_user_recommendations(
    "user_123",
    {"algorithm": "collaborative", "count": 20},
)

# Content-based filtering
content_recs = await client.get_user_recommendations(
    "user_123",
    {"algorithm": "content_based", "count": 15},
)

for rec in recommendations["recommendations"]:
    print(f"{rec['entity_id']}: score={rec['score']}, reason={rec.get('reason')}")

if recommendations["cold_start"]:
    print("User has few interactions, showing trending items")

Get Similar Entities

similar = await client.get_similar_entities(
    "product_1",
    {
        "algorithm": "content_based",
        "count": 10,
        "entity_type": "product",
        "tenant_id": "tenant_a",
    },
)

for item in similar["recommendations"]:
    print(f"Similar to product_1: {item['entity_id']} (score: {item['score']})")

Get Trending Entities

trending = await client.get_trending_entities({
    "entity_type": "product",
    "count": 20,
    "tenant_id": "tenant_a",
})

for index, item in enumerate(trending["trending"], 1):
    print(f"#{index}: {item['entity_id']} (score: {item['score']})")

Health Checks

# Check if API is healthy
is_healthy = await client.is_healthy()
print(f"API healthy: {is_healthy}")

# Check if API is ready (db + redis connected)
is_ready = await client.is_ready()
print(f"API ready: {is_ready}")

Error Handling

The client provides custom exceptions for better error handling:

from recommendation_engine_client import (
    RecommendationClient,
    RecommendationError,
    TimeoutError,
    NetworkError,
)

async with RecommendationClient(base_url="http://localhost:8080") as client:
    try:
        entity = await client.get_entity("non_existent_id")
    except RecommendationError as error:
        print(f"API Error [{error.code}]: {error.message}")
        if error.details:
            print(f"Details: {error.details}")
    except TimeoutError as error:
        print(f"Request timeout: {error}")
    except NetworkError as error:
        print(f"Network error: {error}")

Type Hints

This library is fully typed and provides comprehensive type definitions:

from recommendation_engine_client import (
    Entity,
    Interaction,
    RecommendationResponse,
    ScoredEntity,
    InteractionType,
)

def process_recommendations(response: RecommendationResponse) -> None:
    """Process recommendations with full type safety."""
    for item in response["recommendations"]:
        # item is fully typed as ScoredEntity
        print(item["entity_id"], item["score"])

Multi-Tenancy

The client supports multi-tenancy. You can specify a tenant_id in most operations:

# Create entity for tenant A
await client.create_entity({
    "entity_id": "product_1",
    "entity_type": "product",
    "attributes": {"name": "Product"},
    "tenant_id": "tenant_a",
})

# Get recommendations for tenant A
recs = await client.get_user_recommendations(
    "user_123",
    {"tenant_id": "tenant_a"},
)

Batch Operations

For importing large amounts of data, use the bulk import methods:

# Prepare data
entities = [
    {
        "entity_id": f"product_{i}",
        "entity_type": "product",
        "attributes": {
            "name": f"Product {i}",
            "price": i * 10.0,
        },
    }
    for i in range(1000)
]

# Import in batches
batch_size = 100
for i in range(0, len(entities), batch_size):
    batch = entities[i:i + batch_size]
    result = await client.bulk_import_entities({
        "entities": batch,
        "tenant_id": "tenant_a",
    })
    print(f"Batch {i // batch_size + 1}: {result['successful']}/{result['total_records']} successful")

API Reference

RecommendationClient Methods

Entity Operations

• create_entity(request: CreateEntityRequest) -> Entity
• get_entity(entity_id: str, tenant_id: str | None = None) -> Entity
• update_entity(entity_id: str, request: UpdateEntityRequest) -> Entity
• delete_entity(entity_id: str, tenant_id: str | None = None) -> None
• bulk_import_entities(request: BulkImportEntitiesRequest) -> BulkImportResponse

Interaction Operations

• create_interaction(request: CreateInteractionRequest) -> Interaction
• get_user_interactions(user_id: str, *, limit: int | None = None, offset: int | None = None, tenant_id: str | None = None) -> list[Interaction]
• bulk_import_interactions(request: BulkImportInteractionsRequest) -> BulkImportResponse

Recommendation Operations

• get_user_recommendations(user_id: str, query: UserRecommendationsQuery | None = None) -> RecommendationResponse
• get_similar_entities(entity_id: str, query: EntityRecommendationsQuery | None = None) -> RecommendationResponse
• get_trending_entities(query: TrendingEntitiesQuery | None = None) -> TrendingEntitiesResponse

Health & Status

• is_healthy() -> bool
• is_ready() -> bool

Examples

See the examples/ directory for complete examples:

• examples/basic_usage.py - Basic client usage
• examples/e_commerce.py - E-commerce recommendation flow
• examples/content_platform.py - Content recommendation flow
• examples/bulk_import.py - Bulk data import

Requirements

• Python >= 3.14
• httpx >= 0.28.1
• typing-extensions >= 4.12.2

Development

Setup

# Clone the repository
git clone https://github.com/grooveshop/recommendation-engine
cd recommendation-engine/clients/python

# Install dependencies with uv
uv sync --dev

Running Tests

# Run all tests
uv run pytest

# Run with coverage
uv run pytest --cov

# Run with verbose output
uv run pytest -v

Linting and Formatting

# Check code with ruff
uv run ruff check src/

# Fix issues automatically
uv run ruff check --fix src/

# Format code
uv run ruff format src/

Type Checking

# Type check with mypy
uv run mypy src/

Modern Stack

This client uses a modern Python stack:

• Python 3.14: Latest Python with modern type hints
• httpx: Modern async HTTP client
• uv: Fast Python package installer and resolver
• ruff: Lightning-fast linter and formatter
• pytest: Modern testing framework
• mypy: Static type checker

License

MIT

Support

For issues and questions:

• GitHub Issues: https://github.com/grooveshop/recommendation-engine/issues
• Documentation: https://docs.grooveshop.com/recommendation-engine

Contributing

Contributions are welcome! Please read our contributing guidelines before submitting PRs.