Metadata-Version: 2.4
Name: stellanow-sdk-python
Version: 0.1.5
Summary: Welcome to the StellaNow Python SDK. This SDK is designed to provide an easy-to-use interface for developers integrating their Python applications with the StellaNow Platform. The SDK communicates with the StellaNow Platform using the MQTT.
License: This project is licensed under the terms of the MIT license.
Author: Stella Systems
Author-email: help@stella.systems
Requires-Python: >=3.11,<4.0
Classifier: License :: Other/Proprietary License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Requires-Dist: loguru (>=0.7.2,<0.8.0)
Requires-Dist: nanoid (>=2.0.0,<3.0.0)
Requires-Dist: paho-mqtt (>=2.1.0,<3.0.0)
Requires-Dist: pydantic (>=2.10.6,<3.0.0)
Requires-Dist: stellanow-api-internals (>=0.0.5)
Requires-Dist: typing-extensions (>=4.10.0,<5.0.0)
Description-Content-Type: text/markdown

# StellaNowPythonSDK

## Introduction
Welcome to the StellaNow Python SDK. This SDK is designed to provide an easy-to-use interface for developers integrating their Python applications with the StellaNow Platform. The SDK communicates with the StellaNow Platform using the MQTT protocol over secure WebSockets or TCP, depending on the configuration.

## Key Features
- Automated connection handling (connection, disconnection, and reconnection)
- Message queuing with configurable overflow strategies to handle network instability
- Unlimited queue size by default (WARNING: can cause memory overflow - see Queue Configuration section)
- Configurable overflow handling: drop oldest, drop newest, or raise exceptions for critical data
- Authentication management (login and automatic token refreshing)
- Easy interface to send different types of messages
- Extensibility options for more specific needs
- Production-ready with robust error handling and validation

## Getting Started
Before you start integrating the SDK, ensure you have a Stella Now account.

## Installation
To integrate the StellaNowSDK into your project, follow these steps:

