Last active
February 20, 2022 17:12
-
-
Save yaodong/8b194e744c6cbdac86f7e06284df6fc1 to your computer and use it in GitHub Desktop.
a service layer use flask local stack
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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 |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment