vein package¶

Subpackages¶

Submodules¶

vein.exceptions module¶

exception vein.exceptions.FileReadError[source]¶

Bases: Exception

Exception raised when a file cannot be read.

exception vein.exceptions.FlowItemValidationError(message: str)[source]¶

Bases: Exception

Exception raised when a flow item value fails validation.

exception vein.exceptions.InvalidOperationError(message: str = 'Invalid operation performed.')[source]¶

Bases: Exception

Exception raised when an invalid operation is performed.

exception vein.exceptions.MaxRetriesExceededError(backoff_policy, total_time_slept: float)[source]¶

Bases: Exception

Exception raised when the maximum number of retries is exceeded in exponential backoff.

exception vein.exceptions.MissingCacheKeyError(key)[source]¶

Bases: Exception

Exception raised when a cache key doesn’t exist.

exception vein.exceptions.MissingDependencyError(dependency: str)[source]¶

Bases: Exception

Exception raised when a dependency is missing.

exception vein.exceptions.MissingEnvironmentVariableError(key)[source]¶

Bases: Exception

Exception raised when an environment variable doesn’t exist.

exception vein.exceptions.MissingFlowKeyError(key)[source]¶

Bases: Exception

Exception raised when a flow key doesn’t exist.

exception vein.exceptions.NetworkError[source]¶

Bases: Exception

Exception raised when a flow key doesn’t exist.

Module contents¶

vein - Modern flow management for Python applications.

Created by Josh Breidinger (2025) Author: Josh Breidinger Email: company@breisoft.com GitHub: https://github.com/breisoft

Born from the frustration of serverless flow hell, vein makes flow management simple, fast, and reliable.

vein.needs_value(func: Callable[[...], Any], flow_value: Any, *args: Any, verbose: bool = False, **kwargs: Any) → bool | None[source]¶

Test whether a function requires explicit .value access or works with FastCFG magic.

This helper function determines if you can pass a FastCFG flow value directly to a function, or if you need to use .value to extract the underlying value. This is particularly useful for identifying C extensions that require exact types.

Parameters:

  • func – The function to test compatibility with

  • flow_value – A FastCFG wrapped value (e.g., flow.port)

  • *args – Additional positional arguments to pass to func (after flow_value)

  • verbose – If True, prints diagnostic information about the test

  • **kwargs – Keyword arguments to pass to func

Returns:

  • False: Function works with FastCFG magic (no .value needed)

  • True: Function requires .value to be called on flow_value

  • None: Function doesn’t work even with .value (incompatible)

Return type:

bool or None

Example

flow.size = 10

# Check if np.zeros needs .value if needs_value(np.zeros, flow.size):

arr = np.zeros(flow.size.value)

else:

arr = np.zeros(flow.size)

# With verbose output for debugging needs_value(np.zeros, flow.size, verbose=True) # Prints: zeros requires .value - use flow.size.value

vein.refresh(target: Flow | AbstractFlowItem)[source]¶

Manually refresh a Flow or FlowItem to detect external changes.

Parameters:

target – Flow object or FlowItem to refresh

Example

# In your own loop for _ in range(60):

refresh(flow.my_var) # Triggers change detection time.sleep(1)