Demo executing a function within a context

context.py

import contextvars
import logging
import multiprocessing as mp
import sys
import threading
import typing as t


logging.basicConfig(stream=sys.stdout, 
                    format="%(asctime)s [%(process)d::%(thread)d] %(levelname)s %(name)s %(message)s",
                    level=logging.INFO)
logger = logging.getLogger(__name__)
ctx_var = contextvars.ContextVar("foo")


class ContextThread(threading.Thread):
    """Customized Thread that ensures new threads may change context vars"""

    def __init__(self, *args: t.Any, **kwargs: t.Any) -> None:
        self.ctx = contextvars.copy_context()
        super().__init__(*args, **kwargs)

    def run(self) -> None:
        return self.ctx.run(super().run)


def other(desc: str):
    logger.info(f"other::ctx_var: {ctx_var.get(None)} - {desc}")

def main():
    ctx_var.set(123)
    ctx = contextvars.copy_context()
    
    logger.info(f"main::ctx_var: {ctx_var.get(None)}")

    ctx.run(other, desc='ctx.run')
    
    p = mp.Process(target=other, args=('mp.Process',))
    p.start()
    p.join()

    t = threading.Thread(target=other, args=('threading.Thread',))
    t.start()
    t.join()

    t = ContextThread(target=other, args=('ContextThread',))
    t.start()
    t.join()


if __name__ == "__main__":
    main()

2024-01-09 11:18:05,748 [31867::8449057024] INFO __main__ main::ctx_var: 123
2024-01-09 11:18:05,748 [31867::8449057024] INFO __main__ other::ctx_var: 123 - ctx.run
2024-01-09 11:18:05,784 [31869::8449057024] INFO __mp_main__ other::ctx_var: None - mp.Process
2024-01-09 11:18:05,788 [31867::6178451456] INFO __main__ other::ctx_var: None - threading.Thread
2024-01-09 11:18:05,788 [31867::6178451456] INFO __main__ other::ctx_var: 123 - ContextThread