# Platform Observability Python SDK
A Python SDK for ingesting logs and interacting with the Platform Observability service by Red Duck Labs.
## Installation
### From PyPI (after publishing)
```bash
pip install redducklabs-platform-observability-sdk
```
### From Source (Development)
```bash
cd sdks/python
pip install -e .
```
## Quick Start
### Basic Usage
```python
from platform_observability import ObservabilityClient, LogEntry, LogLevel
# Initialize client
client = ObservabilityClient(
api_key="YOUR_API_KEY",
base_url="https://observability.redducklabs.com/api/v1"
)
# Send a single log
client.ingest_log(
message="Application started",
level=LogLevel.INFO,
source="my-application",
labels={"app": "my-app", "environment": "production"},
fields={"version": "1.0.0", "user_id": "user123"}
)
```
### Using the ObservabilityLogger (Recommended)
The `ObservabilityLogger` provides a higher-level interface with automatic batching and Python logging integration:
```python
from platform_observability import ObservabilityLogger
# Initialize logger with labels for Grafana filtering
logger = ObservabilityLogger(
api_key="YOUR_API_KEY",
name="my-logger",
base_url="https://observability.redducklabs.com/api/v1",
source="my-application",
labels={
"app": "my-app",
"environment": "production",
"version": "1.0.0"
},
batch_size=100, # Batch up to 100 logs
flush_interval=5 # Auto-flush every 5 seconds
)
# Log messages with different levels
logger.info("User logged in", user_id="user123", ip="192.168.1.1")
logger.warning("High memory usage", memory_percent=85.5)
logger.error("Failed to connect to database", error_code="DB_CONN_FAILED")
# Ensure all logs are sent before exit
logger.flush()
logger.close()
```
## Configuration
### Environment Variables
You can configure the SDK using environment variables:
```bash
# Required
export RDL_API_KEY="your_api_key_here"
# Optional
export RDL_LOG_ENDPOINT="https://observability.redducklabs.com/api/v1"
export RDL_BATCH_SIZE="100"
export RDL_FLUSH_INTERVAL="5.0"
```
Then use them in your code:
```python
import os
from platform_observability import ObservabilityLogger
logger = ObservabilityLogger(
api_key=os.getenv('RDL_API_KEY'),
base_url=os.getenv('RDL_LOG_ENDPOINT', 'https://observability.redducklabs.com/api/v1'),
source="my-app",
labels={"app": "my-app", "environment": os.getenv('ENVIRONMENT', 'dev')}
)
```
## Important Concepts
### Labels vs Fields
**Labels** (low-cardinality, indexed):
- Used for filtering and querying in Grafana
- Should have limited unique values
- Examples: `app`, `environment`, `version`, `region`
**Fields** (high-cardinality, not indexed):
- Additional context for debugging
- Can have many unique values
- Examples: `user_id`, `request_id`, `duration_ms`
### Log Format
Logs are sent in this format:
```json
{
"message": "User action completed",
"level": "info",
"source": "my-app",
"labels": {
"app": "my-app",
"environment": "production"
},
"fields": {
"user_id": "user123",
"action": "purchase",
"amount": 99.99
},
"timestamp": "2024-01-25T10:30:00Z"
}
```
## Querying Logs in Grafana
Once logs are ingested, query them in Grafana using LogQL:
```logql
# All logs from your app
{app="my-app"}
# Filter by environment
{app="my-app", environment="production"}
# Search for errors
{app="my-app"} |= "error"
# Parse JSON and filter
{app="my-app"} | json | user_id="user123"
```
## Advanced Usage
### Custom Batching
```python
client = ObservabilityClient(
api_key="YOUR_API_KEY",
base_url="https://observability.redducklabs.com/api/v1",
batch_size=200, # Larger batches
batch_timeout=10.0, # Wait longer before sending
enable_batching=True # Enable automatic batching
)
# Add logs to batch queue
for i in range(100):
client.add_to_batch(
message=f"Log entry {i}",
level=LogLevel.INFO,
source="batch-test"
)
# Manually flush when needed
client.flush_batch()
```
### Error Handling
```python
from platform_observability import ObservabilityClient
from platform_observability.exceptions import (
AuthenticationError,
RateLimitError,
ValidationError
)
client = ObservabilityClient(
api_key="YOUR_API_KEY",
skip_connection_check=True # Skip initial health check
)
try:
client.ingest_log("Test message", level="info")
except AuthenticationError:
print("Invalid API key")
except RateLimitError:
print("Rate limit exceeded, retry later")
except ValidationError as e:
print(f"Invalid log format: {e}")
```
### Integration with Python Logging
```python
import logging
from platform_observability import ObservabilityHandler
# Create custom handler
handler = ObservabilityHandler(
api_key="YOUR_API_KEY",
base_url="https://observability.redducklabs.com/api/v1",
source="my-app",
labels={"app": "my-app", "environment": "dev"}
)
# Add to Python logger
logger = logging.getLogger(__name__)
logger.addHandler(handler)
logger.setLevel(logging.INFO)
# Now regular Python logging will be sent to Platform Observability
logger.info("This goes to Platform Observability")
logger.error("This error too", extra={"user_id": "user123"})
```
## API Key Management
```python
# List API keys
keys = client.list_api_keys()
for key in keys:
print(f"{key.name}: {key.masked_key}")
# Create new API key
new_key = client.create_api_key(
name="ci-pipeline",
description="API key for CI/CD pipeline"
)
print(f"New key created: {new_key.key}")
# Rotate API key
rotated = client.rotate_api_key(key_id="key_123")
print(f"New key value: {rotated['key']}")
# Delete API key
client.delete_api_key(key_id="key_123")
```
## Usage Analytics
```python
from datetime import datetime, timedelta
# Get usage metrics
end_time = datetime.now()
start_time = end_time - timedelta(days=7)
metrics = client.get_usage_analytics(
start_time=start_time,
end_time=end_time
)
print(f"Total logs ingested: {metrics.total_logs_ingested}")
print(f"Data volume: {metrics.total_data_volume_gb:.2f} GB")
print(f"Average logs per request: {metrics.average_logs_per_request:.1f}")
```
## CLI Tool
After installation, use the `obs-cli` command-line tool:
```bash
# Set up configuration
obs-cli config set api_key YOUR_API_KEY
obs-cli config set endpoint https://observability.redducklabs.com/api/v1
# Send a log
obs-cli log "Application deployed" --level info --label app=my-app
# List API keys
obs-cli keys list
# Get usage statistics
obs-cli usage --days 7
```
## Development
### Requirements
- Python >= 3.8
- pip
### Setup Development Environment
```bash
# Clone repository
git clone https://github.com/redducklabs/platform-observability.git
cd platform-observability/sdks/python
# Install in development mode with dependencies
pip install -e .[dev]
```
### Running Tests
```bash
# Run all tests
pytest
# Run with coverage
pytest --cov=platform_observability
# Run specific test file
pytest tests/test_client.py
```
### Building Package
```bash
# Install build tools
pip install build
# Build wheel and source distribution
python -m build
# Output will be in dist/
ls dist/
```
### Code Quality
```bash
# Format code
black platform_observability tests
# Lint code
flake8 platform_observability tests
# Type checking
mypy platform_observability
```
## Troubleshooting
### Logs Not Appearing in Grafana?
1. **Verify API Key**: Ensure your API key is valid and active
2. **Check Endpoint**: Confirm you're using the correct endpoint URL
3. **Validate Labels**: Ensure labels are properly set (required for Grafana queries)
4. **Flush Logs**: Call `logger.flush()` to ensure batched logs are sent
5. **Check Errors**: Enable debug logging to see any errors:
```python
import logging
logging.basicConfig(level=logging.DEBUG)
```
### Connection Errors?
If you're getting connection errors during initialization:
```python
client = ObservabilityClient(
api_key="YOUR_API_KEY",
base_url="https://observability.redducklabs.com/api/v1",
skip_connection_check=True # Skip initial health check
)
```
### Rate Limiting?
The SDK automatically retries on rate limit errors. You can also check rate limit status:
```python
# After making requests
rate_limit_info = client.get_rate_limit_info()
print(f"Remaining requests: {rate_limit_info.get('remaining')}")
print(f"Reset time: {rate_limit_info.get('reset')}")
```
## Examples
See the `examples/` directory for complete working examples:
- `examples/sample-application/sdk_only_example.py` - Basic SDK usage
- `examples/sample-application/sdk_continuous_example.py` - Continuous log generation
- `examples/sample-application/app.py` - FastAPI integration example
## Support
For issues, questions, or contributions:
- GitHub Issues: https://github.com/redducklabs/platform-observability/issues
- Documentation: https://docs.redducklabs.com/observability
## License
MIT License - see LICENSE file for details
Raw data
{
"_id": null,
"home_page": "https://github.com/redducklabs/platform-observability",
"name": "redducklabs-platform-observability-sdk",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.8",
"maintainer_email": null,
"keywords": "observability, logging, monitoring, analytics, platform",
"author": "Red Duck Labs",
"author_email": "Red Duck Labs <support@redducklabs.com>",
"download_url": "https://files.pythonhosted.org/packages/a4/ca/33f3a82dfd3cc827cc96924cc61e9a7da675e02627e35d8f4ca5be1a4337/redducklabs_platform_observability_sdk-1.0.2.tar.gz",
"platform": null,
"description": "# Platform Observability Python SDK\n\nA Python SDK for ingesting logs and interacting with the Platform Observability service by Red Duck Labs.\n\n## Installation\n\n### From PyPI (after publishing)\n```bash\npip install redducklabs-platform-observability-sdk\n```\n\n### From Source (Development)\n```bash\ncd sdks/python\npip install -e .\n```\n\n## Quick Start\n\n### Basic Usage\n\n```python\nfrom platform_observability import ObservabilityClient, LogEntry, LogLevel\n\n# Initialize client\nclient = ObservabilityClient(\n api_key=\"YOUR_API_KEY\",\n base_url=\"https://observability.redducklabs.com/api/v1\"\n)\n\n# Send a single log\nclient.ingest_log(\n message=\"Application started\",\n level=LogLevel.INFO,\n source=\"my-application\",\n labels={\"app\": \"my-app\", \"environment\": \"production\"},\n fields={\"version\": \"1.0.0\", \"user_id\": \"user123\"}\n)\n```\n\n### Using the ObservabilityLogger (Recommended)\n\nThe `ObservabilityLogger` provides a higher-level interface with automatic batching and Python logging integration:\n\n```python\nfrom platform_observability import ObservabilityLogger\n\n# Initialize logger with labels for Grafana filtering\nlogger = ObservabilityLogger(\n api_key=\"YOUR_API_KEY\",\n name=\"my-logger\",\n base_url=\"https://observability.redducklabs.com/api/v1\",\n source=\"my-application\",\n labels={\n \"app\": \"my-app\",\n \"environment\": \"production\",\n \"version\": \"1.0.0\"\n },\n batch_size=100, # Batch up to 100 logs\n flush_interval=5 # Auto-flush every 5 seconds\n)\n\n# Log messages with different levels\nlogger.info(\"User logged in\", user_id=\"user123\", ip=\"192.168.1.1\")\nlogger.warning(\"High memory usage\", memory_percent=85.5)\nlogger.error(\"Failed to connect to database\", error_code=\"DB_CONN_FAILED\")\n\n# Ensure all logs are sent before exit\nlogger.flush()\nlogger.close()\n```\n\n## Configuration\n\n### Environment Variables\n\nYou can configure the SDK using environment variables:\n\n```bash\n# Required\nexport RDL_API_KEY=\"your_api_key_here\"\n\n# Optional\nexport RDL_LOG_ENDPOINT=\"https://observability.redducklabs.com/api/v1\"\nexport RDL_BATCH_SIZE=\"100\"\nexport RDL_FLUSH_INTERVAL=\"5.0\"\n```\n\nThen use them in your code:\n\n```python\nimport os\nfrom platform_observability import ObservabilityLogger\n\nlogger = ObservabilityLogger(\n api_key=os.getenv('RDL_API_KEY'),\n base_url=os.getenv('RDL_LOG_ENDPOINT', 'https://observability.redducklabs.com/api/v1'),\n source=\"my-app\",\n labels={\"app\": \"my-app\", \"environment\": os.getenv('ENVIRONMENT', 'dev')}\n)\n```\n\n## Important Concepts\n\n### Labels vs Fields\n\n**Labels** (low-cardinality, indexed):\n- Used for filtering and querying in Grafana\n- Should have limited unique values\n- Examples: `app`, `environment`, `version`, `region`\n\n**Fields** (high-cardinality, not indexed):\n- Additional context for debugging\n- Can have many unique values\n- Examples: `user_id`, `request_id`, `duration_ms`\n\n### Log Format\n\nLogs are sent in this format:\n```json\n{\n \"message\": \"User action completed\",\n \"level\": \"info\",\n \"source\": \"my-app\",\n \"labels\": {\n \"app\": \"my-app\",\n \"environment\": \"production\"\n },\n \"fields\": {\n \"user_id\": \"user123\",\n \"action\": \"purchase\",\n \"amount\": 99.99\n },\n \"timestamp\": \"2024-01-25T10:30:00Z\"\n}\n```\n\n## Querying Logs in Grafana\n\nOnce logs are ingested, query them in Grafana using LogQL:\n\n```logql\n# All logs from your app\n{app=\"my-app\"}\n\n# Filter by environment\n{app=\"my-app\", environment=\"production\"}\n\n# Search for errors\n{app=\"my-app\"} |= \"error\"\n\n# Parse JSON and filter\n{app=\"my-app\"} | json | user_id=\"user123\"\n```\n\n## Advanced Usage\n\n### Custom Batching\n\n```python\nclient = ObservabilityClient(\n api_key=\"YOUR_API_KEY\",\n base_url=\"https://observability.redducklabs.com/api/v1\",\n batch_size=200, # Larger batches\n batch_timeout=10.0, # Wait longer before sending\n enable_batching=True # Enable automatic batching\n)\n\n# Add logs to batch queue\nfor i in range(100):\n client.add_to_batch(\n message=f\"Log entry {i}\",\n level=LogLevel.INFO,\n source=\"batch-test\"\n )\n\n# Manually flush when needed\nclient.flush_batch()\n```\n\n### Error Handling\n\n```python\nfrom platform_observability import ObservabilityClient\nfrom platform_observability.exceptions import (\n AuthenticationError,\n RateLimitError,\n ValidationError\n)\n\nclient = ObservabilityClient(\n api_key=\"YOUR_API_KEY\",\n skip_connection_check=True # Skip initial health check\n)\n\ntry:\n client.ingest_log(\"Test message\", level=\"info\")\nexcept AuthenticationError:\n print(\"Invalid API key\")\nexcept RateLimitError:\n print(\"Rate limit exceeded, retry later\")\nexcept ValidationError as e:\n print(f\"Invalid log format: {e}\")\n```\n\n### Integration with Python Logging\n\n```python\nimport logging\nfrom platform_observability import ObservabilityHandler\n\n# Create custom handler\nhandler = ObservabilityHandler(\n api_key=\"YOUR_API_KEY\",\n base_url=\"https://observability.redducklabs.com/api/v1\",\n source=\"my-app\",\n labels={\"app\": \"my-app\", \"environment\": \"dev\"}\n)\n\n# Add to Python logger\nlogger = logging.getLogger(__name__)\nlogger.addHandler(handler)\nlogger.setLevel(logging.INFO)\n\n# Now regular Python logging will be sent to Platform Observability\nlogger.info(\"This goes to Platform Observability\")\nlogger.error(\"This error too\", extra={\"user_id\": \"user123\"})\n```\n\n## API Key Management\n\n```python\n# List API keys\nkeys = client.list_api_keys()\nfor key in keys:\n print(f\"{key.name}: {key.masked_key}\")\n\n# Create new API key\nnew_key = client.create_api_key(\n name=\"ci-pipeline\",\n description=\"API key for CI/CD pipeline\"\n)\nprint(f\"New key created: {new_key.key}\")\n\n# Rotate API key\nrotated = client.rotate_api_key(key_id=\"key_123\")\nprint(f\"New key value: {rotated['key']}\")\n\n# Delete API key\nclient.delete_api_key(key_id=\"key_123\")\n```\n\n## Usage Analytics\n\n```python\nfrom datetime import datetime, timedelta\n\n# Get usage metrics\nend_time = datetime.now()\nstart_time = end_time - timedelta(days=7)\n\nmetrics = client.get_usage_analytics(\n start_time=start_time,\n end_time=end_time\n)\n\nprint(f\"Total logs ingested: {metrics.total_logs_ingested}\")\nprint(f\"Data volume: {metrics.total_data_volume_gb:.2f} GB\")\nprint(f\"Average logs per request: {metrics.average_logs_per_request:.1f}\")\n```\n\n## CLI Tool\n\nAfter installation, use the `obs-cli` command-line tool:\n\n```bash\n# Set up configuration\nobs-cli config set api_key YOUR_API_KEY\nobs-cli config set endpoint https://observability.redducklabs.com/api/v1\n\n# Send a log\nobs-cli log \"Application deployed\" --level info --label app=my-app\n\n# List API keys\nobs-cli keys list\n\n# Get usage statistics\nobs-cli usage --days 7\n```\n\n## Development\n\n### Requirements\n- Python >= 3.8\n- pip\n\n### Setup Development Environment\n```bash\n# Clone repository\ngit clone https://github.com/redducklabs/platform-observability.git\ncd platform-observability/sdks/python\n\n# Install in development mode with dependencies\npip install -e .[dev]\n```\n\n### Running Tests\n```bash\n# Run all tests\npytest\n\n# Run with coverage\npytest --cov=platform_observability\n\n# Run specific test file\npytest tests/test_client.py\n```\n\n### Building Package\n```bash\n# Install build tools\npip install build\n\n# Build wheel and source distribution\npython -m build\n\n# Output will be in dist/\nls dist/\n```\n\n### Code Quality\n```bash\n# Format code\nblack platform_observability tests\n\n# Lint code\nflake8 platform_observability tests\n\n# Type checking\nmypy platform_observability\n```\n\n## Troubleshooting\n\n### Logs Not Appearing in Grafana?\n\n1. **Verify API Key**: Ensure your API key is valid and active\n2. **Check Endpoint**: Confirm you're using the correct endpoint URL\n3. **Validate Labels**: Ensure labels are properly set (required for Grafana queries)\n4. **Flush Logs**: Call `logger.flush()` to ensure batched logs are sent\n5. **Check Errors**: Enable debug logging to see any errors:\n ```python\n import logging\n logging.basicConfig(level=logging.DEBUG)\n ```\n\n### Connection Errors?\n\nIf you're getting connection errors during initialization:\n```python\nclient = ObservabilityClient(\n api_key=\"YOUR_API_KEY\",\n base_url=\"https://observability.redducklabs.com/api/v1\",\n skip_connection_check=True # Skip initial health check\n)\n```\n\n### Rate Limiting?\n\nThe SDK automatically retries on rate limit errors. You can also check rate limit status:\n```python\n# After making requests\nrate_limit_info = client.get_rate_limit_info()\nprint(f\"Remaining requests: {rate_limit_info.get('remaining')}\")\nprint(f\"Reset time: {rate_limit_info.get('reset')}\")\n```\n\n## Examples\n\nSee the `examples/` directory for complete working examples:\n- `examples/sample-application/sdk_only_example.py` - Basic SDK usage\n- `examples/sample-application/sdk_continuous_example.py` - Continuous log generation\n- `examples/sample-application/app.py` - FastAPI integration example\n\n## Support\n\nFor issues, questions, or contributions:\n- GitHub Issues: https://github.com/redducklabs/platform-observability/issues\n- Documentation: https://docs.redducklabs.com/observability\n\n## License\n\nMIT License - see LICENSE file for details\n",
"bugtrack_url": null,
"license": null,
"summary": "Python SDK for Platform Observability log ingestion and analytics",
"version": "1.0.2",
"project_urls": {
"Documentation": "https://docs.redducklabs.com/platform-observability",
"Homepage": "https://github.com/redducklabs/platform-observability",
"Issues": "https://github.com/redducklabs/platform-observability/issues",
"Repository": "https://github.com/redducklabs/platform-observability"
},
"split_keywords": [
"observability",
" logging",
" monitoring",
" analytics",
" platform"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "26b33aed9b9204569722f62c37fde65e52a3fbd0d26ab0d32dd98ccacce2103a",
"md5": "70dec06a6139660337704923e9de5577",
"sha256": "2789826e4729cf95789a76741e84c6169bd66771fa6f8c5f4a807ee26eb056a3"
},
"downloads": -1,
"filename": "redducklabs_platform_observability_sdk-1.0.2-py3-none-any.whl",
"has_sig": false,
"md5_digest": "70dec06a6139660337704923e9de5577",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.8",
"size": 17463,
"upload_time": "2025-08-25T16:22:02",
"upload_time_iso_8601": "2025-08-25T16:22:02.399331Z",
"url": "https://files.pythonhosted.org/packages/26/b3/3aed9b9204569722f62c37fde65e52a3fbd0d26ab0d32dd98ccacce2103a/redducklabs_platform_observability_sdk-1.0.2-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "a4ca33f3a82dfd3cc827cc96924cc61e9a7da675e02627e35d8f4ca5be1a4337",
"md5": "87f6c381a94f426a1631ee29644bb4cc",
"sha256": "b1a9a51d55f8d6aa4c2a424a5ef351f7604963c606997641c20b3492ace32d99"
},
"downloads": -1,
"filename": "redducklabs_platform_observability_sdk-1.0.2.tar.gz",
"has_sig": false,
"md5_digest": "87f6c381a94f426a1631ee29644bb4cc",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.8",
"size": 19191,
"upload_time": "2025-08-25T16:22:04",
"upload_time_iso_8601": "2025-08-25T16:22:04.046991Z",
"url": "https://files.pythonhosted.org/packages/a4/ca/33f3a82dfd3cc827cc96924cc61e9a7da675e02627e35d8f4ca5be1a4337/redducklabs_platform_observability_sdk-1.0.2.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-08-25 16:22:04",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "redducklabs",
"github_project": "platform-observability",
"github_not_found": true,
"lcname": "redducklabs-platform-observability-sdk"
}
JSON
