Easy Flow
py_simple.easy_flow
easy_flow is built to simplify work flows
EasyFlowError
Bases: Exception
Raised when a Python file can't be run.
Wraps the underlying error (missing file, bad permissions, an exception raised inside the file being run, etc.) so py_simple functions can fail with one consistent, easy-to-read exception instead of a random builtin one.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
message
|
str
|
Human-readable description of what went wrong. |
required |
retry(func, attempts=3, delay=1)
Calls a function, automatically retrying it if it raises an exception up to a specified number of attempts, with a pause between tries.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
func
|
callable
|
The function to execute. |
required |
attempts
|
int
|
Maximum number of times to try running the function. Defaults to 3. |
3
|
delay
|
int or float
|
Time to wait in seconds between failed attempts. Defaults to 1. |
1
|
Returns:
| Name | Type | Description |
|---|---|---|
Any |
The return value of |
Raises:
| Type | Description |
|---|---|
Exception
|
The last exception raised by |
Example
from py_simple import retry
def flaky_api():
# Might fail sometimes
pass
result = retry(flaky_api, attempts=5, delay=2)
import time
def flaky_api():
pass
attempts = 5
delay = 2
for i in range(attempts):
try:
result = flaky_api()
break
except Exception as e:
if i == attempts - 1:
raise e
time.sleep(delay)
run_if(condition, func, *args, default_value=None, **kwargs)
Runs a function only when a condition is true.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
condition
|
bool
|
Whether to run the function. |
required |
func
|
callable
|
The function to execute when condition is True. |
required |
*args
|
Positional arguments to pass to the function. |
()
|
|
default_value
|
Any
|
Value to return when condition is False. Defaults to None. |
None
|
**kwargs
|
Keyword arguments to pass to the function. |
{}
|
Returns:
| Name | Type | Description |
|---|---|---|
Any |
The function's return value when condition is True, or the default value when condition is False. |
Example
from py_simple import run_if
result = run_if(True, max, 3, 8) # -> 8
skipped = run_if(False, max, 3, 8, default_value=0) # -> 0
if should_run:
result = max(3, 8)
else:
result = 0
run_py_file(filename)
Runs a Python file as if it were called directly from the command
line (i.e. as __main__), using the current process.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
filename
|
str
|
Path to the |
required |
Example
from py_simple import run_py_file
run_py_file("script.py")
import runpy
print("RUNNING: script.py")
try:
runpy.run_path("script.py")
except Exception as e:
print(f"Couldn't run the file: {e}")
run_py_file_safe(filename)
Runs a Python file as if it were called directly from the command
line (i.e. as __main__), returning a success flag instead of
raising an exception if something goes wrong.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
filename
|
str
|
Path to the |
required |
Returns:
| Name | Type | Description |
|---|---|---|
tuple |
|
Example
from py_simple import run_py_file_safe
success, error = run_py_file_safe("script.py")
if not success:
print(f"Couldn't run the file: {error}")
import runpy
print("RUNNING: script.py")
try:
runpy.run_path("script.py")
except Exception as e:
print(f"Couldn't run the file: {e}")
run_py_string(code_string)
Executes a string of Python code in the current global scope, saving you from writing temporary file creation boilerplate.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
code_string
|
str
|
Valid Python code as a string to execute. |
required |
Returns:
| Type | Description |
|---|---|
None
|
None |
Raises:
| Type | Description |
|---|---|
EasyFlowError
|
If executing the code string raises an exception. |
Example
from py_simple import run_py_string
run_py_string("print('Hello from string!')")
try:
exec("print('Hello from string!')")
except Exception as e:
print(f"Execution failed: {e}")
run_with_delay(delay, func, *args, **kwargs)
Waits for a specified number of seconds before executing a function and returning its result.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
delay
|
int or float
|
Time to wait in seconds before running the function. |
required |
func
|
callable
|
The function to execute. |
required |
*args
|
Positional arguments to pass to the function. |
()
|
|
**kwargs
|
Keyword arguments to pass to the function. |
{}
|
Returns:
| Name | Type | Description |
|---|---|---|
Any |
The return value of |
Example
from py_simple import run_with_delay
result = run_with_delay(1, print, "Hello after 1 second!")
import time
time.sleep(1)
result = print("Hello after 1 second!")
run_with_fallback(func, default_value, *args, **kwargs)
Executes a function and returns its result, or returns a default fallback value if an exception is raised.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
func
|
callable
|
The function to execute. |
required |
default_value
|
Any
|
The value to return if the function fails. |
required |
*args
|
Positional arguments to pass to the function. |
()
|
|
**kwargs
|
Keyword arguments to pass to the function. |
{}
|
Returns:
| Name | Type | Description |
|---|---|---|
Any |
The function's return value or the default fallback value. |
Example
from py_simple import run_with_fallback
result = run_with_fallback(int, 0, "not_a_number")
print(result) # -> 0
try:
result = int("not_a_number")
except Exception:
result = 0
print(result) # -> 0
time_function_call(function, args=None)
Runs a function once and returns how long it took to run, in seconds.
Raises EasyFlowError if the function raises an exception while running.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
function
|
The function to run and time. |
required | |
args
|
list
|
Positional arguments to pass to the function. Leave as None to call it with no arguments. |
None
|
Returns:
| Name | Type | Description |
|---|---|---|
float |
float
|
Number of seconds the function took to run. |
Example
from py_simple import time_function_call
def add(a, b):
return a + b
time_function_call(add, [2, 3]) # -> 0.000002
import time
def add(a, b):
return a + b
start = time.time()
try:
add(2, 3)
except Exception as e:
print(f"Couldn't time the function: {e}")
duration = time.time() - start
time_it(func)
Decorator that measures how long a function takes to run, prints the elapsed time, and returns the original result.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
func
|
callable
|
The function to decorate. |
required |
Returns:
| Name | Type | Description |
|---|---|---|
callable |
A wrapped version of |
Example
from py_simple import time_it
@time_it
def add(a, b):
return a + b
add(2, 3) # prints: add took 0.00s
import time
def add(a, b):
return a + b
start = time.time()
result = add(2, 3)
elapsed = time.time() - start
print(f"add took {elapsed:.2f}s")
wait_until(condition, timeout=5, interval=0.1)
Waits until a condition function returns True, or until a timeout is reached.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
condition
|
callable
|
A function that returns True when the waiting should stop. |
required |
timeout
|
int or float
|
Maximum number of seconds to
wait. Defaults to |
5
|
interval
|
int or float
|
Seconds to wait between
checks. Defaults to |
0.1
|
Returns:
| Name | Type | Description |
|---|---|---|
bool |
bool
|
True if the condition became true before the timeout, otherwise False. |
Example
from py_simple import wait_until
ready = wait_until(lambda: file_exists("report.csv"), timeout=10)
import time
start = time.time()
ready = False
while time.time() - start < 10:
if file_exists("report.csv"):
ready = True
break
time.sleep(0.1)