Add reset method for init and reset bp instance; update copy method

This commit is contained in:
Zhiwei Liang 2021-07-20 20:07:37 -05:00
parent d4cd897522
commit 80029d636d
2 changed files with 57 additions and 9 deletions

View File

@ -79,14 +79,9 @@ class Blueprint(BaseSanic):
): ):
super().__init__(name=name) super().__init__(name=name)
self._apps: Set[Sanic] = set()
self.ctx = SimpleNamespace() self.ctx = SimpleNamespace()
self.exceptions: List[RouteHandler] = []
self.host = host self.host = host
self.listeners: Dict[str, List[ListenerType]] = {} self.reset()
self.middlewares: List[MiddlewareType] = []
self.routes: List[Route] = []
self.statics: List[RouteHandler] = []
self.strict_slashes = strict_slashes self.strict_slashes = strict_slashes
self.url_prefix = ( self.url_prefix = (
url_prefix[:-1] url_prefix[:-1]
@ -95,7 +90,6 @@ class Blueprint(BaseSanic):
) )
self.version = version self.version = version
self.version_prefix = version_prefix self.version_prefix = version_prefix
self.websocket_routes: List[Route] = []
def __repr__(self) -> str: def __repr__(self) -> str:
args = ", ".join( args = ", ".join(
@ -153,6 +147,8 @@ class Blueprint(BaseSanic):
version: Optional[Union[int, str, float, Default]] = _default, version: Optional[Union[int, str, float, Default]] = _default,
version_prefix: Union[str, Default] = _default, version_prefix: Union[str, Default] = _default,
strict_slashes: Optional[Union[bool, Default]] = _default, strict_slashes: Optional[Union[bool, Default]] = _default,
with_registration: bool = True,
with_ctx: bool = False,
): ):
""" """
Copy a blueprint instance with some optional parameters to Copy a blueprint instance with some optional parameters to
@ -164,7 +160,22 @@ class Blueprint(BaseSanic):
:param version_prefix: the prefix of the version number shown in the URL. :param version_prefix: the prefix of the version number shown in the URL.
:param strict_slashes: Enforce the API urls are requested with a :param strict_slashes: Enforce the API urls are requested with a
trailing */* trailing */*
:param with_registration: whether register new blueprint instance with
sanic apps that were registered with the old instance or not.
:param with_ctx: whether `ctx` will be copied or not.
""" """
attrs_backup = {
"_apps": self._apps,
"routes": self.routes,
"websocket_routes": self.websocket_routes,
"middlewares": self.middlewares,
"exceptions": self.exceptions,
"listeners": self.listeners,
"statics": self.statics,
}
self.reset()
new_bp = deepcopy(self) new_bp = deepcopy(self)
new_bp.name = name new_bp.name = name
@ -177,6 +188,21 @@ class Blueprint(BaseSanic):
if not isinstance(version_prefix, Default): if not isinstance(version_prefix, Default):
new_bp.version_prefix = version_prefix new_bp.version_prefix = version_prefix
for key, value in attrs_backup.items():
setattr(self, key, value)
if with_registration:
if len(new_bp._future_statics) > 0 and len(self._apps) > 0:
raise Exception(
"Static routes registered with the old blueprint instance, "
"cannot be registered again."
)
for app in self._apps:
app.blueprint(new_bp)
if not with_ctx:
new_bp.ctx = SimpleNamespace()
return new_bp return new_bp
@staticmethod @staticmethod
@ -329,6 +355,15 @@ class Blueprint(BaseSanic):
*[app.dispatch(*args, **kwargs) for app in self.apps] *[app.dispatch(*args, **kwargs) for app in self.apps]
) )
def reset(self):
self._apps: Set[Sanic] = set()
self.exceptions: List[RouteHandler] = []
self.listeners: Dict[str, List[ListenerType]] = {}
self.middlewares: List[MiddlewareType] = []
self.routes: List[Route] = []
self.statics: List[RouteHandler] = []
self.websocket_routes: List[Route] = []
def event(self, event: str, timeout: Optional[Union[int, float]] = None): def event(self, event: str, timeout: Optional[Union[int, float]] = None):
events = set() events = set()
for app in self.apps: for app in self.apps:

View File

@ -5,20 +5,33 @@ from sanic.response import text
def test_bp_copy(app: Sanic): def test_bp_copy(app: Sanic):
bp1 = Blueprint("test1", version=1) bp1 = Blueprint("test_bp1", version=1)
@bp1.route("/page") @bp1.route("/page")
def handle_request(request): def handle_request(request):
return text("Hello world!") return text("Hello world!")
bp2 = bp1.copy(name="test2", version=2) bp2 = bp1.copy(name="test_bp2", version=2)
assert id(bp1) != id(bp2) assert id(bp1) != id(bp2)
app.blueprint(bp1) app.blueprint(bp1)
app.blueprint(bp2) app.blueprint(bp2)
bp3 = bp1.copy(name="test_bp3", version=3, with_registration=True)
bp4 = bp1.copy(name="test_bp4", version=4, with_ctx=True)
bp5 = bp1.copy(name="test_bp5", version=5, with_registration=False)
app.blueprint(bp5)
_, response = app.test_client.get("/v1/page") _, response = app.test_client.get("/v1/page")
assert "Hello world!" in response.text assert "Hello world!" in response.text
_, response = app.test_client.get("/v2/page") _, response = app.test_client.get("/v2/page")
assert "Hello world!" in response.text assert "Hello world!" in response.text
_, response = app.test_client.get("/v3/page")
assert "Hello world!" in response.text
_, response = app.test_client.get("/v4/page")
assert "Hello world!" in response.text
_, response = app.test_client.get("/v5/page")
assert "Hello world!" in response.text