### Via pip
The easiest way to add the StellaNowSDK to your project is through pip, the Python package installer. Install the package from PyPI ([stellanow-sdk-python](https://pypi.org/project/stellanow-sdk-python/)) by running the following command in your terminal:
```bash
pip install stellanow-sdk-python
```

## Configuration
The SDK supports multiple authentication strategies when connecting to the Stella Now MQTT Sink. You can configure the SDK using OIDC authentication, username/password authentication, or no authentication, depending on your environment.

### MQTT Broker URL
The SDK uses an MQTT URL to connect to the StellaNow Platform or a local broker. The URL scheme dictates the transport and TLS settings:

- mqtt or mqtt-tcp: TCP transport, no TLS (default port: 1883).
- mqtts: TCP transport with TLS (default port: 8883).
- ws: WebSockets transport, no TLS (default port: 80).
- wss: WebSockets transport with TLS (default port: 443).

Each configuration function (e.g., `configure_dev_oidc_mqtt_fifo_sdk`) uses a default URL tailored to its environment (development, production, or local).


### Setting Up StellaNow SDK
To use the SDK, first ensure you have set the necessary environment variables:
- ORGANIZATION_ID
- PROJECT_ID
- For `OIDC`: OIDC_USERNAME, OIDC_PASSWORD
- For `Username/Password`: MQTT_USERNAME, MQTT_PASSWORD

Then, configure the SDK with the appropriate authentication strategy.

### Using OIDC Authentication
To authenticate with `StellaNow's` OIDC (OpenID Connect) on DEV environment using FIFO order, use `configure_dev_oidc_mqtt_fifo_sdk`:

```python
from stellanow_sdk_python.configure_sdk import configure_dev_oidc_mqtt_fifo_sdk

async def main():
    sdk = configure_dev_oidc_mqtt_fifo_sdk()
    await sdk.start()
```

This will:
* Authenticate with OIDC using the provided username and password, using a specific OIDC Client designed for data ingestion.
* Resulting token will be used in MQTT broker authentication with specific claim.
* Connect to the MQTT sink securely.
* Automatically refresh the access token before expiration using the refresh token.
* **Automatically recover** from expired refresh tokens by re-authenticating with username/password when needed.

### Using Username/Password Authentication
For scenarios requiring simple username/password authentication, use `configure_dev_basic_mqtt_lifo_sdk`:
```python
from stellanow_sdk_python.configure_sdk import configure_dev_basic_mqtt_lifo_sdk

async def main():
    sdk = configure_dev_basic_mqtt_lifo_sdk()
    await sdk.start()
```

This will:
- Use the USERNAME and PASSWORD environment variables for MQTT broker authentication.
- Connect to the MQTT sink with the specified transport and TLS settings.

### Using No Authentication
For local development or scenarios where authentication is not required, use configure_local_nanomq_basic_mqtt_fifo_sdk or configure_prod_none_mqtt_fifo_sdk:

```python
from stellanow_sdk_python.configure_sdk import configure_prod_none_mqtt_fifo_sdk

async def main():
    sdk = configure_prod_none_mqtt_fifo_sdk()
    await sdk.start()
```

This will:
- Connect to the MQTT sink without authentication (or with dummy credentials for local testing).
- Be useful for testing against local MQTT brokers like NanoMQ.

Ensure you have set the appropriate environment variables.

> **Note:** The client_id used in the MQTT connection must be unique per connection. If two connections use the same client_id, the first connection will be dropped. To prevent conflicts, the SDK automatically generates a unique client_id at startup in the format StellaNowSDKPython__<10-char-nanoid>. You can find the generated value in the logs:
```python
2025-03-14 13:18:36.802 | INFO     | stellanow_sdk_python.sinks.mqtt.stellanow_mqtt_sink:__init__:67 - SDK Client ID is "StellaNowSDKPython_oLMA971RWD"
```

### Configuring Queue Overflow Behavior

⚠️ **IMPORTANT: The SDK now uses an unlimited queue by default (max_size=0). This means the queue will NEVER drop messages, but can cause memory overflow if messages are produced faster than they are consumed. Monitor your application's memory usage carefully.**

The SDK provides configurable queue overflow strategies to handle different data criticality requirements. By default, the SDK uses an unlimited queue with RAISE_EXCEPTION strategy (which only applies if you set a max_size limit).

#### For Critical Data (e.g., ANPR, Financial Transactions)

When you cannot afford to lose any messages, use the `RAISE_EXCEPTION` strategy:

```python
from stellanow_sdk_python.configure_sdk import configure_sdk
from stellanow_sdk_python.config.eniviroment_config.stellanow_env_config import EnvConfig
from stellanow_sdk_python.config.enums.auth_strategy import AuthStrategyTypes
from stellanow_sdk_python.message_queue.message_queue_strategy.overflow_strategy import (
    OverflowStrategy,
    QueueFullError
)

# Configure for critical ANPR data
sdk = configure_sdk(
    auth_strategy_type=AuthStrategyTypes.OIDC.value,
    env_config=EnvConfig.stellanow_prod(),
    queue_max_size=200_000,  # Large queue for extended outages
    queue_overflow_strategy=OverflowStrategy.RAISE_EXCEPTION  # Never lose data silently
)

await sdk.start()

# In your processing loop
try:
    sdk.send_message(critical_event)
except QueueFullError as e:
    logger.critical(
        f"Queue full! Size: {e.queue_size}, "
        f"Already dropped: {e.dropped_count}, "
        f"Failed message: {e.message_id}"
    )
    # Implement your backup strategy:
    await write_to_backup_file(critical_event)
    # Or wait and retry
    await asyncio.sleep(5)
```

#### Available Overflow Strategies

- **RAISE_EXCEPTION** (default): Raises `QueueFullError` when queue is full. Best for critical data where application needs to decide what to do.
- **DROP_OLDEST**: Drops oldest messages when queue is full. Good for real-time data where newest matters most.
- **DROP_NEWEST**: Rejects new messages when queue is full. Good for preserving historical data.

#### Production Queue Recommendations

⚠️ **RECOMMENDED FOR PRODUCTION: Implement a persistent queue strategy.**

The built-in in-memory queues (FIFO/LIFO) are designed for development and testing. For production systems, you should implement a custom persistent queue that:
- Survives application restarts and crashes
- Persists messages to disk, database, or distributed queue system
- Provides guaranteed message delivery

See the [Implementing a Persistent Queue](#implementing-a-persistent-queue) section below for details.

#### Adjusting In-Memory Queue Size (Development/Testing)

If using the default in-memory queue, configure size limits to prevent memory overflow:

```python
# For high-throughput systems (e.g., ANPR with 100+ cameras)
sdk = configure_sdk(
    auth_strategy_type=AuthStrategyTypes.OIDC.value,
    env_config=EnvConfig.stellanow_prod(),
    queue_max_size=500_000,  # ~1.5 GB for metadata-only messages with S3 links
    queue_overflow_strategy=OverflowStrategy.RAISE_EXCEPTION  # Alert on queue full
)

# For typical environments (explicit size + strategy)
sdk = configure_sdk(
    auth_strategy_type=AuthStrategyTypes.OIDC.value,
    env_config=EnvConfig.stellanow_prod(),
    queue_max_size=100_000,  # ~300 MB for metadata-only messages
    queue_overflow_strategy=OverflowStrategy.DROP_OLDEST  # Or use default RAISE_EXCEPTION
)

# For low-memory environments
sdk = configure_sdk(
    auth_strategy_type=AuthStrategyTypes.OIDC.value,
    env_config=EnvConfig.stellanow_prod(),
    queue_max_size=10_000  # ~30 MB
)

# For unlimited queue (default - use ONLY for development/testing!)
sdk = configure_sdk(
    auth_strategy_type=AuthStrategyTypes.OIDC.value,
    env_config=EnvConfig.stellanow_prod(),
    queue_max_size=0  # No limit - can cause out of memory errors!
)
```

> **Memory Usage:** For messages containing only metadata and S3 links (no embedded images), expect ~3 KB per message. A 100,000-message queue uses approximately 300 MB of memory. An unlimited queue will grow without bound and can exhaust system memory.
>
> **⚠️ Important:** In-memory queues lose all unsent messages if the application terminates unexpectedly. For production systems requiring message durability, implement a persistent queue strategy.

## Sample Application
Here is a simple application that uses StellaNowSDK to send user details messages to the Stella Now platform.

This script is the main entry point for our demonstration. The main function does a few things:
- It establishes a connection to StellaNow.
- It creates a UserDetailsMessage and sends it using the StellaNow SDK.
- It checks and logs the queue status.
- Finally, it disconnects from StellaNow.
```python
import asyncio
import sys
from stellanow_sdk_python.configure_sdk import configure_dev_oidc_mqtt_fifo_sdk
from stellanow_sdk_python_demo.messages.models.phone_number_model import PhoneNumberModel
from stellanow_sdk_python_demo.messages.user_details_update_message import UserDetailsUpdateMessage
from loguru import logger

async def main():
    """Main entry point for the StellaNow SDK demo."""
    sdk = configure_dev_oidc_mqtt_fifo_sdk()
    shutdown_event = asyncio.Event()  # Event to signal shutdown

    try:
        await sdk.start()
        for i in range(10):
            message = UserDetailsUpdateMessage(
                patron="12345", user_id=f"user_{i}", phone_number=PhoneNumberModel(country_code=44, number=753594 + i)
            )
            logger.info(f"Sending message {i + 1}...")
            await sdk.send_message(message)
            await asyncio.sleep(2)
        logger.info("Initial messages sent. Keeping SDK alive...")
        await shutdown_event.wait()  # Wait indefinitely until shutdown
    
    except KeyboardInterrupt:
        logger.warning("Received Ctrl+C, shutting down gracefully...")
        shutdown_event.set()  # Signal shutdown
        await sdk.stop()
    except Exception as e:
        logger.error(f"Unexpected error occurred: {e}")
        await sdk.stop()
        sys.exit(1)

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

### Setting the Message Origin Timestamp

The `message_origin_date_utc` field in a `StellaNowMessageBase`-derived message specifies the UTC timestamp when the message was created. This field is optional but highly recommended for accurate event tracking and auditing in the StellaNow Platform. The timestamp must be in ISO 8601 format with microseconds and a `Z` suffix (e.g., `2025-04-25T12:34:56.789012Z`) when serialized.

#### Requirements
- **Type**: Python `datetime` object or ISO 8601 string.
- **Format**: If provided as a string, it must include microseconds and end with `Z` (e.g., `2025-04-25T12:34:56.789012Z`).
- **Timezone**: Must represent UTC time. Non-UTC `datetime` objects will raise a validation error.
- **Validation**: The SDK validates the timestamp to ensure it meets these requirements. Invalid formats will raise a `ValueError`.

#### Setting `message_origin_date_utc`
Set `message_origin_date_utc` after creating a message by assigning a `datetime` object or a valid ISO 8601 string to the `message_origin_date_utc` property. If not set, the field remains `None` in the message object, but the SDK automatically assigns the current UTC time when the message is wrapped in a `StellaNowMessageWrapper` (e.g., during `sdk.send_message`).
```python
import asyncio
import sys
from stellanow_sdk_python.configure_sdk import configure_dev_oidc_mqtt_fifo_sdk
from stellanow_sdk_python_demo.messages.models.phone_number_model import PhoneNumberModel
from stellanow_sdk_python_demo.messages.user_details_update_message import UserDetailsUpdateMessage
from loguru import logger

async def main():
    """Main entry point for the StellaNow SDK demo."""
    sdk = configure_dev_oidc_mqtt_fifo_sdk()
    shutdown_event = asyncio.Event()  # Event to signal shutdown

    try:
        await sdk.start()
        for i in range(10):
            message = UserDetailsUpdateMessage(
                patron="12345", user_id=f"user_{i}", phone_number=PhoneNumberModel(country_code=44, number=753594 + i)
            )
            message.message_origin_date_utc = "2025-04-25T12:34:56.789012Z"
            logger.info(f"Sending message {i + 1}...")
            await sdk.send_message(message)
            await asyncio.sleep(2)
        logger.info("Initial messages sent. Keeping SDK alive...")
        await shutdown_event.wait()  # Wait indefinitely until shutdown
    
    except KeyboardInterrupt:
        logger.warning("Received Ctrl+C, shutting down gracefully...")
        shutdown_event.set()  # Signal shutdown
        await sdk.stop()
    except Exception as e:
        logger.error(f"Unexpected error occurred: {e}")
        await sdk.stop()
        sys.exit(1)

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

The full code for the sample application can be found here:
- [Basic Application](stellanow_sdk_python_demo/main.py)

### IMPORTANT

> **`stop` METHOD USAGE**
>
>The stop method in the StellaNowSDK is asynchronous and stops the SDK's message processing immediately. When called as await sdk.stop(), it halts the processing of the message queue and disconnects from the MQTT broker.
>
>This behavior can have important consequences when using non-persistent queue implementations (like the default in-memory FIFO or LIFO queues). Any messages that have been added to the queue but not yet sent will remain unsent until start is called again.
>
>If your application shuts down before start is called again, those unsent messages will be lost because non-persistent queues do not store their contents when the application terminates.

## Message Formatting
Messages in StellaNowSDK are wrapped in a StellaNowEventWrapper, and each specific message type extends this class to define its own properties. Each message needs to follow a certain format, including a type, list of entities, and optional fields. Here is an example:

```python
from pydantic import Field
from stellanow_sdk_python_demo.messages.models.phone_number_model import PhoneNumberModel
from stellanow_sdk_python.messages.message import Entity, StellaNowMessageBase


class UserDetailsUpdateMessage(StellaNowMessageBase):
    user_id: str = Field(None, serialization_alias="user_id")
    phone_number: PhoneNumberModel = Field(None, serialization_alias="phone_number")

    def __init__(self, patron: str, user_id: str, phone_number: PhoneNumberModel):
        super().__init__(
            event_name="user_details_update",
            entities=[Entity(entity_type_definition_id="patron", entity_id=patron)],
            user_id=user_id,
            phone_number=phone_number,
        )
 ```

Creating these classes by hand can be prone to errors. Therefore, we provide a command line interface (CLI) tool, [stellanow-cli](https://pypi.org/project/stellanow-cli/), to automate this task. This tool generates the code for the message classes automatically based on the configuration defined in the Operators Console.

You can install the StellaNow CLI tool using pip:
```bash
pip install stellanow-cli
```
Once you have installed the **StellaNow CLI** tool, you can use it to generate message classes. Detailed instructions on how to use the **StellaNow CLI** tool can be found in the tool's documentation.

Please note that it is discouraged to write these classes yourself. Using the CLI tool ensures that the message format aligns with the configuration defined in the Operators Console and reduces the potential for errors.

## Customization

StellaNowSDK provides extensive flexibility for developers to adapt the SDK to their specific needs. You can extend key components, including message queuing strategies, sinks (where messages are sent), connection strategies, and authentication mechanisms.

### Customizing the Message Queue Strategy
By default, `StellaNowPythonSDK` uses an in-memory queue to temporarily hold messages before sending them to a sink. The SDK provides built-in FIFO and LIFO queue strategies with configurable overflow behavior:

- **Default Size:** 0 (unlimited) - ⚠️ **WARNING: Can cause memory overflow in production!**
- **Default Overflow Strategy:** RAISE_EXCEPTION (raises error when queue reaches limit)
- **Alternative Strategies:** DROP_OLDEST, DROP_NEWEST
- **Configurable via:** `configure_sdk()` parameters: `queue_max_size` and `queue_overflow_strategy`
- **Recommendation:** Always set an explicit `queue_max_size` in production environments

These non-persistent queues will lose all messages if the application terminates unexpectedly.

#### Implementing a Persistent Queue
If your application requires a persistent queue that survives restarts or crashes, you can implement a custom queue strategy by extending `IMessageQueueStrategy` and integrating it with a database, file system, or distributed queue system.

Your custom strategy should implement:
- `enqueue(message)` - Add message to queue
- `try_dequeue()` - Remove and return message from queue
- `is_empty()` - Check if queue is empty
- `get_message_count()` - Return current queue size

Example integration:
```python
from stellanow_sdk_python.message_queue.message_queue_strategy.i_message_queue_strategy import IMessageQueueStrategy

class DatabaseQueueStrategy(IMessageQueueStrategy):
    def __init__(self, db_connection):
        self.db = db_connection

    def enqueue(self, message):
        self.db.insert("queue", message.model_dump_json())

    def try_dequeue(self):
        row = self.db.fetch_oldest("queue")
        if row:
            self.db.delete("queue", row.id)
            return StellaNowEventWrapper.model_validate_json(row.data)
        return None
    # ... implement other methods
```

>⚠️ **Performance Considerations:** Persistent queues introduce additional latency and require careful design to balance reliability and performance.

#### Adding a Custom Sink & Connection Strategy
A sink is where messages are ultimately delivered. StellaNowSDK supports MQTT-based sinks, but you can extend this to support Kafka, Webhooks, Databases, or any custom integration.

Each sink is paired with a connection strategy, which determines how it establishes a connection. You can implement a custom connection strategy for different protocols or authentication mechanisms.

##### Example: Adding a Custom Sink
To add a new sink, implement `IStellaNowSink` and update configure_sdk.py:
```python
from stellanow_sdk_python.sinks.custom_sink import CustomSink

mqtt_sink = CustomSink(auth_strategy=auth_strategy, env_config=env_config, project_info=project_info)
```
##### Extending the Authentication Service
StellaNowSDK supports multiple authentication mechanisms, including:
- OIDC-based authentication
- Username/password authentication
- No authentication

If your project requires a new authentication mechanism, implement IMqttAuthStrategy and register it in configure_sdk.py:

```python
from stellanow_sdk_python.sinks.mqtt.auth_strategy.custom_auth_strategy import CustomAuthStrategy

auth_strategy = CustomAuthStrategy(project_info, credentials, env_config)
```
> ⚠️ **Security Considerations:** Ensure your authentication mechanism is properly secured and does not expose credentials in logs or configuration files.

## Support
For any issues or feature requests, feel free to create a new issue on our GitHub repository. If you need further assistance, contact our support team at help@stella.systems.

## Documentation
Detailed documentation will be available soon.

## License
This project is licensed under the terms of the MIT license.
