Reputation: 81
I'm having a hard time finding the correct library or a way to view the context as needed. I've looked here, but no dice so far: https://docs.python.org/3/library/contextlib.html.
I would want to:
class MyObj:
def __init__(self):
self.prop = "some property"
print("init MyObj")
def __enter__(self):
print("enter MyObj")
def __exit__(self, exc_type, exc_value, exc_traceback):
print("exit MyObj")
def lower_func(text):
if "This func is called from a block where an instance of MyObj is entered":
# Print MyObj.prop here
pass
print(text)
def higher_func():
with MyObj():
lower_func("Printing this while MyObj is entered!")
lower_func("Printing this while MyObj is exited!")
Is there a way to do that?
Currently, higher_func() is outputting:
init MyObj
enter MyObj
Printing this while MyObj is entered!
exit MyObj
Printing this while MyObj is exited!
What I want it to output:
init MyObj
enter MyObj
some property
Printing this while MyObj is entered!
exit MyObj
Printing this while MyObj is exited!
The goal is to implement this in a logging type of a function where it logs some information when it's available. Also to make the implementation as straight forward as possible. Basically it should work like a custom scope, where if MyObj is in the "entered" state, the lower-level functions would behave differently.
Upvotes: 1
Views: 57
Reputation: 81
What I was looking for is exactly what is described in PEP 555, but unfortunately the proposal was withdrawn and never implemented in that state.
What I did find functionally similar was PEP 567 / contextvars and that solves my problem. In my use case, I'd just have to make the ContextVar global.
A quick demonstration, which shows that the custom context is also thread-safe. They get evaluated in one order, and called in another. my_vars
's value is never explicitly handed down to lower_func
:
import contextvars
import threading
import random
import time
print_lock = threading.Lock()
my_vars = contextvars.ContextVar("my_vars", default={})
def lower_func(expected_str, wait_time):
context = contextvars.copy_context()
# Checks if var exists in context and has a non-empty value
if my_vars in context and context[my_vars]:
with print_lock:
print(f"{expected_str}, "
f"Actual: "
f"a={context[my_vars]['a']}, "
f"b={context[my_vars]['b']}, "
f"waited {wait_time}s")
def worker():
wait_time, a, b = random.randint(0, 9), random.randint(0, 9), random.randint(0, 9)
expected_str = f"Expected: a={a}, b={b}" # Some values to cross-check against
token = my_vars.set({"a": a, "b": b})
time.sleep(wait_time)
lower_func(expected_str, wait_time)
my_vars.reset(token)
def higher_func():
for _ in range(10):
threading.Thread(target=worker).start()
Output example:
Expected: a=5, b=2, Actual: a=5, b=2, waited 0s
Expected: a=5, b=5, Actual: a=5, b=5, waited 0s
Expected: a=6, b=7, Actual: a=6, b=7, waited 1s
Expected: a=7, b=3, Actual: a=7, b=3, waited 4s
Expected: a=1, b=1, Actual: a=1, b=1, waited 5s
Expected: a=5, b=2, Actual: a=5, b=2, waited 6s
Expected: a=4, b=8, Actual: a=4, b=8, waited 6s
Expected: a=2, b=9, Actual: a=2, b=9, waited 7s
Expected: a=6, b=0, Actual: a=6, b=0, waited 8s
Expected: a=3, b=6, Actual: a=3, b=6, waited 9s
Upvotes: 1