Dataclass for SDK configuration.
from botanu.sdk.config import BotanuConfig| Field | Type | Default | Description |
|---|---|---|---|
service_name |
str |
From env / "unknown_service" |
Service name |
service_version |
str |
From env | Service version |
service_namespace |
str |
From env | Service namespace |
deployment_environment |
str |
From env / "production" |
Deployment environment |
auto_detect_resources |
bool |
True |
Auto-detect cloud resources |
otlp_endpoint |
str |
From env / "http://localhost:4318" |
OTLP endpoint |
otlp_headers |
dict |
None |
Custom headers for OTLP exporter |
max_export_batch_size |
int |
512 |
Max spans per batch |
max_queue_size |
int |
65536 |
Max spans in queue (~64 MB at ~1 KB/span) |
schedule_delay_millis |
int |
5000 |
Delay between batch exports |
export_timeout_millis |
int |
30000 |
Timeout for export operations |
propagation_mode |
str |
"lean" |
"lean" or "full" |
auto_instrument_packages |
list |
See below | Packages to auto-instrument |
config = BotanuConfig(
service_name="my-service",
deployment_environment="production",
otlp_endpoint="http://collector:4318",
)Load configuration from a YAML file.
@classmethod
def from_yaml(cls, path: Optional[str] = None) -> BotanuConfigParameters:
path: Path to YAML config file
Raises:
FileNotFoundError: If config file does not existValueError: If YAML is malformedImportError: If PyYAML is not installed
Example:
config = BotanuConfig.from_yaml("config/botanu.yaml")Load config from file if it exists, otherwise use environment variables.
@classmethod
def from_file_or_env(cls, path: Optional[str] = None) -> BotanuConfigSearch order:
- Explicit
pathargument BOTANU_CONFIG_FILEenvironment variable./botanu.yaml./botanu.yml./config/botanu.yaml./config/botanu.yml- Falls back to environment-only config
Example:
# Auto-discovers config file
config = BotanuConfig.from_file_or_env()
# Explicit path
config = BotanuConfig.from_file_or_env("my-config.yaml")Export configuration as dictionary.
def to_dict(self) -> Dict[str, Any]Example:
config = BotanuConfig(service_name="my-service")
print(config.to_dict())
# {
# "service": {"name": "my-service", ...},
# "otlp": {"endpoint": "...", ...},
# ...
# }service:
name: string # Service name
version: string # Service version
namespace: string # Service namespace
environment: string # Deployment environment
resource:
auto_detect: boolean # Auto-detect cloud resources
otlp:
endpoint: string # OTLP endpoint URL
headers: # Custom headers
header-name: value
export:
batch_size: integer # Max spans per batch
queue_size: integer # Max spans in queue
delay_ms: integer # Delay between exports
export_timeout_ms: integer # Export timeout
propagation:
mode: string # "lean" or "full"
auto_instrument_packages: # List of packages to instrument
- package_nameservice:
name: ${OTEL_SERVICE_NAME:-default-service}
environment: ${ENVIRONMENT}
otlp:
endpoint: ${COLLECTOR_URL:-http://localhost:4318}
headers:
Authorization: Bearer ${API_TOKEN}Syntax:
${VAR_NAME}- Required variable${VAR_NAME:-default}- Variable with default value
Bootstrap function to initialise the SDK.
from botanu import enable
enable(
service_name: Optional[str] = None,
otlp_endpoint: Optional[str] = None,
environment: Optional[str] = None,
auto_instrumentation: bool = True,
propagators: Optional[List[str]] = None,
log_level: str = "INFO",
config: Optional[BotanuConfig] = None,
config_file: Optional[str] = None,
) -> bool| Parameter | Type | Default | Description |
|---|---|---|---|
service_name |
str |
From env | Service name |
otlp_endpoint |
str |
From env | OTLP endpoint URL |
environment |
str |
From env | Deployment environment |
auto_instrumentation |
bool |
True |
Enable auto-instrumentation |
propagators |
list[str] |
["tracecontext", "baggage"] |
Propagator list |
log_level |
str |
"INFO" |
Logging level |
config |
BotanuConfig |
None |
Pre-built configuration (overrides individual params) |
config_file |
str |
None |
Path to YAML config file |
True if successfully initialised, False if already initialised.
- Creates/merges
BotanuConfig - Configures
TracerProviderwithRunContextEnricher - Sets up OTLP exporter
- Enables auto-instrumentation (if requested)
- Configures W3C Baggage propagation
from botanu import enable
enable(service_name="my-service")from botanu import enable
from botanu.sdk.config import BotanuConfig
config = BotanuConfig.from_yaml("config/botanu.yaml")
enable(config=config)from botanu import enable
# Reads OTEL_SERVICE_NAME, OTEL_EXPORTER_OTLP_ENDPOINT, etc.
enable()Disable the SDK and clean up resources.
from botanu import disable
disable() -> None- Flushes pending spans
- Shuts down span processors
- Disables instrumentation
Check if the SDK is currently enabled.
from botanu import is_enabled
is_enabled() -> boolif not is_enabled():
enable(service_name="my-service")| Variable | Description | Default |
|---|---|---|
OTEL_SERVICE_NAME |
Service name | "unknown_service" |
OTEL_SERVICE_VERSION |
Service version | None |
OTEL_SERVICE_NAMESPACE |
Service namespace | None |
OTEL_DEPLOYMENT_ENVIRONMENT |
Deployment environment | "production" |
OTEL_EXPORTER_OTLP_ENDPOINT |
OTLP base endpoint | "http://localhost:4318" |
OTEL_EXPORTER_OTLP_TRACES_ENDPOINT |
OTLP traces endpoint (full URL) | None |
OTEL_EXPORTER_OTLP_HEADERS |
OTLP headers (key=value pairs) | None |
| Variable | Description | Default |
|---|---|---|
BOTANU_ENVIRONMENT |
Fallback for environment | "production" |
BOTANU_PROPAGATION_MODE |
"lean" or "full" |
"lean" |
BOTANU_AUTO_DETECT_RESOURCES |
Auto-detect cloud resources | "true" |
BOTANU_CONFIG_FILE |
Path to YAML config file | None |
BOTANU_COLLECTOR_ENDPOINT |
Override for OTLP endpoint | None |
BOTANU_MAX_QUEUE_SIZE |
Override max queue size | 65536 |
BOTANU_MAX_EXPORT_BATCH_SIZE |
Override max batch size | 512 |
BOTANU_EXPORT_TIMEOUT_MILLIS |
Override export timeout | 30000 |
Model for run metadata. Created automatically by @botanu_workflow and
run_botanu.
from botanu.models.run_context import RunContextCreate a new run context.
@classmethod
def create(
cls,
workflow: str,
event_id: str,
customer_id: str,
workflow_version: Optional[str] = None,
environment: Optional[str] = None,
tenant_id: Optional[str] = None,
parent_run_id: Optional[str] = None,
root_run_id: Optional[str] = None,
attempt: int = 1,
retry_of_run_id: Optional[str] = None,
deadline_seconds: Optional[float] = None,
) -> RunContextCreate a retry context from a previous run.
@classmethod
def create_retry(cls, previous: RunContext) -> RunContextReconstruct context from baggage dictionary.
@classmethod
def from_baggage(cls, baggage: Dict[str, str]) -> Optional[RunContext]Serialise to baggage format.
def to_baggage_dict(self, lean_mode: Optional[bool] = None) -> Dict[str, str]Serialise to span attributes.
def to_span_attributes(self) -> Dict[str, Union[str, float, int, bool]]Mark the run as complete.
def complete(
self,
status: RunStatus,
reason_code: Optional[str] = None,
error_class: Optional[str] = None,
value_type: Optional[str] = None,
value_amount: Optional[float] = None,
confidence: Optional[float] = None,
) -> Nonedef is_past_deadline(self) -> booldef is_cancelled(self) -> booldef request_cancellation(self, reason: str = "user") -> Nonedef remaining_time_seconds(self) -> Optional[float]| Field | Type | Description |
|---|---|---|
run_id |
str |
Unique UUIDv7 identifier |
workflow |
str |
Workflow name |
event_id |
str |
Business event identifier |
customer_id |
str |
Customer identifier |
environment |
str |
Deployment environment |
workflow_version |
str |
Version hash |
tenant_id |
str |
Tenant identifier |
parent_run_id |
str |
Parent run ID |
root_run_id |
str |
Root run ID (same as run_id for first attempt) |
attempt |
int |
Attempt number |
retry_of_run_id |
str |
Run ID of the previous attempt |
start_time |
datetime |
Run start time |
deadline |
float |
Absolute deadline (epoch seconds) |
cancelled |
bool |
Whether the run is cancelled |
outcome |
RunOutcome |
Recorded outcome |
Enum for run outcome status.
from botanu.models.run_context import RunStatus
class RunStatus(str, Enum):
SUCCESS = "success"
FAILURE = "failure"
PARTIAL = "partial"
TIMEOUT = "timeout"
CANCELED = "canceled"- Configuration Guide - Configuration how-to
- Architecture - SDK design
- Existing OTel Setup - Integration patterns