Flow Architecture¶
vein’s configuration management is built on a powerful state management engine called Flow. This document explains the technical architecture and key innovations that make Flow so powerful.
The Flow Engine¶
Flow is the core state management primitive that powers everything in vein. It’s designed to be:
Transparent: Values behave like regular Python objects 99% of the time
Performant: Highly optimized with minimal overhead
Extensible: Easy to add new state types and behaviors
Type-safe: Preserves types while adding functionality
from vein import config # Built on Flow
# Flow handles the complexity transparently
config.api_key = "secret" # Real-time updates
config.database.host = "localhost" # Nested state
config.on_change("api_key", handle_key_change) # Event system
Core Components¶
Flow is built from several key components that work together:
Flow Core¶
The main Flow class (vein.flow.core.Flow) is surprisingly simple:
class Flow(AbstractFlowUnit):
def __init__(self, **kwargs):
attributes = FlowAttributes(self)
self.__dict__["__attributes"] = attributes
self.__dict__["__interface"] = FlowInterface(self, attributes)
for k, v in kwargs.items():
if isinstance(v, dict):
v = create_flow_dict(v)
self.__setattr__(k, v)
The magic happens in the attribute access methods:
def __getattribute__(self, name):
if name.startswith("_"):
return object.__getattribute__(self, name)
interface = object.__getattribute__(self, "__interface")
if hasattr(interface, name):
return getattr(interface, name)
attributes = object.__getattribute__(self, "__attributes")
attr = attributes.get_attribute(name)
if isinstance(attr, Flow):
return attr # Return nested Flow directly
else:
return ValueWrapper.factory(attr) # Wrap other values
FlowAttributes¶
FlowAttributes manages the actual storage and retrieval of flow values:
class FlowAttributes(AbstractFlowUnit):
def __init__(self, flow: "Flow"):
self.__attributes: dict[str, AbstractFlowItem] = {}
self._flow: "Flow" = flow
def add_or_update_attribute(self, name: str, value: Any) -> None:
if isinstance(value, dict):
value = create_flow_dict(value) # Convert dicts to nested Flow
if name in self.__attributes:
flow_item = self.__attributes[name]
if isinstance(flow_item, BuiltInFlowItem):
flow_item.value = value # Update existing
else:
flow_item = self._add_attribute(name, value) # Replace
else:
flow_item = self._add_attribute(name, value) # New attribute
flow_item.validate() # Trigger validation
FlowInterface¶
FlowInterface handles the public API and environment management:
class FlowInterface(ValidatableMixin, EventListenerMixin, AbstractFlowUnit):
def __init__(self, flow, flow_attributes, **kwargs):
self._flow = flow
self._flow_attributes = flow_attributes
self._current_env = None
def update(self, other=None, **kwargs) -> "FlowInterface":
# Deep merge updates that preserve nested values
if self._current_env:
target = getattr(self._flow, self._current_env)
else:
target = self._flow
if other is not None:
deep_merge_flow(target, other)
if kwargs:
deep_merge_flow(target, kwargs)
return self # Method chaining
ValueWrapper: The Magic Ingredient¶
ValueWrapper is the key innovation that makes Flow values behave like regular Python objects:
class ValueWrapper(OperatorsMixin):
def __init__(self, item: AbstractFlowItem):
self._item = item
def __getattr__(self, name):
# First check if it's a Flow method (like add_validator)
if hasattr(self._item, name) and not name.startswith("_"):
return getattr(self._item, name)
# Otherwise delegate to the underlying value
value = self._item.value
return getattr(value, name)
@property
def __class__(self):
# This is the key trick - make isinstance() work!
return type(self._item.value)
OperatorsMixin: Complete Transparency¶
The OperatorsMixin implements ALL Python operators, making wrapped values indistinguishable from regular values:
class OperatorsMixin:
def __lt__(self, other):
return operator.lt(self._item.value, self._unwrap_if_wrapper(other))
def __add__(self, other):
return operator.add(self._item.value, self._unwrap_if_wrapper(other))
def __str__(self):
return str(self._item.value)
# ... implements ALL operators (50+ methods)
@property
def __class__(self):
return type(self._item.value) # Fool isinstance()!
This means Flow values work seamlessly with any Python code:
config.port = 8080
config.timeout = 30
# All of these just work!
total_time = config.timeout * config.retries # Math
if config.port > 8000: # Comparison
config.debug = True
message = f"Port: {config.port}" # String formatting
isinstance(config.port, int) # Type checking
Key Innovations¶
Flow introduces several key innovations that make it unique:
Transparent Wrapping: Values behave like regular Python objects 99% of the time, only requiring .value for external libraries that bypass Python’s protocols.
Nested Flow Objects: Dictionaries are automatically converted to nested Flow objects, enabling infinite nesting with full functionality at every level.
Dual Interface: Values can be accessed both as regular Python objects AND as Flow objects with methods like add_validator() and on_change().
Environment Management: Built-in support for environment-specific state with automatic fallbacks and inheritance.
Event-Driven Architecture: Real-time change notifications with pattern matching and wildcard support.
Validation Integration: Seamless validation that triggers on value changes without performance overhead.
Type Preservation: Full type safety with isinstance() working correctly on wrapped values.
Live Flow Items¶
Flow supports both static and dynamic values through the AbstractFlowItem system:
class BuiltInFlowItem(AbstractFlowItem):
"""Static values stored in memory"""
def _get_value(self) -> Any:
return self._value
class LiveFlowItem(AbstractFlowItem):
"""Dynamic values fetched on access"""
def _get_value(self) -> Any:
return self._state_tracker.get_state()
This enables powerful patterns like:
# Static configuration
config.api_key = "secret"
# Live configuration from AWS
config.secret_key = from_secrets_manager("prod/api-key")
# Both work identically!
print(config.api_key) # "secret"
print(config.secret_key) # Fetched from AWS
config.secret_key.add_validator(RequiredValidator()) # Validation works on both
Performance Characteristics¶
Flow is designed for high performance:
Minimal Overhead: ValueWrapper adds only ~0.5μs per access
Lazy Evaluation: Live values are only fetched when accessed
Intelligent Caching: Built-in optional caching with TTL and LRU strategies
Exponential Backoff: Built-in optional exponential backoff with configurable policies
Memory Efficient: Shared state across multiple Flow instances
Using Flow Directly¶
Flow can be used for any type of state management:
from vein import Flow
# Application state
app_state = Flow()
app_state.current_user = {"id": 123, "name": "Alice"}
app_state.feature_flags = {"dark_mode": True, "beta": False}
# User session
user_session = Flow()
user_session.last_activity = datetime.now()
user_session.preferences = {"theme": "dark", "notifications": True}
# All the same patterns work!
app_state.on_change("current_user", handle_user_change)
user_session.feature_flags.add_validator(PydanticValidator(FeatureFlags))
# Nested state works everywhere
app_state.users = {
"alice": {"id": 123, "role": "admin"},
"bob": {"id": 456, "role": "user"}
}
# Access nested state naturally
print(app_state.users.alice.role) # "admin"
app_state.users.alice.role = "super_admin" # Triggers events
The Future of Flow¶
Flow is designed to be the foundation for all state management in Python applications. As vein evolves, you’ll see Flow powering:
Application State: Global application state with real-time updates
User Sessions: Per-user state with automatic cleanup
Feature Flags: Dynamic feature management with percentage rollouts
Caching: Intelligent caching with automatic invalidation
Configuration: Environment-specific configuration management
Event Sourcing: Time-travel debugging and state replay
Distributed State: Multi-service state synchronization
The beauty of Flow is that all these use cases share the same patterns, performance characteristics, and developer experience. Once you learn Flow, you can use it for any state management need.
Welcome to the future of Python state management. 🩸⚡️