| Name | oteilo JSON |
| Version |
0.1.0
JSON |
| download |
| home_page | None |
| Summary | A collection of intuitive Python utilities for common development tasks |
| upload_time | 2025-02-06 22:40:00 |
| maintainer | None |
| docs_url | None |
| author | Teilo Millet |
| requires_python | >=3.12 |
| license | None |
| keywords |
io
logging
timing
utilities
validation
|
| VCS |
 |
| bugtrack_url |
|
| requirements |
No requirements were recorded.
|
| Travis-CI |
No Travis.
|
| coveralls test coverage |
No coveralls.
|
# Oteilo
A robust Python toolkit that helps your team write safer, more maintainable code with built-in monitoring and validation. Think of it as a safety net that catches errors early and provides clear insights into your application's behavior.
## Why Oteilo?
- **Catch Problems Early**: Automatically validates data types and catches errors before they reach production
- **Clear Visibility**: Built-in monitoring shows exactly what your code is doing and how long it takes
- **Maintainable Code**: Enforces clean, consistent patterns that make code easier to understand and modify
- **Reduced Risk**: Strong type checking and validation prevent data-related bugs
- **Developer Productivity**: Common patterns are pre-built, tested, and ready to use
## Quick Start
```bash
pip install oteilo
```
## Understanding Imports
Oteilo is organized into focused modules, each handling specific concerns. Here's how to import and use them:
### Core Pipeline Components
```python
from oteilo.pipeline import Pipeline, Step
from typing import Dict, Any # Python's typing for type hints
# Use Pipeline when you need to:
# - Chain multiple operations together
# - Validate data types between steps
# - Process data through a sequence of transformations
pipeline = Pipeline().then(step1).then(step2)
# Use Step when you need to:
# - Create reusable, configurable pipeline steps
# - Add automatic type validation
# - Provide clear documentation and metadata
class MyStep(Step):
def process(self, data: Dict[str, Any]) -> Dict[str, Any]:
return transformed_data
```
### Monitoring and Debugging
```python
from oteilo.functional import trace # For function monitoring
from oteilo.timing import timing_block # For performance tracking
from oteilo.logging import configure_logging # For setting up logging
# Use @trace when you need to:
# - Monitor function execution
# - Track timing and performance
# - Log input/output data
@trace(level="DEBUG")
def my_function(): ...
# Use timing_block when you need to:
# - Measure specific code blocks
# - Track nested operations
# - Profile performance
with timing_block("operation_name"):
do_something()
```
### Error Handling
```python
from oteilo.pipeline import PipelineError # For catching pipeline-specific errors
# Use PipelineError when you need to:
# - Distinguish pipeline errors from other exceptions
# - Access detailed error information
# - Handle different types of validation failures
try:
result = pipeline.process(data)
except PipelineError as e:
print(f"Error details: {e.details}")
print(f"Original cause: {e.cause}")
```
### Type Validation
```python
from oteilo.validation import validate_type # For manual type checking
from typing import Dict, List, Any, Optional # Python's type hints
# Use validate_type when you need to:
# - Check types manually
# - Handle complex type validation
# - Support Union and Optional types
if not validate_type(data, Dict[str, Any]):
raise ValueError("Expected dictionary")
```
### Complete Example
```python
from oteilo.pipeline import Pipeline, Step, PipelineError
from oteilo.functional import trace
from oteilo.timing import timing_block
from oteilo.logging import configure_logging
from oteilo.validation import validate_type
from typing import Dict, Any, Optional
# Set up logging first
configure_logging(level="DEBUG")
# Create monitored functions
@trace
def process_data(data: Dict[str, Any]) -> Dict[str, Any]:
with timing_block("processing"):
# Your processing logic here
return processed_data
# Create pipeline steps
class ValidationStep(Step):
def validate(self, data: Dict[str, Any]) -> Dict[str, Any]:
# Validation logic here
return validated_data
# Build and use pipeline
pipeline = Pipeline().then(ValidationStep()).then(process_data)
# Handle errors appropriately
def handle_request(data: Dict[str, Any]) -> Dict[str, Any]:
try:
return pipeline.process(data)
except PipelineError as e:
# Handle pipeline-specific errors
return {"error": str(e), "details": e.details}
except Exception as e:
# Handle unexpected errors
return {"error": "Unexpected error", "details": str(e)}
```
## Key Features (With Real-World Examples)
### 1. Automatic Function Monitoring
Monitor any function to see what's happening inside your application. The `@trace` decorator provides both timing and comprehensive monitoring:
```python
from oteilo.functional import trace
# 1. Simple timing only (default)
@trace
def calculate_total(order):
return order["quantity"] * order["price"]
# Output:
# DEBUG: calculate_total took 0.001 seconds
# 2. Full tracing with timing and monitoring
@trace(level="DEBUG")
def calculate_order_total(order):
return order["quantity"] * order["price"]
# Output:
# DEBUG: calculate_order_total called with {'order': {'quantity': 5, 'price': 9.99}}
# DEBUG: calculate_order_total type information {'params': {'order': 'Dict[str, Any]'}, 'return': 'float'}
# DEBUG: Executing calculate_order_total
# DEBUG: calculate_order_total took 0.001 seconds
# DEBUG: calculate_order_total completed with result: 49.95
# 3. Custom configuration
@trace(
level="INFO", # Set logging level
timing=True, # Include timing (default)
args=True, # Log arguments (default)
types=True, # Log type information (default)
nested=False # Don't trace nested calls
)
def process_order(order):
validate_order(order)
total = calculate_total(order)
return format_output(total)
# You can also combine @trace with timing_block for more granular timing:
@trace(level="DEBUG")
def process_large_order(order):
with timing_block("validation"):
validate_order(order)
with timing_block("calculation"):
total = calculate_total(order)
with timing_block("formatting"):
return format_output(total)
# Output:
# DEBUG: process_large_order called with {'order': {...}}
# DEBUG: Executing process_large_order
# DEBUG: validation took 0.001 seconds
# DEBUG: calculation took 0.002 seconds
# DEBUG: formatting took 0.001 seconds
# DEBUG: process_large_order took 0.004 seconds
# DEBUG: process_large_order completed with result: {...}
```
**Business Value**:
- Instantly see performance bottlenecks with automatic timing
- Debug issues with detailed logging of inputs and outputs
- Monitor critical business functions with type validation
- Track nested operations and their timing
- Flexible configuration to match your monitoring needs
### 2. Safe Data Processing Pipeline
Build reliable data transformation workflows that validate every step:
```python
from oteilo.pipeline import Pipeline
from typing import Dict, Any
# Define each step of your process
def validate_order(order):
"""Make sure we have all required order data."""
if not order.get("customer_id"):
raise ValueError("Missing customer ID")
return order
def calculate_total(order):
"""Calculate order total with tax."""
subtotal = order["quantity"] * order["price"]
tax = subtotal * 0.1
return {**order, "subtotal": subtotal, "tax": tax, "total": subtotal + tax}
def format_currency(order):
"""Format money values for display."""
return {
**order,
"subtotal": f"${order['subtotal']:.2f}",
"tax": f"${order['tax']:.2f}",
"total": f"${order['total']:.2f}"
}
# 1. First, define your pipeline - this validates type compatibility
order_pipeline = (Pipeline()
.then(validate_order) # Dict -> Dict
.then(calculate_total) # Dict -> Dict
.then(format_currency)) # Dict -> Dict
# 2. Then use it to process data - this is where runtime checks happen
def process_order(order_data: Dict[str, Any]) -> Dict[str, Any]:
"""Process an order with full validation."""
# Type validation and business rules are checked during processing
return order_pipeline.process(order_data)
# 3. Now you can handle errors at your application level
def handle_order_request(request):
try:
order_data = {
"customer_id": "12345",
"quantity": 5,
"price": 9.99
}
result = process_order(order_data)
return {"status": "success", "data": result}
except PipelineError as e:
if "Invalid input type" in str(e):
return {"status": "error", "message": "Invalid data format"}
if "quantity must be positive" in str(e):
return {"status": "error", "message": "Invalid quantity"}
return {"status": "error", "message": str(e)}
# This is cleaner than mixing pipeline usage and error handling
response = handle_order_request(request)
if response["status"] == "success":
send_confirmation(response["data"])
else:
notify_user(response["message"])
```
**Business Value**:
- Complete pipeline validation before any data processing
- Immediate feedback on type mismatches in the entire chain
- Prevents wasted resources on invalid pipelines
- Catches configuration errors during development
### 3. Error Prevention at Multiple Levels
Oteilo prevents errors at three distinct stages:
1. **Construction Time** (When building your pipeline):
```python
from oteilo.pipeline import Pipeline, Step
from typing import Dict, Any
# This will fail immediately due to type mismatch
try:
pipeline = (Pipeline()
.then(lambda x: str(x)) # Output: str
.then(lambda x: x * 2)) # Expects int, not str
except PipelineError as e:
print("Pipeline construction failed: incompatible steps")
# This will fail due to missing type hints
try:
pipeline = Pipeline().then(lambda x: x * 2) # No type hints
except PipelineError as e:
print("Pipeline construction failed: missing type information")
```
2. **Runtime Validation** (When processing data):
```python
class PriceCalculator(Step):
"""Calculate prices with built-in validation."""
def calculate(self, data: Dict[str, Any]) -> Dict[str, Any]:
"""Calculate final price with discounts."""
# Types are checked automatically
base_price = data["price"]
quantity = data["quantity"]
discount = data.get("discount", 0)
total = (base_price * quantity) * (1 - discount)
return {"total": total}
calculator = PriceCalculator()
# Type validation happens automatically
result = calculator.calculate({
"price": 99.99,
"quantity": 5,
"discount": 0.1
})
# Invalid types are caught immediately
try:
result = calculator.calculate({
"price": "not a number", # Wrong type
"quantity": 5
})
except PipelineError as e:
print("Validation failed: incorrect data type")
```
3. **Business Logic Validation** (Your custom rules):
```python
def validate_order(order: Dict[str, Any]) -> Dict[str, Any]:
"""Validate business rules for orders."""
if order["quantity"] <= 0:
raise ValueError("Quantity must be positive")
if order["price"] <= 0:
raise ValueError("Price must be positive")
return order
# Business rules are checked during processing
pipeline = Pipeline().then(validate_order)
try:
result = pipeline.process({
"quantity": -1,
"price": 99.99
})
except PipelineError as e:
print("Business rule violated: quantity must be positive")
```
**Business Value**:
- Configuration errors are caught during development
- Data validation errors are caught before processing
- Business rule violations are clearly reported
- Each type of error has its own clear message
### 4. Performance Monitoring
Track performance of critical operations:
```python
from oteilo.timing import timing_block
def process_customer_data(customers):
with timing_block("customer_processing"):
for customer in customers:
with timing_block(f"processing_{customer['id']}"):
update_customer(customer)
process_orders(customer)
send_notification(customer)
# You'll see exactly how long each operation takes:
# DEBUG: customer_processing took 1.234 seconds
# DEBUG: processing_12345 took 0.123 seconds
# DEBUG: processing_12346 took 0.456 seconds
```
**Business Value**:
- Identify slow operations
- Track processing times
- Optimize performance bottlenecks
## Real-World Use Cases
### 1. Order Processing System
```python
from oteilo.pipeline import Pipeline
from oteilo.functional import trace
# Monitor critical business functions
@trace(level="INFO")
def validate_inventory(order):
"""Check if we have enough inventory."""
# Your inventory check logic here
return order
@trace(level="INFO")
def process_payment(order):
"""Process payment and return confirmation."""
# Your payment processing logic here
return {**order, "payment_status": "confirmed"}
# Build a reliable order pipeline
order_system = (Pipeline()
.then(validate_order) # Step 1: Validate order data
.then(validate_inventory) # Step 2: Check inventory
.then(calculate_total) # Step 3: Calculate prices
.then(process_payment) # Step 4: Process payment
.then(format_currency)) # Step 5: Format for display
# Process orders safely
try:
result = order_system.process({
"customer_id": "12345",
"product_id": "PROD-1",
"quantity": 5
})
print("Order processed successfully")
except Exception as e:
print(f"Order failed: {e}") # Clear error message
```
### 2. Data Transformation Pipeline
```python
from oteilo.pipeline import Pipeline, Step
from typing import Dict, Any
class DataCleaner(Step):
"""Clean and validate customer data."""
def clean(self, data: Dict[str, Any]) -> Dict[str, Any]:
"""Remove invalid characters and normalize format."""
return {
"name": data["name"].strip().title(),
"email": data["email"].lower().strip(),
"phone": self.format_phone(data.get("phone", ""))
}
class DataEnricher(Step):
"""Add additional customer information."""
def enrich(self, data: Dict[str, Any]) -> Dict[str, Any]:
"""Add derived data fields."""
return {
**data,
"customer_segment": self.calculate_segment(data),
"lifetime_value": self.calculate_ltv(data)
}
# Create a data processing pipeline
data_pipeline = (Pipeline()
.then(DataCleaner()) # Step 1: Clean data
.then(DataEnricher()) # Step 2: Add business data
.then(format_output)) # Step 3: Format for use
# Process customer data safely
try:
clean_data = data_pipeline.process({
"name": "john smith ",
"email": " John@Example.COM ",
"phone": "1234567890"
})
print("Data processed successfully")
except Exception as e:
print(f"Data processing failed: {e}")
```
## Requirements
- Python 3.12 or higher
- Additional packages are automatically installed
## Support
Need help? Have questions? Contact our support team at [support email/link].
## Contributing
We welcome contributions! See our [Contributing Guide](CONTRIBUTING.md) for details.
## License
MIT License - Feel free to use in your projects.
Raw data
{
"_id": null,
"home_page": null,
"name": "oteilo",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.12",
"maintainer_email": null,
"keywords": "io, logging, timing, utilities, validation",
"author": "Teilo Millet",
"author_email": null,
"download_url": "https://files.pythonhosted.org/packages/4b/8a/699ed4ffb28291f37eddb71b932404373b5cd17dd54f9857dcc151b82ad3/oteilo-0.1.0.tar.gz",
"platform": null,
"description": "# Oteilo\n\nA robust Python toolkit that helps your team write safer, more maintainable code with built-in monitoring and validation. Think of it as a safety net that catches errors early and provides clear insights into your application's behavior.\n\n## Why Oteilo?\n\n- **Catch Problems Early**: Automatically validates data types and catches errors before they reach production\n- **Clear Visibility**: Built-in monitoring shows exactly what your code is doing and how long it takes\n- **Maintainable Code**: Enforces clean, consistent patterns that make code easier to understand and modify\n- **Reduced Risk**: Strong type checking and validation prevent data-related bugs\n- **Developer Productivity**: Common patterns are pre-built, tested, and ready to use\n\n## Quick Start\n\n```bash\npip install oteilo\n```\n\n## Understanding Imports\n\nOteilo is organized into focused modules, each handling specific concerns. Here's how to import and use them:\n\n### Core Pipeline Components\n```python\nfrom oteilo.pipeline import Pipeline, Step\nfrom typing import Dict, Any # Python's typing for type hints\n\n# Use Pipeline when you need to:\n# - Chain multiple operations together\n# - Validate data types between steps\n# - Process data through a sequence of transformations\npipeline = Pipeline().then(step1).then(step2)\n\n# Use Step when you need to:\n# - Create reusable, configurable pipeline steps\n# - Add automatic type validation\n# - Provide clear documentation and metadata\nclass MyStep(Step):\n def process(self, data: Dict[str, Any]) -> Dict[str, Any]:\n return transformed_data\n```\n\n### Monitoring and Debugging\n```python\nfrom oteilo.functional import trace # For function monitoring\nfrom oteilo.timing import timing_block # For performance tracking\nfrom oteilo.logging import configure_logging # For setting up logging\n\n# Use @trace when you need to:\n# - Monitor function execution\n# - Track timing and performance\n# - Log input/output data\n@trace(level=\"DEBUG\")\ndef my_function(): ...\n\n# Use timing_block when you need to:\n# - Measure specific code blocks\n# - Track nested operations\n# - Profile performance\nwith timing_block(\"operation_name\"):\n do_something()\n```\n\n### Error Handling\n```python\nfrom oteilo.pipeline import PipelineError # For catching pipeline-specific errors\n\n# Use PipelineError when you need to:\n# - Distinguish pipeline errors from other exceptions\n# - Access detailed error information\n# - Handle different types of validation failures\ntry:\n result = pipeline.process(data)\nexcept PipelineError as e:\n print(f\"Error details: {e.details}\")\n print(f\"Original cause: {e.cause}\")\n```\n\n### Type Validation\n```python\nfrom oteilo.validation import validate_type # For manual type checking\nfrom typing import Dict, List, Any, Optional # Python's type hints\n\n# Use validate_type when you need to:\n# - Check types manually\n# - Handle complex type validation\n# - Support Union and Optional types\nif not validate_type(data, Dict[str, Any]):\n raise ValueError(\"Expected dictionary\")\n```\n\n### Complete Example\n```python\nfrom oteilo.pipeline import Pipeline, Step, PipelineError\nfrom oteilo.functional import trace\nfrom oteilo.timing import timing_block\nfrom oteilo.logging import configure_logging\nfrom oteilo.validation import validate_type\nfrom typing import Dict, Any, Optional\n\n# Set up logging first\nconfigure_logging(level=\"DEBUG\")\n\n# Create monitored functions\n@trace\ndef process_data(data: Dict[str, Any]) -> Dict[str, Any]:\n with timing_block(\"processing\"):\n # Your processing logic here\n return processed_data\n\n# Create pipeline steps\nclass ValidationStep(Step):\n def validate(self, data: Dict[str, Any]) -> Dict[str, Any]:\n # Validation logic here\n return validated_data\n\n# Build and use pipeline\npipeline = Pipeline().then(ValidationStep()).then(process_data)\n\n# Handle errors appropriately\ndef handle_request(data: Dict[str, Any]) -> Dict[str, Any]:\n try:\n return pipeline.process(data)\n except PipelineError as e:\n # Handle pipeline-specific errors\n return {\"error\": str(e), \"details\": e.details}\n except Exception as e:\n # Handle unexpected errors\n return {\"error\": \"Unexpected error\", \"details\": str(e)}\n```\n\n## Key Features (With Real-World Examples)\n\n### 1. Automatic Function Monitoring\n\nMonitor any function to see what's happening inside your application. The `@trace` decorator provides both timing and comprehensive monitoring:\n\n```python\nfrom oteilo.functional import trace\n\n# 1. Simple timing only (default)\n@trace\ndef calculate_total(order):\n return order[\"quantity\"] * order[\"price\"]\n\n# Output:\n# DEBUG: calculate_total took 0.001 seconds\n\n# 2. Full tracing with timing and monitoring\n@trace(level=\"DEBUG\")\ndef calculate_order_total(order):\n return order[\"quantity\"] * order[\"price\"]\n\n# Output:\n# DEBUG: calculate_order_total called with {'order': {'quantity': 5, 'price': 9.99}}\n# DEBUG: calculate_order_total type information {'params': {'order': 'Dict[str, Any]'}, 'return': 'float'}\n# DEBUG: Executing calculate_order_total\n# DEBUG: calculate_order_total took 0.001 seconds\n# DEBUG: calculate_order_total completed with result: 49.95\n\n# 3. Custom configuration\n@trace(\n level=\"INFO\", # Set logging level\n timing=True, # Include timing (default)\n args=True, # Log arguments (default)\n types=True, # Log type information (default)\n nested=False # Don't trace nested calls\n)\ndef process_order(order):\n validate_order(order)\n total = calculate_total(order)\n return format_output(total)\n\n# You can also combine @trace with timing_block for more granular timing:\n@trace(level=\"DEBUG\")\ndef process_large_order(order):\n with timing_block(\"validation\"):\n validate_order(order)\n \n with timing_block(\"calculation\"):\n total = calculate_total(order)\n \n with timing_block(\"formatting\"):\n return format_output(total)\n\n# Output:\n# DEBUG: process_large_order called with {'order': {...}}\n# DEBUG: Executing process_large_order\n# DEBUG: validation took 0.001 seconds\n# DEBUG: calculation took 0.002 seconds\n# DEBUG: formatting took 0.001 seconds\n# DEBUG: process_large_order took 0.004 seconds\n# DEBUG: process_large_order completed with result: {...}\n```\n\n**Business Value**: \n- Instantly see performance bottlenecks with automatic timing\n- Debug issues with detailed logging of inputs and outputs\n- Monitor critical business functions with type validation\n- Track nested operations and their timing\n- Flexible configuration to match your monitoring needs\n\n### 2. Safe Data Processing Pipeline\n\nBuild reliable data transformation workflows that validate every step:\n\n```python\nfrom oteilo.pipeline import Pipeline\nfrom typing import Dict, Any\n\n# Define each step of your process\ndef validate_order(order):\n \"\"\"Make sure we have all required order data.\"\"\"\n if not order.get(\"customer_id\"):\n raise ValueError(\"Missing customer ID\")\n return order\n\ndef calculate_total(order):\n \"\"\"Calculate order total with tax.\"\"\"\n subtotal = order[\"quantity\"] * order[\"price\"]\n tax = subtotal * 0.1\n return {**order, \"subtotal\": subtotal, \"tax\": tax, \"total\": subtotal + tax}\n\ndef format_currency(order):\n \"\"\"Format money values for display.\"\"\"\n return {\n **order,\n \"subtotal\": f\"${order['subtotal']:.2f}\",\n \"tax\": f\"${order['tax']:.2f}\",\n \"total\": f\"${order['total']:.2f}\"\n }\n\n# 1. First, define your pipeline - this validates type compatibility\norder_pipeline = (Pipeline()\n .then(validate_order) # Dict -> Dict\n .then(calculate_total) # Dict -> Dict\n .then(format_currency)) # Dict -> Dict\n\n# 2. Then use it to process data - this is where runtime checks happen\ndef process_order(order_data: Dict[str, Any]) -> Dict[str, Any]:\n \"\"\"Process an order with full validation.\"\"\"\n # Type validation and business rules are checked during processing\n return order_pipeline.process(order_data)\n\n# 3. Now you can handle errors at your application level\ndef handle_order_request(request):\n try:\n order_data = {\n \"customer_id\": \"12345\",\n \"quantity\": 5,\n \"price\": 9.99\n }\n result = process_order(order_data)\n return {\"status\": \"success\", \"data\": result}\n \n except PipelineError as e:\n if \"Invalid input type\" in str(e):\n return {\"status\": \"error\", \"message\": \"Invalid data format\"}\n if \"quantity must be positive\" in str(e):\n return {\"status\": \"error\", \"message\": \"Invalid quantity\"}\n return {\"status\": \"error\", \"message\": str(e)}\n\n# This is cleaner than mixing pipeline usage and error handling\nresponse = handle_order_request(request)\nif response[\"status\"] == \"success\":\n send_confirmation(response[\"data\"])\nelse:\n notify_user(response[\"message\"])\n```\n\n**Business Value**:\n- Complete pipeline validation before any data processing\n- Immediate feedback on type mismatches in the entire chain\n- Prevents wasted resources on invalid pipelines\n- Catches configuration errors during development\n\n### 3. Error Prevention at Multiple Levels\n\nOteilo prevents errors at three distinct stages:\n\n1. **Construction Time** (When building your pipeline):\n```python\nfrom oteilo.pipeline import Pipeline, Step\nfrom typing import Dict, Any\n\n# This will fail immediately due to type mismatch\ntry:\n pipeline = (Pipeline()\n .then(lambda x: str(x)) # Output: str\n .then(lambda x: x * 2)) # Expects int, not str\nexcept PipelineError as e:\n print(\"Pipeline construction failed: incompatible steps\")\n\n# This will fail due to missing type hints\ntry:\n pipeline = Pipeline().then(lambda x: x * 2) # No type hints\nexcept PipelineError as e:\n print(\"Pipeline construction failed: missing type information\")\n```\n\n2. **Runtime Validation** (When processing data):\n```python\nclass PriceCalculator(Step):\n \"\"\"Calculate prices with built-in validation.\"\"\"\n \n def calculate(self, data: Dict[str, Any]) -> Dict[str, Any]:\n \"\"\"Calculate final price with discounts.\"\"\"\n # Types are checked automatically\n base_price = data[\"price\"]\n quantity = data[\"quantity\"]\n discount = data.get(\"discount\", 0)\n \n total = (base_price * quantity) * (1 - discount)\n return {\"total\": total}\n\ncalculator = PriceCalculator()\n\n# Type validation happens automatically\nresult = calculator.calculate({\n \"price\": 99.99,\n \"quantity\": 5,\n \"discount\": 0.1\n})\n\n# Invalid types are caught immediately\ntry:\n result = calculator.calculate({\n \"price\": \"not a number\", # Wrong type\n \"quantity\": 5\n })\nexcept PipelineError as e:\n print(\"Validation failed: incorrect data type\")\n```\n\n3. **Business Logic Validation** (Your custom rules):\n```python\ndef validate_order(order: Dict[str, Any]) -> Dict[str, Any]:\n \"\"\"Validate business rules for orders.\"\"\"\n if order[\"quantity\"] <= 0:\n raise ValueError(\"Quantity must be positive\")\n if order[\"price\"] <= 0:\n raise ValueError(\"Price must be positive\")\n return order\n\n# Business rules are checked during processing\npipeline = Pipeline().then(validate_order)\ntry:\n result = pipeline.process({\n \"quantity\": -1,\n \"price\": 99.99\n })\nexcept PipelineError as e:\n print(\"Business rule violated: quantity must be positive\")\n```\n\n**Business Value**:\n- Configuration errors are caught during development\n- Data validation errors are caught before processing\n- Business rule violations are clearly reported\n- Each type of error has its own clear message\n\n### 4. Performance Monitoring\n\nTrack performance of critical operations:\n\n```python\nfrom oteilo.timing import timing_block\n\ndef process_customer_data(customers):\n with timing_block(\"customer_processing\"):\n for customer in customers:\n with timing_block(f\"processing_{customer['id']}\"):\n update_customer(customer)\n process_orders(customer)\n send_notification(customer)\n\n# You'll see exactly how long each operation takes:\n# DEBUG: customer_processing took 1.234 seconds\n# DEBUG: processing_12345 took 0.123 seconds\n# DEBUG: processing_12346 took 0.456 seconds\n```\n\n**Business Value**:\n- Identify slow operations\n- Track processing times\n- Optimize performance bottlenecks\n\n## Real-World Use Cases\n\n### 1. Order Processing System\n\n```python\nfrom oteilo.pipeline import Pipeline\nfrom oteilo.functional import trace\n\n# Monitor critical business functions\n@trace(level=\"INFO\")\ndef validate_inventory(order):\n \"\"\"Check if we have enough inventory.\"\"\"\n # Your inventory check logic here\n return order\n\n@trace(level=\"INFO\")\ndef process_payment(order):\n \"\"\"Process payment and return confirmation.\"\"\"\n # Your payment processing logic here\n return {**order, \"payment_status\": \"confirmed\"}\n\n# Build a reliable order pipeline\norder_system = (Pipeline()\n .then(validate_order) # Step 1: Validate order data\n .then(validate_inventory) # Step 2: Check inventory\n .then(calculate_total) # Step 3: Calculate prices\n .then(process_payment) # Step 4: Process payment\n .then(format_currency)) # Step 5: Format for display\n\n# Process orders safely\ntry:\n result = order_system.process({\n \"customer_id\": \"12345\",\n \"product_id\": \"PROD-1\",\n \"quantity\": 5\n })\n print(\"Order processed successfully\")\nexcept Exception as e:\n print(f\"Order failed: {e}\") # Clear error message\n```\n\n### 2. Data Transformation Pipeline\n\n```python\nfrom oteilo.pipeline import Pipeline, Step\nfrom typing import Dict, Any\n\nclass DataCleaner(Step):\n \"\"\"Clean and validate customer data.\"\"\"\n \n def clean(self, data: Dict[str, Any]) -> Dict[str, Any]:\n \"\"\"Remove invalid characters and normalize format.\"\"\"\n return {\n \"name\": data[\"name\"].strip().title(),\n \"email\": data[\"email\"].lower().strip(),\n \"phone\": self.format_phone(data.get(\"phone\", \"\"))\n }\n\nclass DataEnricher(Step):\n \"\"\"Add additional customer information.\"\"\"\n \n def enrich(self, data: Dict[str, Any]) -> Dict[str, Any]:\n \"\"\"Add derived data fields.\"\"\"\n return {\n **data,\n \"customer_segment\": self.calculate_segment(data),\n \"lifetime_value\": self.calculate_ltv(data)\n }\n\n# Create a data processing pipeline\ndata_pipeline = (Pipeline()\n .then(DataCleaner()) # Step 1: Clean data\n .then(DataEnricher()) # Step 2: Add business data\n .then(format_output)) # Step 3: Format for use\n\n# Process customer data safely\ntry:\n clean_data = data_pipeline.process({\n \"name\": \"john smith \",\n \"email\": \" John@Example.COM \",\n \"phone\": \"1234567890\"\n })\n print(\"Data processed successfully\")\nexcept Exception as e:\n print(f\"Data processing failed: {e}\")\n```\n\n## Requirements\n\n- Python 3.12 or higher\n- Additional packages are automatically installed\n\n## Support\n\nNeed help? Have questions? Contact our support team at [support email/link].\n\n## Contributing\n\nWe welcome contributions! See our [Contributing Guide](CONTRIBUTING.md) for details.\n\n## License\n\nMIT License - Feel free to use in your projects.",
"bugtrack_url": null,
"license": null,
"summary": "A collection of intuitive Python utilities for common development tasks",
"version": "0.1.0",
"project_urls": {
"Repository": "https://github.com/teilomillet/oteilo"
},
"split_keywords": [
"io",
" logging",
" timing",
" utilities",
" validation"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "58ede909b725507ad8154d2a7ced6c37afda772cf0d88966a411b1aeb3c74d43",
"md5": "6b1164b473de93e7379c5ae575bb2fb9",
"sha256": "9fc85ce8e2faa11a4e410844732e7d96c6bd887b26701f932196d2e75b0b043f"
},
"downloads": -1,
"filename": "oteilo-0.1.0-py3-none-any.whl",
"has_sig": false,
"md5_digest": "6b1164b473de93e7379c5ae575bb2fb9",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.12",
"size": 22484,
"upload_time": "2025-02-06T22:39:58",
"upload_time_iso_8601": "2025-02-06T22:39:58.126302Z",
"url": "https://files.pythonhosted.org/packages/58/ed/e909b725507ad8154d2a7ced6c37afda772cf0d88966a411b1aeb3c74d43/oteilo-0.1.0-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "4b8a699ed4ffb28291f37eddb71b932404373b5cd17dd54f9857dcc151b82ad3",
"md5": "679d2891767de50df4c77d78ff7731ae",
"sha256": "683d4125fc1829011ddbc6199d84da134a23ded09156fbbdaf7e2cc40748bf41"
},
"downloads": -1,
"filename": "oteilo-0.1.0.tar.gz",
"has_sig": false,
"md5_digest": "679d2891767de50df4c77d78ff7731ae",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.12",
"size": 38179,
"upload_time": "2025-02-06T22:40:00",
"upload_time_iso_8601": "2025-02-06T22:40:00.441644Z",
"url": "https://files.pythonhosted.org/packages/4b/8a/699ed4ffb28291f37eddb71b932404373b5cd17dd54f9857dcc151b82ad3/oteilo-0.1.0.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-02-06 22:40:00",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "teilomillet",
"github_project": "oteilo",
"github_not_found": true,
"lcname": "oteilo"
}
