| Name | finalsa-traceability JSON |
| Version |
0.1.2
JSON |
| download |
| home_page | None |
| Summary | Thread-safe distributed tracing and correlation ID management for Python microservices |
| upload_time | 2025-07-10 02:54:45 |
| maintainer | None |
| docs_url | None |
| author | None |
| requires_python | >=3.9 |
| license | MIT License Copyright (c) 2021 Luis Diego Jiménez Delgado Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. |
| keywords |
async
context-management
correlation-id
distributed-tracing
finalsa
microservices
observability
thread-safe
traceability
|
| VCS |
 |
| bugtrack_url |
|
| requirements |
No requirements were recorded.
|
| Travis-CI |
No Travis.
|
| coveralls test coverage |
No coveralls.
|
# Finalsa Traceability
A comprehensive Python library for managing distributed tracing and correlation IDs across microservices and distributed systems. Provides thread-safe context management using Python's `contextvars` for proper isolation between concurrent operations.
## Features
- **Thread-Safe Context Management**: Uses Python's `contextvars` for proper isolation
- **Automatic ID Generation**: Creates correlation IDs, trace IDs, and span IDs
- **Hop Tracking**: Automatically extends correlation IDs as requests flow through services
- **HTTP Integration**: Standard header constants for easy integration
- **Async Support**: Full support for async/await operations
- **Type Hints**: Complete type annotations for better development experience
- **100% Test Coverage**: Comprehensive test suite ensuring reliability
## Installation
```bash
# Using uv (recommended)
uv add finalsa-traceability
# Using pip
pip install finalsa-traceability
# For development with examples
uv sync --group examples --group test
```
## Quick Start
### Basic Usage
```python
from finalsa.traceability import set_context, get_context
# Set traceability context for your service
set_context(
correlation_id="user-request-123",
service_name="auth-service"
)
# Get current context
context = get_context()
print(context["correlation_id"]) # "user-request-123-XXXXX"
print(context["trace_id"]) # Auto-generated UUID
print(context["span_id"]) # Auto-generated UUID
```
### HTTP Service Integration
```python
from finalsa.traceability import (
set_context_from_dict,
get_context,
HTTP_HEADER_CORRELATION_ID,
HTTP_HEADER_TRACE_ID,
HTTP_HEADER_SPAN_ID
)
# Extract traceability from incoming HTTP request
def handle_request(request):
headers = {
'correlation_id': request.headers.get(HTTP_HEADER_CORRELATION_ID),
'trace_id': request.headers.get(HTTP_HEADER_TRACE_ID),
'span_id': request.headers.get(HTTP_HEADER_SPAN_ID)
}
# Set context for this service
set_context_from_dict(headers, service_name="api-gateway")
# Your business logic here
result = process_request(request)
# Add traceability to outgoing response
context = get_context()
response.headers[HTTP_HEADER_CORRELATION_ID] = context["correlation_id"]
response.headers[HTTP_HEADER_TRACE_ID] = context["trace_id"]
response.headers[HTTP_HEADER_SPAN_ID] = context["span_id"]
return response
```
### Flask Integration
```python
from flask import Flask, request, g
from finalsa.traceability import set_context_from_dict, get_context
app = Flask(__name__)
@app.before_request
def before_request():
# Extract traceability from request headers
headers = {
'correlation_id': request.headers.get('X-Correlation-ID'),
'trace_id': request.headers.get('X-Trace-ID'),
'span_id': request.headers.get('X-Span-ID')
}
set_context_from_dict(headers, service_name="my-flask-app")
@app.after_request
def after_request(response):
# Add traceability to response headers
context = get_context()
response.headers['X-Correlation-ID'] = context["correlation_id"]
response.headers['X-Trace-ID'] = context["trace_id"]
response.headers['X-Span-ID'] = context["span_id"]
return response
@app.route('/api/users')
def get_users():
# Context is automatically available
context = get_context()
print(f"Processing request with correlation_id: {context['correlation_id']}")
return {"users": []}
```
### FastAPI Integration
```python
from fastapi import FastAPI, Request, Response
from finalsa.traceability import set_context_from_dict, get_context
app = FastAPI()
@app.middleware("http")
async def traceability_middleware(request: Request, call_next):
# Extract traceability from request headers
headers = {
'correlation_id': request.headers.get('x-correlation-id'),
'trace_id': request.headers.get('x-trace-id'),
'span_id': request.headers.get('x-span-id')
}
set_context_from_dict(headers, service_name="my-fastapi-app")
# Process request
response = await call_next(request)
# Add traceability to response headers
context = get_context()
response.headers['x-correlation-id'] = context["correlation_id"]
response.headers['x-trace-id'] = context["trace_id"]
response.headers['x-span-id'] = context["span_id"]
return response
@app.get("/api/users")
async def get_users():
context = get_context()
print(f"Processing request with correlation_id: {context['correlation_id']}")
return {"users": []}
```
### Message Queue Integration
```python
from finalsa.traceability import set_context_from_dict, get_context
# Publishing messages
def publish_message(message_data):
context = get_context()
# Add traceability to message properties
message_properties = {
'correlation_id': context["correlation_id"],
'trace_id': context["trace_id"],
'span_id': context["span_id"]
}
publish_to_queue(message_data, properties=message_properties)
# Consuming messages
def handle_message(message):
# Extract traceability from message properties
headers = {
'correlation_id': message.properties.get('correlation_id'),
'trace_id': message.properties.get('trace_id'),
'span_id': message.properties.get('span_id')
}
set_context_from_dict(
headers,
service_name="order-processor",
queue_name="orders",
message_id=message.id
)
# Process message with traceability context
process_order(message.data)
```
## API Reference
### Context Management
#### `set_context(correlation_id=None, trace_id=None, span_id=None, service_name=None, **kwargs)`
Set multiple traceability IDs and custom variables in one call.
**Parameters:**
- `correlation_id` (str, optional): Correlation ID to extend with hop
- `trace_id` (str, optional): Trace ID to set (generates UUID if None)
- `span_id` (str, optional): Span ID to set (generates UUID if None)
- `service_name` (str, optional): Service name for correlation ID generation
- `**kwargs`: Additional custom variables to store
#### `set_context_from_dict(context, service_name=None, **kwargs)`
Set traceability context from a dictionary (e.g., HTTP headers).
**Parameters:**
- `context` (dict): Dictionary with 'correlation_id', 'trace_id', 'span_id' keys
- `service_name` (str, optional): Service name for new correlation ID generation
- `**kwargs`: Additional custom variables to store
#### `get_context() -> Dict`
Get the complete current traceability context.
**Returns:** Dictionary containing correlation_id, trace_id, span_id, and custom variables.
### Individual Setters/Getters
#### `set_correlation_id(value=None, service_name=None)`
#### `set_trace_id(value=None)`
#### `set_span_id(value=None)`
#### `get_correlation_id() -> Optional[str]`
#### `get_trace_id() -> Optional[str]`
#### `get_span_id() -> Optional[str]`
### ID Generation Functions
#### `default_correlation_id(service_name=None) -> str`
Generate correlation ID in format "service_name-XXXXX".
#### `default_trace_id() -> str`
Generate trace ID using UUID4.
#### `default_span_id() -> str`
Generate span ID using UUID4.
#### `add_hop_to_correlation(correlation_id) -> str`
Add a hop to existing correlation ID.
#### `id_generator(size=5, chars=string.ascii_uppercase + string.digits) -> str`
Generate random alphanumeric ID.
### Constants
#### HTTP Headers
- `HTTP_HEADER_CORRELATION_ID` = "X-Correlation-ID"
- `HTTP_HEADER_TRACE_ID` = "X-Trace-ID"
- `HTTP_HEADER_SPAN_ID` = "X-Span-ID"
- `HTTP_AUTHORIZATION_HEADER` = "Authorization"
#### Async Context Keys
- `ASYNC_CONTEXT_CORRELATION_ID` = "correlation_id"
- `ASYNC_CONTEXT_TRACE_ID` = "trace_id"
- `ASYNC_CONTEXT_SPAN_ID` = "span_id"
- `ASYNC_CONTEXT_TOPIC` = "topic"
- `ASYNC_CONTEXT_SUBTOPIC` = "subtopic"
- `ASYNC_CONTEXT_AUTHORIZATION` = "auth"
## Advanced Usage
### Custom Variables
```python
from finalsa.traceability import set_context, get_context
# Store custom variables with traceability context
set_context(
correlation_id="request-123",
service_name="user-service",
user_id="user-456",
operation="login",
ip_address="192.168.1.1"
)
context = get_context()
print(context["user_id"]) # "user-456"
print(context["operation"]) # "login"
print(context["ip_address"]) # "192.168.1.1"
```
### Thread Safety
The library uses Python's `contextvars` which ensures proper isolation between threads and async tasks:
```python
import threading
from finalsa.traceability import set_context, get_context
def worker_thread(thread_id):
# Each thread has its own context
set_context(
correlation_id=f"thread-{thread_id}",
service_name="worker-service"
)
context = get_context()
print(f"Thread {thread_id}: {context['correlation_id']}")
# Create multiple threads
threads = []
for i in range(5):
thread = threading.Thread(target=worker_thread, args=(i,))
threads.append(thread)
thread.start()
for thread in threads:
thread.join()
```
### Async/Await Support
```python
import asyncio
from finalsa.traceability import set_context, get_context
async def async_task(task_id):
# Each async task has its own context
set_context(
correlation_id=f"task-{task_id}",
service_name="async-service"
)
await asyncio.sleep(0.1) # Simulate async work
context = get_context()
print(f"Task {task_id}: {context['correlation_id']}")
async def main():
# Run multiple async tasks
tasks = [async_task(i) for i in range(5)]
await asyncio.gather(*tasks)
asyncio.run(main())
```
## Best Practices
### 1. Service Naming
Use consistent service names across your application:
```python
SERVICE_NAME = "auth-service"
set_context(service_name=SERVICE_NAME)
```
### 2. HTTP Header Propagation
Always propagate traceability headers between services:
```python
import requests
from finalsa.traceability import get_context, HTTP_HEADER_CORRELATION_ID
def call_downstream_service():
context = get_context()
headers = {
HTTP_HEADER_CORRELATION_ID: context["correlation_id"],
"X-Trace-ID": context["trace_id"],
"X-Span-ID": context["span_id"]
}
response = requests.get("http://downstream-service/api", headers=headers)
return response
```
### 3. Error Handling
Include traceability in error logs:
```python
import logging
from finalsa.traceability import get_context
def handle_error(error):
context = get_context()
logging.error(
f"Error occurred - "
f"correlation_id: {context.get('correlation_id')}, "
f"trace_id: {context.get('trace_id')}, "
f"error: {error}"
)
```
### 4. Database Operations
Include traceability in database logs:
```python
from finalsa.traceability import get_context
def execute_query(sql, params):
context = get_context()
# Log query with traceability
logging.info(
f"Executing query - "
f"correlation_id: {context.get('correlation_id')}, "
f"sql: {sql}"
)
# Execute query
return database.execute(sql, params)
```
## Examples
The `examples/` directory contains comprehensive examples demonstrating different use cases:
### Basic Usage
- **`examples/basic_usage.py`** - Simple usage patterns and basic context operations
### Web Framework Integration
- **`examples/fastapi_integration.py`** - Complete FastAPI application with middleware, dependency injection, async patterns, and error handling
### Concurrency & Thread Safety
- **`examples/thread_safety_demo.py`** - Thread and async safety demonstrations
### Running Examples
Using uv (recommended):
```bash
# Install example dependencies
uv sync --group examples
# Basic usage example
uv run python examples/basic_usage.py
# FastAPI example
uv run python examples/fastapi_integration.py
# Or use the convenience script
./run_fastapi_example.sh
# Thread safety demo
uv run python examples/thread_safety_demo.py
```
Using pip:
```bash
# Install example dependencies
pip install -r examples/requirements.txt
# Run examples
python examples/basic_usage.py
python examples/fastapi_integration.py
python examples/thread_safety_demo.py
```
See `examples/README.md` for detailed usage instructions and API patterns.
## Development
### Requirements
- Python 3.9+ (uses only standard library for core functionality)
- Optional dependencies for examples and testing (see `pyproject.toml`)
### Running Tests
```bash
# Install development dependencies
uv add --dev pytest coverage
# Run tests
uv run pytest
# Run tests with coverage
uv run coverage run -m pytest
uv run coverage report --show-missing
```
### Contributing
1. Fork the repository
2. Create a feature branch
3. Add tests for new functionality
4. Ensure all tests pass
5. Submit a pull request
## License
This project is licensed under the MIT License - see the LICENSE file for details.
## Changelog
### v0.1.1
- Initial release
- Basic traceability context management
- HTTP header constants
- Thread-safe operation using contextvars
- Comprehensive test suite with 100% coverage
Raw data
{
"_id": null,
"home_page": null,
"name": "finalsa-traceability",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.9",
"maintainer_email": null,
"keywords": "async, context-management, correlation-id, distributed-tracing, finalsa, microservices, observability, thread-safe, traceability",
"author": null,
"author_email": "Luis Jimenez <luis@finalsa.com>",
"download_url": "https://files.pythonhosted.org/packages/08/04/d89a93041f4e79fa8191af8f4ace2fdc40188cfe1bba0b6f1543fef0dd8e/finalsa_traceability-0.1.2.tar.gz",
"platform": null,
"description": "# Finalsa Traceability\n\nA comprehensive Python library for managing distributed tracing and correlation IDs across microservices and distributed systems. Provides thread-safe context management using Python's `contextvars` for proper isolation between concurrent operations.\n\n## Features\n\n- **Thread-Safe Context Management**: Uses Python's `contextvars` for proper isolation\n- **Automatic ID Generation**: Creates correlation IDs, trace IDs, and span IDs\n- **Hop Tracking**: Automatically extends correlation IDs as requests flow through services\n- **HTTP Integration**: Standard header constants for easy integration\n- **Async Support**: Full support for async/await operations\n- **Type Hints**: Complete type annotations for better development experience\n- **100% Test Coverage**: Comprehensive test suite ensuring reliability\n\n## Installation\n\n```bash\n# Using uv (recommended)\nuv add finalsa-traceability\n\n# Using pip\npip install finalsa-traceability\n\n# For development with examples\nuv sync --group examples --group test\n```\n\n## Quick Start\n\n### Basic Usage\n\n```python\nfrom finalsa.traceability import set_context, get_context\n\n# Set traceability context for your service\nset_context(\n correlation_id=\"user-request-123\",\n service_name=\"auth-service\"\n)\n\n# Get current context\ncontext = get_context()\nprint(context[\"correlation_id\"]) # \"user-request-123-XXXXX\"\nprint(context[\"trace_id\"]) # Auto-generated UUID\nprint(context[\"span_id\"]) # Auto-generated UUID\n```\n\n### HTTP Service Integration\n\n```python\nfrom finalsa.traceability import (\n set_context_from_dict, \n get_context,\n HTTP_HEADER_CORRELATION_ID,\n HTTP_HEADER_TRACE_ID,\n HTTP_HEADER_SPAN_ID\n)\n\n# Extract traceability from incoming HTTP request\ndef handle_request(request):\n headers = {\n 'correlation_id': request.headers.get(HTTP_HEADER_CORRELATION_ID),\n 'trace_id': request.headers.get(HTTP_HEADER_TRACE_ID),\n 'span_id': request.headers.get(HTTP_HEADER_SPAN_ID)\n }\n \n # Set context for this service\n set_context_from_dict(headers, service_name=\"api-gateway\")\n \n # Your business logic here\n result = process_request(request)\n \n # Add traceability to outgoing response\n context = get_context()\n response.headers[HTTP_HEADER_CORRELATION_ID] = context[\"correlation_id\"]\n response.headers[HTTP_HEADER_TRACE_ID] = context[\"trace_id\"]\n response.headers[HTTP_HEADER_SPAN_ID] = context[\"span_id\"]\n \n return response\n```\n\n### Flask Integration\n\n```python\nfrom flask import Flask, request, g\nfrom finalsa.traceability import set_context_from_dict, get_context\n\napp = Flask(__name__)\n\n@app.before_request\ndef before_request():\n # Extract traceability from request headers\n headers = {\n 'correlation_id': request.headers.get('X-Correlation-ID'),\n 'trace_id': request.headers.get('X-Trace-ID'), \n 'span_id': request.headers.get('X-Span-ID')\n }\n set_context_from_dict(headers, service_name=\"my-flask-app\")\n\n@app.after_request \ndef after_request(response):\n # Add traceability to response headers\n context = get_context()\n response.headers['X-Correlation-ID'] = context[\"correlation_id\"]\n response.headers['X-Trace-ID'] = context[\"trace_id\"]\n response.headers['X-Span-ID'] = context[\"span_id\"]\n return response\n\n@app.route('/api/users')\ndef get_users():\n # Context is automatically available\n context = get_context()\n print(f\"Processing request with correlation_id: {context['correlation_id']}\")\n return {\"users\": []}\n```\n\n### FastAPI Integration\n\n```python\nfrom fastapi import FastAPI, Request, Response\nfrom finalsa.traceability import set_context_from_dict, get_context\n\napp = FastAPI()\n\n@app.middleware(\"http\")\nasync def traceability_middleware(request: Request, call_next):\n # Extract traceability from request headers\n headers = {\n 'correlation_id': request.headers.get('x-correlation-id'),\n 'trace_id': request.headers.get('x-trace-id'),\n 'span_id': request.headers.get('x-span-id')\n }\n set_context_from_dict(headers, service_name=\"my-fastapi-app\")\n \n # Process request\n response = await call_next(request)\n \n # Add traceability to response headers\n context = get_context()\n response.headers['x-correlation-id'] = context[\"correlation_id\"]\n response.headers['x-trace-id'] = context[\"trace_id\"]\n response.headers['x-span-id'] = context[\"span_id\"]\n \n return response\n\n@app.get(\"/api/users\")\nasync def get_users():\n context = get_context()\n print(f\"Processing request with correlation_id: {context['correlation_id']}\")\n return {\"users\": []}\n```\n\n### Message Queue Integration\n\n```python\nfrom finalsa.traceability import set_context_from_dict, get_context\n\n# Publishing messages\ndef publish_message(message_data):\n context = get_context()\n \n # Add traceability to message properties\n message_properties = {\n 'correlation_id': context[\"correlation_id\"],\n 'trace_id': context[\"trace_id\"],\n 'span_id': context[\"span_id\"]\n }\n \n publish_to_queue(message_data, properties=message_properties)\n\n# Consuming messages \ndef handle_message(message):\n # Extract traceability from message properties\n headers = {\n 'correlation_id': message.properties.get('correlation_id'),\n 'trace_id': message.properties.get('trace_id'),\n 'span_id': message.properties.get('span_id')\n }\n \n set_context_from_dict(\n headers, \n service_name=\"order-processor\",\n queue_name=\"orders\",\n message_id=message.id\n )\n \n # Process message with traceability context\n process_order(message.data)\n```\n\n## API Reference\n\n### Context Management\n\n#### `set_context(correlation_id=None, trace_id=None, span_id=None, service_name=None, **kwargs)`\n\nSet multiple traceability IDs and custom variables in one call.\n\n**Parameters:**\n- `correlation_id` (str, optional): Correlation ID to extend with hop\n- `trace_id` (str, optional): Trace ID to set (generates UUID if None)\n- `span_id` (str, optional): Span ID to set (generates UUID if None) \n- `service_name` (str, optional): Service name for correlation ID generation\n- `**kwargs`: Additional custom variables to store\n\n#### `set_context_from_dict(context, service_name=None, **kwargs)`\n\nSet traceability context from a dictionary (e.g., HTTP headers).\n\n**Parameters:**\n- `context` (dict): Dictionary with 'correlation_id', 'trace_id', 'span_id' keys\n- `service_name` (str, optional): Service name for new correlation ID generation\n- `**kwargs`: Additional custom variables to store\n\n#### `get_context() -> Dict`\n\nGet the complete current traceability context.\n\n**Returns:** Dictionary containing correlation_id, trace_id, span_id, and custom variables.\n\n### Individual Setters/Getters\n\n#### `set_correlation_id(value=None, service_name=None)`\n#### `set_trace_id(value=None)`\n#### `set_span_id(value=None)`\n#### `get_correlation_id() -> Optional[str]`\n#### `get_trace_id() -> Optional[str]`\n#### `get_span_id() -> Optional[str]`\n\n### ID Generation Functions\n\n#### `default_correlation_id(service_name=None) -> str`\n\nGenerate correlation ID in format \"service_name-XXXXX\".\n\n#### `default_trace_id() -> str`\n\nGenerate trace ID using UUID4.\n\n#### `default_span_id() -> str`\n\nGenerate span ID using UUID4.\n\n#### `add_hop_to_correlation(correlation_id) -> str`\n\nAdd a hop to existing correlation ID.\n\n#### `id_generator(size=5, chars=string.ascii_uppercase + string.digits) -> str`\n\nGenerate random alphanumeric ID.\n\n### Constants\n\n#### HTTP Headers\n- `HTTP_HEADER_CORRELATION_ID` = \"X-Correlation-ID\"\n- `HTTP_HEADER_TRACE_ID` = \"X-Trace-ID\"\n- `HTTP_HEADER_SPAN_ID` = \"X-Span-ID\"\n- `HTTP_AUTHORIZATION_HEADER` = \"Authorization\"\n\n#### Async Context Keys\n- `ASYNC_CONTEXT_CORRELATION_ID` = \"correlation_id\"\n- `ASYNC_CONTEXT_TRACE_ID` = \"trace_id\"\n- `ASYNC_CONTEXT_SPAN_ID` = \"span_id\"\n- `ASYNC_CONTEXT_TOPIC` = \"topic\"\n- `ASYNC_CONTEXT_SUBTOPIC` = \"subtopic\"\n- `ASYNC_CONTEXT_AUTHORIZATION` = \"auth\"\n\n## Advanced Usage\n\n### Custom Variables\n\n```python\nfrom finalsa.traceability import set_context, get_context\n\n# Store custom variables with traceability context\nset_context(\n correlation_id=\"request-123\",\n service_name=\"user-service\",\n user_id=\"user-456\",\n operation=\"login\",\n ip_address=\"192.168.1.1\"\n)\n\ncontext = get_context()\nprint(context[\"user_id\"]) # \"user-456\"\nprint(context[\"operation\"]) # \"login\"\nprint(context[\"ip_address\"]) # \"192.168.1.1\"\n```\n\n### Thread Safety\n\nThe library uses Python's `contextvars` which ensures proper isolation between threads and async tasks:\n\n```python\nimport threading\nfrom finalsa.traceability import set_context, get_context\n\ndef worker_thread(thread_id):\n # Each thread has its own context\n set_context(\n correlation_id=f\"thread-{thread_id}\",\n service_name=\"worker-service\"\n )\n \n context = get_context()\n print(f\"Thread {thread_id}: {context['correlation_id']}\")\n\n# Create multiple threads\nthreads = []\nfor i in range(5):\n thread = threading.Thread(target=worker_thread, args=(i,))\n threads.append(thread)\n thread.start()\n\nfor thread in threads:\n thread.join()\n```\n\n### Async/Await Support\n\n```python\nimport asyncio\nfrom finalsa.traceability import set_context, get_context\n\nasync def async_task(task_id):\n # Each async task has its own context\n set_context(\n correlation_id=f\"task-{task_id}\",\n service_name=\"async-service\"\n )\n \n await asyncio.sleep(0.1) # Simulate async work\n \n context = get_context()\n print(f\"Task {task_id}: {context['correlation_id']}\")\n\nasync def main():\n # Run multiple async tasks\n tasks = [async_task(i) for i in range(5)]\n await asyncio.gather(*tasks)\n\nasyncio.run(main())\n```\n\n## Best Practices\n\n### 1. Service Naming\nUse consistent service names across your application:\n\n```python\nSERVICE_NAME = \"auth-service\"\nset_context(service_name=SERVICE_NAME)\n```\n\n### 2. HTTP Header Propagation\nAlways propagate traceability headers between services:\n\n```python\nimport requests\nfrom finalsa.traceability import get_context, HTTP_HEADER_CORRELATION_ID\n\ndef call_downstream_service():\n context = get_context()\n headers = {\n HTTP_HEADER_CORRELATION_ID: context[\"correlation_id\"],\n \"X-Trace-ID\": context[\"trace_id\"],\n \"X-Span-ID\": context[\"span_id\"]\n }\n \n response = requests.get(\"http://downstream-service/api\", headers=headers)\n return response\n```\n\n### 3. Error Handling\nInclude traceability in error logs:\n\n```python\nimport logging\nfrom finalsa.traceability import get_context\n\ndef handle_error(error):\n context = get_context()\n logging.error(\n f\"Error occurred - \"\n f\"correlation_id: {context.get('correlation_id')}, \"\n f\"trace_id: {context.get('trace_id')}, \"\n f\"error: {error}\"\n )\n```\n\n### 4. Database Operations\nInclude traceability in database logs:\n\n```python\nfrom finalsa.traceability import get_context\n\ndef execute_query(sql, params):\n context = get_context()\n \n # Log query with traceability\n logging.info(\n f\"Executing query - \"\n f\"correlation_id: {context.get('correlation_id')}, \"\n f\"sql: {sql}\"\n )\n \n # Execute query\n return database.execute(sql, params)\n```\n\n## Examples\n\nThe `examples/` directory contains comprehensive examples demonstrating different use cases:\n\n### Basic Usage\n- **`examples/basic_usage.py`** - Simple usage patterns and basic context operations\n\n### Web Framework Integration\n- **`examples/fastapi_integration.py`** - Complete FastAPI application with middleware, dependency injection, async patterns, and error handling\n\n### Concurrency & Thread Safety\n- **`examples/thread_safety_demo.py`** - Thread and async safety demonstrations\n\n### Running Examples\n\nUsing uv (recommended):\n```bash\n# Install example dependencies\nuv sync --group examples\n\n# Basic usage example\nuv run python examples/basic_usage.py\n\n# FastAPI example \nuv run python examples/fastapi_integration.py\n# Or use the convenience script\n./run_fastapi_example.sh\n\n# Thread safety demo\nuv run python examples/thread_safety_demo.py\n```\n\nUsing pip:\n```bash\n# Install example dependencies\npip install -r examples/requirements.txt\n\n# Run examples\npython examples/basic_usage.py\npython examples/fastapi_integration.py\npython examples/thread_safety_demo.py\n```\n\nSee `examples/README.md` for detailed usage instructions and API patterns.\n\n## Development\n\n### Requirements\n- Python 3.9+ (uses only standard library for core functionality)\n- Optional dependencies for examples and testing (see `pyproject.toml`)\n\n### Running Tests\n\n```bash\n# Install development dependencies\nuv add --dev pytest coverage\n\n# Run tests\nuv run pytest\n\n# Run tests with coverage\nuv run coverage run -m pytest\nuv run coverage report --show-missing\n```\n\n### Contributing\n\n1. Fork the repository\n2. Create a feature branch\n3. Add tests for new functionality \n4. Ensure all tests pass\n5. Submit a pull request\n\n## License\n\nThis project is licensed under the MIT License - see the LICENSE file for details.\n\n## Changelog\n\n### v0.1.1\n- Initial release\n- Basic traceability context management\n- HTTP header constants\n- Thread-safe operation using contextvars\n- Comprehensive test suite with 100% coverage",
"bugtrack_url": null,
"license": "MIT License Copyright (c) 2021 Luis Diego Jim\u00e9nez Delgado Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the \"Software\"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED \"AS IS\", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.",
"summary": "Thread-safe distributed tracing and correlation ID management for Python microservices",
"version": "0.1.2",
"project_urls": {
"Changelog": "https://github.com/finalsa/finalsa-traceability/blob/main/CHANGELOG.md",
"Documentation": "https://github.com/finalsa/finalsa-traceability#readme",
"Homepage": "https://github.com/finalsa/finalsa-traceability",
"Issues": "https://github.com/finalsa/finalsa-traceability/issues",
"Repository": "https://github.com/finalsa/finalsa-traceability.git"
},
"split_keywords": [
"async",
" context-management",
" correlation-id",
" distributed-tracing",
" finalsa",
" microservices",
" observability",
" thread-safe",
" traceability"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "b226c9a24785751b4abc79f23686e653ee29afa42d4dfb26cf99d9225b2d9aa0",
"md5": "8e6ab0350b181723837e72b290d4677c",
"sha256": "8bc88e3b1f11e5ea6d9b583516374ab2212fe3298e18124711f1736d294f597d"
},
"downloads": -1,
"filename": "finalsa_traceability-0.1.2-py3-none-any.whl",
"has_sig": false,
"md5_digest": "8e6ab0350b181723837e72b290d4677c",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.9",
"size": 12706,
"upload_time": "2025-07-10T02:54:44",
"upload_time_iso_8601": "2025-07-10T02:54:44.577096Z",
"url": "https://files.pythonhosted.org/packages/b2/26/c9a24785751b4abc79f23686e653ee29afa42d4dfb26cf99d9225b2d9aa0/finalsa_traceability-0.1.2-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "0804d89a93041f4e79fa8191af8f4ace2fdc40188cfe1bba0b6f1543fef0dd8e",
"md5": "d6edd661ac9916b5da9497707e6a1de5",
"sha256": "c15e98e1825d0a4926ebcfffaf3f0441398a016d2632383c427fdeb7e30b709b"
},
"downloads": -1,
"filename": "finalsa_traceability-0.1.2.tar.gz",
"has_sig": false,
"md5_digest": "d6edd661ac9916b5da9497707e6a1de5",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.9",
"size": 10965,
"upload_time": "2025-07-10T02:54:45",
"upload_time_iso_8601": "2025-07-10T02:54:45.760619Z",
"url": "https://files.pythonhosted.org/packages/08/04/d89a93041f4e79fa8191af8f4ace2fdc40188cfe1bba0b6f1543fef0dd8e/finalsa_traceability-0.1.2.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-07-10 02:54:45",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "finalsa",
"github_project": "finalsa-traceability",
"github_not_found": true,
"lcname": "finalsa-traceability"
}
