Skip to content

Instantly share code, notes, and snippets.

@yaodong
Last active February 20, 2022 17:12
Show Gist options
  • Select an option

  • Save yaodong/8b194e744c6cbdac86f7e06284df6fc1 to your computer and use it in GitHub Desktop.

Select an option

Save yaodong/8b194e744c6cbdac86f7e06284df6fc1 to your computer and use it in GitHub Desktop.

Revisions

  1. yaodong renamed this gist Feb 20, 2022. 1 changed file with 0 additions and 0 deletions.
    File renamed without changes.
  2. yaodong created this gist Feb 20, 2022.
    43 changes: 43 additions & 0 deletions gistfile1.txt
    Original file line number Diff line number Diff line change
    @@ -0,0 +1,43 @@
    from functools import partial

    from flask import Flask
    from flask import current_app
    from werkzeug.local import LocalProxy
    from werkzeug.local import LocalStack

    from .boto import Boto3Service
    from .cognito import CognitoService

    _thread_ctx = LocalStack()


    def _lookup_thread_ctx_stack(cls, *deps):
    ctx = _thread_ctx.top

    if ctx is None:
    _thread_ctx.push(ctx := {})

    if cls not in ctx:
    ctx[cls.__name__] = cls(current_app, *deps)

    return ctx[cls.__name__]


    def thread_context(cls, *deps):
    return LocalProxy(partial(_lookup_thread_ctx_stack, cls, *deps))


    def setup(app: Flask):
    @app.teardown_appcontext
    def teardown(*_):
    ctx = _thread_ctx.top
    if ctx is None:
    return

    for name, value in ctx.items():
    if hasattr(value, "__teardown__"):
    value.__teardown__()


    boto3: BotoService = thread_context(Boto3Service) # type: ignore
    cognito: CognitoContext = thread_context(CognitoService, boto3) # type: ignore