chacha/pyrestresource
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 21 Wiki Activity

fix dict delete + plugin

Browse Source
This commit is contained in:
cclecle
2023-11-11 12:09:03 +00:00
parent 6cc6056220
commit f677f7bf28
3 changed files with 82 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files

95
src/pyrestresource/rest_resource_handler.py
View File

@@ -312,9 +312,31 @@ class ResourceHandler_dict(
_dict: dict[T_DictKey, T_DictValues] = cast(dict[T_DictKey, T_DictValues], self.resource)
plugin_dict: ResourcePlugin_dict | None = None
if self.req.get_resource_origin(1) in self.prev_handler.resource._plugins_:
plugin_dict = cast(
ResourcePlugin_dict,
self.prev_handler.resource._plugins_[self.req.get_resource_origin(1)],
)
plugin_dict.set_context(self.req, self.req.get_root_resource())
if params.API_key is not None:
del _dict[dict_key_type(params.API_key)]
if issubclass(dict_key_type, bytes):
key_byte = dict_key_type(params.API_key, "utf-8")
if plugin_dict:
plugin_dict.handle_dict_delete(_dict, key_byte, params)
return None
del _dict[key_byte]
else:
key_std = dict_key_type(params.API_key)
if plugin_dict:
plugin_dict.handle_dict_delete(_dict, key_std, params)
return None
del _dict[key_std]
else:
if plugin_dict:
plugin_dict.handle_dict_delete_all(_dict, params)
return None
_dict.clear()
return
@@ -407,35 +429,26 @@ class ResourceHandler_dict_elem(
assert isinstance(self.prev_handler.resource, RestResourceBase)
dict_key_type: T_T_DictKey = self.prev_handler.resource._dict_key_type_[self.req.get_resource_origin(1)]
_dict: dict[T_DictKey, T_DictValues] = cast(dict[T_DictKey, T_DictValues], self.resource)
plugin_dict: ResourcePlugin_dict | None = None
if self.req.get_resource_origin(1) in self.prev_handler.resource._plugins_:
plugin_dict = cast(
ResourcePlugin_dict,
self.prev_handler.resource._plugins_[self.req.get_resource_origin(1)],
)
plugin_dict.set_context(self.req, self.req.get_root_resource())
if issubclass(dict_key_type, bytes):
key_byte = dict_key_type(self.req.get_resource_origin(0), "utf-8")
dict_byte: dict[T_DictKey, T_DictValues] = cast(dict[T_DictKey, T_DictValues], self.resource)
if self.req.get_resource_origin(1) in self.prev_handler.resource._plugins_:
plugin_dict: ResourcePlugin_dict = cast(
ResourcePlugin_dict,
self.prev_handler.resource._plugins_[self.req.get_resource_origin(1)],
)
plugin_dict.set_context(self.req, self.req.get_root_resource())
return plugin_dict.handle_dict_elem_get(dict_byte, key_byte, params)
return dict_byte[key_byte]
if plugin_dict:
return plugin_dict.handle_dict_elem_get(_dict, key_byte, params)
return _dict[key_byte]
else:
key_std = dict_key_type(self.req.get_resource_origin(0))
dict_std: dict[T_DictKey, T_DictValues] = cast(dict[T_DictKey, T_DictValues], self.resource)
if self.req.get_resource_origin(1) in self.prev_handler.resource._plugins_:
plugin_dict: ResourcePlugin_dict = cast(
ResourcePlugin_dict,
self.prev_handler.resource._plugins_[self.req.get_resource_origin(1)],
)
plugin_dict.set_context(self.req, self.req.get_root_resource())
return plugin_dict.handle_dict_elem_get(dict_std, key_std, params)
return dict_std[key_std]
if plugin_dict:
return plugin_dict.handle_dict_elem_get(_dict, key_std, params)
return _dict[key_std]
def _handle_process_delete(self, params) -> None:
# print(f"{type(self).__name__}->_handle_process_delete()")
@@ -449,33 +462,25 @@ class ResourceHandler_dict_elem(
assert isinstance(self.prev_handler.resource, RestResourceBase)
dict_key_type: T_T_DictKey = self.prev_handler.resource._dict_key_type_[self.req.get_resource_origin(2)]
_dict: dict[T_DictKey, T_DictValues] = cast(dict[T_DictKey, T_DictValues], self.resource)
plugin_dict: ResourcePlugin_dict | None = None
if self.req.get_resource_origin(2) in self.prev_handler.resource._plugins_:
plugin_dict = cast(ResourcePlugin_dict, self.prev_handler.resource._plugins_[self.req.get_resource_origin(2)])
plugin_dict.set_context(self.req, self.req.get_root_resource())
if issubclass(dict_key_type, bytes):
key_byte = dict_key_type(self.req.get_resource_origin(1), "utf-8")
dict_byte: dict[T_DictKey, T_DictValues] = cast(dict[T_DictKey, T_DictValues], self.resource)
if self.req.get_resource_origin(2) in self.prev_handler.resource._plugins_:
plugin_dict: ResourcePlugin_dict = cast(
ResourcePlugin_dict, self.prev_handler.resource._plugins_[self.req.get_resource_origin(2)]
)
plugin_dict.set_context(self.req, self.req.get_root_resource())
plugin_dict.handle_dict_delete(dict_byte, key_byte, params)
if plugin_dict:
plugin_dict.handle_dict_delete(_dict, key_byte, params)
return None
del dict_byte[key_byte]
del _dict[key_byte]
else:
key_std = dict_key_type(self.req.get_resource_origin(1))
dict_std: dict[T_DictKey, T_DictValues] = cast(dict[T_DictKey, T_DictValues], self.resource)
if self.req.get_resource_origin(2) in self.prev_handler.resource._plugins_:
plugin_dict: ResourcePlugin_dict = cast(
ResourcePlugin_dict, self.prev_handler.resource._plugins_[self.req.get_resource_origin(2)]
)
plugin_dict.set_context(self.req, self.req.get_root_resource())
plugin_dict.handle_dict_delete(dict_std, key_std, params)
if plugin_dict:
plugin_dict.handle_dict_delete(_dict, key_std, params)
return None
del dict_std[key_std]
del _dict[key_std]
@ResourceHandler.register_resource_handler

1
src/pyrestresource/rest_resource_rootpoint.py
View File

@@ -6,7 +6,6 @@ from typing import (
TYPE_CHECKING,
)
from pydantic import BaseModel
from pydantic.fields import FieldInfo
from .rest_resource import RestResourceBase

48
test/test_rest_resource_plugins_dict.py
View File

@@ -38,9 +38,6 @@ def init_classes():
resource_dict: dict[str, Test_Record],
params: RestRequestParams_Dict_GET,
) -> list[str]:
# print("HOOK handle_dict_get_keys")
# print(resource_dict)
# print(params)
result = super().handle_dict_get_keys(resource_dict, params)
if self.static_Test_Record_active is True:
result.append("static_elem")
@@ -52,10 +49,6 @@ def init_classes():
key: str,
params: RestRequestParams_Dict_elem_GET,
) -> Test_Record:
# print("HOOK handle_dict_elem_get")
# print(resource_dict)
# print(key)
# print(params)
if key == "static_elem":
if self.static_Test_Record_active is True:
return self.static_Test_Record
@@ -69,16 +62,19 @@ def init_classes():
key: str,
params: RestRequestParams_Dict_DELETE[str],
) -> None:
print("HOOK handle_dict_delete")
print(resource_dict)
print(key)
print(params)
print(self.static_Test_Record_active)
if key == "static_elem":
self.static_Test_Record_active = False
else:
del resource_dict[key]
def handle_dict_delete_all(
self,
resource_dict: dict[str, Test_Record],
params: RestRequestParams_Dict_DELETE[str],
) -> None:
self.static_Test_Record_active = False
resource_dict.clear()
@register_rest_rootpoint
class RootApp(RestResourceBase):
str_dict_Test_Record: dict[str, Test_Record] = RestField(
@@ -116,14 +112,34 @@ class Test_RestAPI_Plugin_Dict(unittest.TestCase):
self.assertEqual(result.get_result(), '{"test_str": "mytest", "test_int": 84}')
def test_delete_dict_elems(self):
result = self.testapp.process_request("/str_dict_Test_Record/test", rsrc_verb.DELETE)
print(result.get_result())
self.testapp.process_request("/str_dict_Test_Record/test", rsrc_verb.DELETE)
result = self.testapp.process_request("/str_dict_Test_Record", rsrc_verb.GET)
self.assertEqual(result.get_result(), '["static_elem"]')
result = self.testapp.process_request("/str_dict_Test_Record/static_elem", rsrc_verb.DELETE)
print(result.get_result())
self.testapp.process_request("/str_dict_Test_Record/static_elem", rsrc_verb.DELETE)
result = self.testapp.process_request("/str_dict_Test_Record", rsrc_verb.GET)
self.assertEqual(result.get_result(), "[]")
def test_delete_all_dict_elems(self):
self.testapp.process_request("/str_dict_Test_Record/test", rsrc_verb.DELETE)
result = self.testapp.process_request("/str_dict_Test_Record", rsrc_verb.GET)
self.assertEqual(result.get_result(), '["static_elem"]')
self.testapp.process_request("/str_dict_Test_Record", rsrc_verb.DELETE)
result = self.testapp.process_request("/str_dict_Test_Record", rsrc_verb.GET)
self.assertEqual(result.get_result(), "[]")
def test_delete_dict_elems_API_key(self):
self.testapp.process_request("/str_dict_Test_Record/test", rsrc_verb.DELETE)
result = self.testapp.process_request("/str_dict_Test_Record", rsrc_verb.GET)
self.assertEqual(result.get_result(), '["static_elem"]')
self.testapp.process_request("/str_dict_Test_Record?API_key=static_elem", rsrc_verb.DELETE)
result = self.testapp.process_request("/str_dict_Test_Record", rsrc_verb.GET)
self.assertEqual(result.get_result(), "[]")