Serialization
- YAML encoding/decoding with custom tags
- JSON via orjson for speed
- TOML support
- Base64 with wrapping
Extended Data provides four main packages that work together or independently:
# Install all Extended Data packagespip install extended-data-types directed-inputs-class lifecyclelogging vendor-connectors# Core data type utilitiespip install extended-data-types
# Input handlingpip install directed-inputs-class
# Structured loggingpip install lifecyclelogging
# Vendor API connectorspip install vendor-connectors# Vendor connectors with all integrationspip install vendor-connectors[all]
# Or specific extraspip install vendor-connectors[aws] # AWS boto3pip install vendor-connectors[meshy] # Meshy AI 3Dpip install vendor-connectors[mcp] # MCP server supportpip install vendor-connectors[webhooks] # Webhook handlingInstall the core package
pip install extended-data-typesUse utilities in your code
from extended_data_types import encode_yaml, decode_yaml, is_nothing
# Serialize dataconfig = {"name": "myapp", "debug": True}yaml_str = encode_yaml(config)print(yaml_str)# name: myapp# debug: true
# Check for empty/null valuesif not is_nothing(config.get("name")): print("Name is set!")Add logging
pip install lifecycleloggingfrom lifecyclelogging import Logging
logger = Logging(enable_console=True)logger.logged_statement("Application started", log_level="info")Add input handling
pip install directed-inputs-classfrom directed_inputs_class import directed_inputs
@directed_inputs(from_env=True)class Config: def get_setting(self, debug: bool = False, port: int = 8080): return {"debug": debug, "port": port}
# DEBUG=true PORT=3000 python app.pyconfig = Config()print(config.get_setting()) # {"debug": True, "port": 3000}Core utility library providing:
Serialization
Transformations
Data Structures
State Utilities
is_nothing() — Check for empty/nullfirst_non_empty() — Coalesce valuesall_non_empty() / any_non_empty()from extended_data_types import ( # Serialization encode_yaml, decode_yaml, encode_json, decode_json, encode_toml, decode_toml, base64_encode, base64_decode,
# String transformations to_camel_case, to_snake_case, to_kebab_case, to_pascal_case, humanize, titleize, pluralize, singularize, ordinalize,
# Data structures deep_merge, flatten_map, flatten_list, filter_map, filter_list,
# State is_nothing, first_non_empty, all_non_empty, any_non_empty,)Transparent input handling with:
@directed_inputs on any classbool, int, float, Path, datetime@input_config for fine-grained controlimport osfrom directed_inputs_class import directed_inputs, input_config
os.environ["API_KEY"] = "secret123"os.environ["TIMEOUT"] = "30"
@directed_inputs(from_env=True)class Service: @input_config("api_key", required=True) def call_api(self, api_key: str, timeout: int = 60) -> dict: # api_key loaded from API_KEY (required) # timeout loaded from TIMEOUT, coerced to int return {"key": api_key[:4], "timeout": timeout}
service = Service()print(service.call_api()) # {"key": "secr", "timeout": 30}Enhanced Python logging with:
from lifecyclelogging import Logging
logger = Logging( enable_console=True, enable_file=True, log_file_path="app.log", enable_verbose_output=True, verbosity_threshold=2,)
# Standard logginglogger.logged_statement("Starting up", log_level="info")
# With context and datalogger.logged_statement( "Request processed", context_marker="API", json_data={"method": "POST", "status": 200}, log_level="info")
# Stored for later retrievallogger.logged_statement( "Error occurred", storage_marker="ERRORS", log_level="error")errors = logger.stored_messages["ERRORS"]Universal connectors with three interfaces each:
| Connector | Direct API | LangChain Tools | MCP Server |
|---|---|---|---|
| AWS | ✅ | — | — |
| Google Cloud | ✅ | — | — |
| GitHub | ✅ | ✅ | — |
| Slack | ✅ | — | — |
| Vault | ✅ | — | — |
| Zoom | ✅ | — | — |
| Meshy AI | ✅ | ✅ | ✅ |
| Anthropic | ✅ | — | — |
| Cursor | ✅ | — | — |
from vendor_connectors import VendorConnectors
vc = VendorConnectors()
# Cloudaws_s3 = vc.get_aws_client("s3")gcp = vc.get_google_client()
# Servicesgithub = vc.get_github_client(github_owner="myorg")slack = vc.get_slack_client()
# AIanthropic = vc.get_anthropic_client()cursor = vc.get_cursor_client()| Variable | Package | Description |
|---|---|---|
ANTHROPIC_API_KEY | vendor-connectors | Anthropic Claude API |
CURSOR_API_KEY | vendor-connectors | Cursor Background Agents |
AWS_ACCESS_KEY_ID | vendor-connectors | AWS credentials |
AWS_SECRET_ACCESS_KEY | vendor-connectors | AWS credentials |
GITHUB_TOKEN | vendor-connectors | GitHub API |
SLACK_BOT_TOKEN | vendor-connectors | Slack bot |
VAULT_ADDR | vendor-connectors | HashiCorp Vault |
MESHY_API_KEY | vendor-connectors | Meshy AI |
Extended Data Types
Deep dive into serialization, transformations, and utilities.
Directed Inputs
Master the decorator API and input handling.
Lifecycle Logging
Configure rich logging for your applications.
Vendor Connectors
Connect to cloud providers and AI services.