Prompt
When implementing features that may run concurrently, always ensure thread safety using appropriate synchronization mechanisms:
- Use thread-safe context variables for isolating execution contexts:
# Good: Using contextvars-based solutions for thread isolation ctx = baggage.set_baggage(...) # Uses contextvars internally for thread safety - Protect shared mutable state with locks:
# Good: Adding locks to prevent race conditions _lock: Lock = threading.Lock() def singleton_method(self): with self._lock: # Access or modify shared state safely - Implement thread-based timeouts efficiently:
# Better: Simple thread-based timeout pattern thread = threading.Thread(target=target, daemon=True) thread.start() thread.join(timeout=timeout) if thread.is_alive(): # Handle timeout case
These patterns help prevent hard-to-debug race conditions, ensure consistent behavior in concurrent environments, and make code more maintainable by clearly indicating thread safety considerations.