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

first commit

Browse Source
This commit is contained in:
cclecle
2023-10-31 23:58:51 +00:00
parent 7e2f4c34bd
commit 62fedab552
26 changed files with 3517 additions and 148 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
.project
View File

@@ -1,6 +1,6 @@
<?xml version="1.0" encoding="UTF-8"?>
<projectDescription>
<name>{{project_name}}</name>
<name>pyrestresource</name>
<comment></comment>
<projects>
</projects>

1
.settings/org.eclipse.core.resources.prefs
View File

@@ -1,2 +1,3 @@
eclipse.preferences.version=1
encoding//src/pyrestresource/__init__.py=utf-8
encoding/<project>=UTF-8

53
README.md
View File

@@ -8,46 +8,25 @@
![](docs-static/Library.jpg)
# Python project template
# pyrestresource
A nice template to start blank python projets.
This template automate a lot of handy things and allow CI/CD automatic releases generation.
A RESTful API library built on top of pydantic & uvicorn to make service API from a data model.
It is also collectings data to feed Jenkins build.
/!\ early in-progress project for internal use ATM.
Checkout [Latest Documentation](https://chacha.ddns.net/mkdocs-web/chacha/{{repository}}/{{branch}}/latest/).
Feel free to contribute.
## Features
Features:
- use annotation
- support containers (dict)
- support plugins (for hook and biding)
- user authentification (WIP)
- ACL (WIP)
- python internal model instance (with possible serialization/auto-save on-disk)
- daemon mode
### Generic pipeline skeleton:
- Prepare
- GetCode
- BuildPackage
- Install
- CheckCode
- PlotMetrics
- RunUnitTests
- GenDOC
- PostRelease
### CI/CD Environment
- Jenkins
- Gitea (with patch for dynamic Readme variables: https://chacha.ddns.net/gitea/chacha/GiteaMarkupVariable)
- Docker
- MkDocsWeb
### CI/CD Helper libs
- VirtualEnv
- Changelog generation based on commits
- copier
- pylint + pylint_json2html
- mypy
- unittest + xmlrunner + junitparser + junit2htmlreport
- mkdocs
Limitations:
- no nested reads / writes
- weak unitest (atm)
### Python project
- Full .toml implementation
- .whl automatic generation
- dynamic versionning using git repository
- embedded unit-test
Checkout [Latest Documentation](https://chacha.ddns.net/mkdocs-web/chacha/pyrestresource/master/latest/).

9
pyproject.toml
View File

@@ -17,7 +17,7 @@ version_scheme= "post-release"
name = "pyrestresource"
description = "pyrestresource"
readme = "README.md"
requires-python = ">=3.9"
requires-python = ">=3.11"
keywords = ["chacha","chacha","template","pyrestresource"]
license = { file = "LICENSE.md" }
@@ -30,11 +30,12 @@ maintainers = [
classifiers = [
"Development Status :: 4 - Beta",
"Programming Language :: Python :: 3 :: Only",
"Programming Language :: Python :: 3.9",
"Programming Language :: Python :: 3.11",
]
dependencies = [
'importlib-metadata; python_version<"3.9"',
'packaging'
'packaging',
'pydantic>=2.4,<3',
'uvicorn>=0.23'
]
dynamic = ["version"]

59
src/pyrestresource/__init__.py
View File

@@ -1,3 +1,6 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# pyrestresource (c) by chacha
#
# pyrestresource is licensed under a
@@ -5,32 +8,48 @@
#
# You should have received a copy of the license along with this
# work. If not, see <https://creativecommons.org/licenses/by-nc-sa/4.0/>.
# pylint: disable=wrong-import-position
"""
Main module __init__ file.
"""
from importlib.metadata import version, distribution, PackageNotFoundError
import warnings
from typing import TYPE_CHECKING
from .test_module import test_function
from .__metadata__ import __version__, __Summuary__, __Name__
try: # pragma: no cover
__version__ = version("pyrestresource")
except PackageNotFoundError: # pragma: no cover
warnings.warn("can not read __version__, assuming local test context, setting it to ?.?.?")
__version__ = "?.?.?"
try: # pragma: no cover
dist = distribution("pyrestresource")
__Summuary__ = dist.metadata["Summary"]
except PackageNotFoundError: # pragma: no cover
warnings.warn('can not read dist.metadata["Summary"], assuming local test context, setting it to <pyrestresource description>')
__Summuary__ = "pyrestresource description"
from .rest_resource import (
register_rest_rootpoint,
RestResourceBase,
)
try: # pragma: no cover
dist = distribution("pyrestresource")
__Name__ = dist.metadata["Name"]
except PackageNotFoundError: # pragma: no cover
warnings.warn('can not read dist.metadata["Name"], assuming local test context, setting it to <pyrestresource>')
__Name__ = "pyrestresource"
from .rest_types import rsrc_verb, T_SupportedRESTFields
if TYPE_CHECKING:
from .rest_types import (
T_ListIndex,
T_ListSize,
T_DictKey,
T_T_DictKey,
T_DictValues,
T_T_DictValues,
)
from .rest_request_opt import (
RestRequestParams_POST,
RestRequestParams_DELETE,
RestRequestParams_GET,
RestRequestParams_PUT,
RestRequestParams_RestResourceBase_PUT,
RestRequestParams_RestResourceBase_GET,
RestRequestParams_Dict_POST,
RestRequestParams_Dict_DELETE,
RestRequestParams_Dict_GET,
)
from .rest_resource_plugin import (
ResourcePlugin_field_default,
ResourcePlugin_RestResourceBase_default,
ResourcePlugin_dict_default,
)

42
src/pyrestresource/__metadata__.py Normal file
View File

@@ -0,0 +1,42 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# pyrestresource(c) by chacha
#
# pyrestresource is licensed under a
# Creative Commons Attribution-NonCommercial-ShareAlike 4.0 International Unported License.
#
# You should have received a copy of the license along with this
# work. If not, see <https://creativecommons.org/licenses/by-nc-sa/4.0/>.
"""Metadata module module"""
from importlib.metadata import version, distribution, PackageNotFoundError
import warnings
try: # pragma: no cover
__version__ = version("pyrestresource")
except PackageNotFoundError: # pragma: no cover
warnings.warn(
"can not read __version__, assuming local test context, setting it to ?.?.?"
)
__version__ = "?.?.?"
try: # pragma: no cover
dist = distribution("pyrestresource")
__Summuary__ = dist.metadata["Summary"]
except PackageNotFoundError: # pragma: no cover
warnings.warn(
'can not read dist.metadata["Summary"], assuming local test context, setting it to <pyrestresource description>'
)
__Summuary__ = "pyrestresource description"
try: # pragma: no cover
dist = distribution("pyrestresource")
__Name__ = dist.metadata["Name"]
except PackageNotFoundError: # pragma: no cover
warnings.warn(
'can not read dist.metadata["Name"], assuming local test context, setting it to <pyrestresource>'
)
__Name__ = "pyrestresource"

0
src/pyrestresource/data/.keep
View File

7
src/pyrestresource/data/__init__.py
View File

@@ -1,7 +0,0 @@
# pyrestresource (c) by chacha
#
# pyrestresource is licensed under a
# Creative Commons Attribution-NonCommercial-ShareAlike 4.0 International Unported License.
#
# You should have received a copy of the license along with this
# work. If not, see <https://creativecommons.org/licenses/by-nc-sa/4.0/>.

17
src/pyrestresource/helpers.py Normal file
View File

@@ -0,0 +1,17 @@
# pylint: disable=missing-module-docstring,missing-class-docstring,missing-function-docstring
from __future__ import annotations
from uuid import UUID
import json
from .rest_types import T_Gen_DictKeys
class _JSONEncoder(json.JSONEncoder):
def default(self, o):
if isinstance(o, T_Gen_DictKeys): # pylint: disable=isinstance-second-argument-not-valid-type
return list(o)
if isinstance(o, UUID):
# if the obj is uuid, we simply return the value of uuid
return str(o)
return json.JSONEncoder.default(self, o)

212
src/pyrestresource/rest_request.py Normal file
View File

@@ -0,0 +1,212 @@
# pylint: disable=missing-module-docstring,missing-class-docstring,missing-function-docstring
"""A module to handle http requests context"""
from __future__ import annotations
from typing import (
Optional,
Generic,
)
from re import sub
from urllib.parse import urlparse, parse_qs
from pydantic import BaseModel, Field
from .rest_types import rsrc_verb, T_SupportedRESTFields
from .rest_request_opt import (
RestRequestParams_POST,
RestRequestParams_DELETE,
RestRequestParams_GET,
RestRequestParams_PUT,
_T_RestRequestParams,
_T_RestRequestParams_POST,
_T_RestRequestParams_DELETE,
_T_RestRequestParams_GET,
_T_RestRequestParams_PUT,
)
class RequestFactory(
Generic[
_T_RestRequestParams_POST,
_T_RestRequestParams_DELETE,
_T_RestRequestParams_GET,
_T_RestRequestParams_PUT,
],
BaseModel,
):
"""RestRequets class factory"""
cls_RestRequestParams_GET: type[RestRequestParams_GET] = Field(default=RestRequestParams_GET)
cls_RestRequestParams_PUT: type[RestRequestParams_PUT] = Field(default=RestRequestParams_PUT)
cls_RestRequestParams_POST: type[RestRequestParams_POST] = Field(default=RestRequestParams_POST)
cls_RestRequestParams_DELETE: type[RestRequestParams_DELETE] = Field(default=RestRequestParams_DELETE)
def get_RestRequest(self, url: str, verb: rsrc_verb, data: dict, query_string: Optional[str] = None) -> RestRequest:
"""get a RestRequets instance based on LUT_verb configuration
Args:
url: http url of the request
verb: http verb received
data: data associated with the request
"""
# /!\ mypy seems not being able to propagate typevar to composed classes
if verb is rsrc_verb.GET:
return RestRequest[RestRequestParams_GET](self.cls_RestRequestParams_GET, url, verb, data, query_string)
if verb is rsrc_verb.PUT:
return RestRequest[RestRequestParams_PUT](self.cls_RestRequestParams_PUT, url, verb, data, query_string)
if verb is rsrc_verb.POST:
return RestRequest[RestRequestParams_POST](self.cls_RestRequestParams_POST, url, verb, data, query_string)
if verb is rsrc_verb.DELETE:
return RestRequest[RestRequestParams_DELETE](self.cls_RestRequestParams_DELETE, url, verb, data, query_string)
raise RuntimeError("Invalid Verb")
def update_RestRequest(self, origin_request: RestRequest) -> RestRequest:
"""create an updated copy of a RestRequest object based on a different LUT_verb configuration
Args:
origin_request: the original request
"""
# /!\ mypy seems not being able to propagate typevar to composed classes
if origin_request.verb is rsrc_verb.GET:
return RestRequest[RestRequestParams_GET](self.cls_RestRequestParams_GET, None, None, None, None, origin_request)
if origin_request.verb is rsrc_verb.PUT:
return RestRequest[RestRequestParams_PUT](self.cls_RestRequestParams_PUT, None, None, None, None, origin_request)
if origin_request.verb is rsrc_verb.POST:
return RestRequest[RestRequestParams_POST](self.cls_RestRequestParams_POST, None, None, None, None, origin_request)
if origin_request.verb is rsrc_verb.DELETE:
return RestRequest[RestRequestParams_DELETE](self.cls_RestRequestParams_DELETE, None, None, None, None, origin_request)
raise RuntimeError("Invalid Verb")
class RestRequest(Generic[_T_RestRequestParams]):
# pylint: disable=too-many-instance-attributes
"""Main RestRequets class"""
def __init__(
self,
type_request_params: type[_T_RestRequestParams],
url: Optional[str] = None,
verb: Optional[rsrc_verb] = None,
data: Optional[dict[str, T_SupportedRESTFields]] = None,
query_string: Optional[str] = None,
origin_request: Optional[RestRequest] = None,
) -> None:
"""class to handle a request context, that will be kept and updated while walking url parts
Args:
url: http url of the request
verb: http verb received
data: data associated with the request
type_request_params: type of the request param
origin_request: orginial request in case of updates.
In this case, all other argument - but type_request_params - are ignored and inherited from the origin_request
"""
# defining all types
self.url: str
self.verb: rsrc_verb
self.data: dict
self._saved_url_params: dict
self.ReqParams: _T_RestRequestParams = type_request_params()
self.url_stack: list[str]
self._saved_url_stack: list[str]
self.url_stack_index: int
# detecting Optional fields in type_request_params (to extract real type)
# if False: # deprecated => Generic
# if get_origin(type_request_params) is Union:
# datatype = get_args(type_request_params)
# if len(datatype) == 2:
# if datatype[0] is type(None):
# type_request_params = datatype[1]
# elif datatype[1] is type(None):
# type_request_params = datatype[0]
# else:
# raise RuntimeError("Union is only allowed to describe Optional (e.g. Union[XXX,None])")
# = updating request from a previous one =
if origin_request:
self.__dict__ = origin_request.__dict__.copy()
if type_request_params:
self.ReqParams = type_request_params(**self._saved_url_params)
# print("request updated")
return
# = or create a fresh one =
if url is None or verb is None or data is None:
raise RuntimeError("url and verb and data must be set")
self.url = url
self.verb = verb
self.data = data
# parse_qs returns list[] for all keys, the command convert list to single items so pydantic can eat them :)
if query_string:
self._saved_url_params = dict((k, v if len(v) > 1 else v[0]) for k, v in parse_qs(query_string).items())
else:
self._saved_url_params = dict((k, v if len(v) > 1 else v[0]) for k, v in parse_qs(urlparse(url).query).items())
if type_request_params:
self.ReqParams = type_request_params(**self._saved_url_params) # actual lunch
self._parse_url(url)
# keeping a backup of the original url stack
self._saved_url_stack = self.url_stack.copy()
self.url_stack_index = 0
def _parse_url(self, url: str) -> None:
# remove repeated slash ('/')
url = sub(r"\/{2,}", "/", url)
# root url need to be added manually because it is trimmed by the url split function
self.url_stack = ["/"]
self.url_stack.extend([_ for _ in urlparse(url).path.split("/") if _ != ""])
def reset_url_stack(self) -> None:
self.url_stack = self._saved_url_stack.copy()
self.url_stack_index = 0
def get_url_stack(self) -> list[str]:
"""retrieve the current url stack"""
return self.url_stack
def get_url(self) -> str:
"""retrieve the raw url"""
return self.url
def consume_url_stack(self, count: int) -> list[str]:
"""consume some url stack elements
Args:
count: number of element to consume
"""
returned_stack: list[str] = []
for _ in range(count):
returned_stack.append(self.url_stack.pop(0))
self.url_stack_index = self.url_stack_index + count
return returned_stack
def get_data(self) -> dict:
"""get the request data"""
return self.data
def get_resource_origin(self, deepness: int = 0) -> str:
"""get current or previous (consumed) resource in the url
Args:
deepness: backward amount
"""
return self._saved_url_stack[self.url_stack_index - deepness]
def get_req_params(self) -> _T_RestRequestParams:
"""get extracted req_params"""
return self.ReqParams
def get_verb(self) -> rsrc_verb:
"""get http request verb"""
return self.verb

73
src/pyrestresource/rest_request_opt.py Normal file
View File

@@ -0,0 +1,73 @@
# pylint: disable=missing-module-docstring,missing-class-docstring,missing-function-docstring
from __future__ import annotations
from typing import Optional, Generic, TypeVar
from pydantic import BaseModel, Extra
from .rest_types import (
_T_DictKey,
)
class RestRequestParams(BaseModel, extra=Extra.allow):
pass
class RestRequestParams_POST(RestRequestParams):
pass
class RestRequestParams_DELETE(RestRequestParams):
pass
class RestRequestParams_GET(RestRequestParams):
pass
class RestRequestParams_PUT(RestRequestParams):
pass
class RestRequestParams_RestResourceBase(RestRequestParams):
pass
class RestRequestParams_RestResourceBase_PUT(RestRequestParams_PUT, RestRequestParams_RestResourceBase):
pass
class RestRequestParams_RestResourceBase_GET(RestRequestParams_GET, RestRequestParams_RestResourceBase):
pass
class RestRequestParams_Dict(RestRequestParams):
pass
class RestRequestParams_Dict_POST(RestRequestParams_Dict, RestRequestParams_POST, Generic[_T_DictKey]):
API_key: Optional[_T_DictKey] = None
class RestRequestParams_Dict_DELETE(RestRequestParams_Dict, RestRequestParams_DELETE, Generic[_T_DictKey]):
API_key: Optional[_T_DictKey] = None
class RestRequestParams_Dict_GET(RestRequestParams_Dict, RestRequestParams_GET):
pass
class RestRequestParams_Dict_elem_GET(RestRequestParams_Dict, RestRequestParams_GET):
pass
class RestRequestParams_Dict_elem_PUT(RestRequestParams_Dict, RestRequestParams_GET):
pass
_T_RestRequestParams = TypeVar("_T_RestRequestParams", bound=RestRequestParams)
_T_RestRequestParams_POST = TypeVar("_T_RestRequestParams_POST", bound=RestRequestParams_POST)
_T_RestRequestParams_DELETE = TypeVar("_T_RestRequestParams_DELETE", bound=RestRequestParams_DELETE)
_T_RestRequestParams_GET = TypeVar("_T_RestRequestParams_GET", bound=RestRequestParams_GET)
_T_RestRequestParams_PUT = TypeVar("_T_RestRequestParams_PUT", bound=RestRequestParams_PUT)

281
src/pyrestresource/rest_resource.py Normal file
View File

@@ -0,0 +1,281 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# pyrestresource(c) by chacha
#
# pyrestresource is licensed under a
# Creative Commons Attribution-NonCommercial-ShareAlike 4.0 International Unported License.
#
# You should have received a copy of the license along with this
# work. If not, see <https://creativecommons.org/licenses/by-nc-sa/4.0/>.
# pylint: disable=missing-module-docstring,missing-class-docstring,missing-function-docstring
"""CLI interface module"""
from __future__ import annotations
from abc import ABC
from typing import (
cast,
ClassVar,
get_args,
get_origin,
Optional,
TYPE_CHECKING,
)
import json
from pydantic.fields import FieldInfo
from pydantic import BaseModel
from .helpers import _JSONEncoder
from .rest_types import rsrc_verb, _T_SupportedRESTFields
from .rest_resource_plugin import (
ResourcePlugin_field,
ResourcePlugin_RestResourceBase,
ResourcePlugin_dict,
)
from .rest_resource_walker import (
RestResourceWalkerFutureResult,
RestResourceWalker_Root,
RestResourceWalker_Sub_T_Dict,
RestResourceWalker_Sub_RestFields,
RestResourceWalker_Sub_RestResourceBase,
)
if TYPE_CHECKING:
from .rest_types import (
T_ListIndex,
T_ListSize,
T_DictKey,
T_T_DictKey,
T_DictValues,
T_T_DictValues,
T_SupportedRESTFields,
)
class RestResourceWalkerFutureResult_RestResourceBase_tree_exclude(RestResourceWalkerFutureResult[dict]):
def process_future(self, result: Optional[list[dict]]) -> Optional[dict]:
res = {}
res[self.source.resource_name] = dict()
for subres in result:
key = next(iter(subres))
if (
key in self.source.annotation._model_dump_excluded_ # pylint: disable=protected-access
and self.source.annotation._model_dump_excluded_[key] is True # pylint: disable=protected-access
):
res[self.source.resource_name] = res[self.source.resource_name] | {key: True}
else:
res[self.source.resource_name] = res[self.source.resource_name] | subres
return res
class RestResourceWalkerFutureResult_Dict_tree_exclude(RestResourceWalkerFutureResult[dict]):
def process_future(self, result: Optional[list[dict]]) -> Optional[dict]:
res = {}
for subres in result:
res = res | subres
return res
class RestResourceWalker_Sub_T_Dict__tree_exclude(RestResourceWalker_Sub_T_Dict):
cls_RestResourceWalkerFutureResult = RestResourceWalkerFutureResult_Dict_tree_exclude
class RestResourceWalker_Sub_RestResourceBase__tree_exclude(RestResourceWalker_Sub_RestResourceBase):
cls_RestResourceWalkerFutureResult = RestResourceWalkerFutureResult_RestResourceBase_tree_exclude
class RestResourceWalker_Root__tree_exclude(RestResourceWalker_Root):
cls_RestResourceWalker_Sub = [
RestResourceWalker_Sub_T_Dict__tree_exclude,
RestResourceWalker_Sub_RestFields,
RestResourceWalker_Sub_RestResourceBase__tree_exclude,
]
class RestResourceWalker_Sub_T_Dict__tree_init(RestResourceWalker_Sub_T_Dict):
def process(self) -> None:
datatype = get_args(self.annotation)
# checking compatibility
if not get_origin(datatype[1]) is None:
raise RuntimeError("complex dict types are not supported (should create a RestResourceBase container)")
if not datatype[0] in _T_SupportedRESTFields:
raise RuntimeError(f"Unsupported Dict Field value type in class (key)")
# preprocessing types / structure
if self.parent is not None and isinstance(self.parent, RestResourceWalker_Sub_RestResourceBase):
self.parent.annotation._dict_key_type_[self.resource_name] = datatype[0] # pylint: disable=protected-access
self.parent.annotation._dict_value_type_[self.resource_name] = datatype[1] # pylint: disable=protected-access
self.parent.annotation._model_dump_excluded_[self.resource_name] = True # pylint: disable=protected-access
if (
isinstance(self.resource, FieldInfo)
and self.resource.json_schema_extra is not None
and type(self.resource.json_schema_extra) is dict
and "plugin" in self.resource.json_schema_extra
):
plugin_dict: ResourcePlugin_dict = self.resource.json_schema_extra["plugin"]()
if not isinstance(plugin_dict, ResourcePlugin_dict):
raise RuntimeError("Wrong plugin signature provided")
self.parent.annotation._plugins_[self.resource_name] = plugin_dict
# print("ADD DICT PLUGIN")
else:
raise RuntimeError("dict must be contained in a RestResourceBase")
class RestResourceWalker_Sub_RestFields__tree_init(RestResourceWalker_Sub_RestFields):
def process(self) -> None:
if self.parent is not None and isinstance(self.parent, RestResourceWalker_Sub_RestResourceBase):
if (
isinstance(self.resource, FieldInfo)
and self.resource.json_schema_extra is not None
and type(self.resource.json_schema_extra) is dict
):
if "primary_key" in self.resource.json_schema_extra and self.resource.json_schema_extra["primary_key"] is True:
if self.parent.annotation._primary_key_ is not None:
raise RuntimeError(f"Only one primary key is allowed {self.parent.resource_name}.{self.resource_name}")
self.parent.annotation._primary_key_ = self.resource_name
if "plugin" in self.resource.json_schema_extra and self.resource.json_schema_extra["plugin"]:
plugin_field: ResourcePlugin_field = self.resource.json_schema_extra["plugin"]()
if not isinstance(plugin_field, ResourcePlugin_field):
raise RuntimeError("Wrong plugin signature provided")
self.parent.annotation._plugins_[self.resource_name] = plugin_field
# print("ADD FIELD PLUGIN")
else:
raise RuntimeError("fields must be contained in a RestResourceBase")
class RestResourceWalker_Sub_RestResourceBase__tree_init(RestResourceWalker_Sub_RestResourceBase):
def process(self) -> None:
setattr(self.annotation, "_dict_key_type_", {})
setattr(self.annotation, "_dict_value_type_", {})
setattr(self.annotation, "_model_dump_excluded_", {})
setattr(self.annotation, "_primary_key_", None)
setattr(self.annotation, "_plugins_", {})
# preprocessing types / structure
if self.parent is not None and isinstance(self.parent, RestResourceWalker_Sub_RestResourceBase):
self.parent.annotation._model_dump_excluded_[self.resource_name] = True
if (
isinstance(self.resource, FieldInfo)
and self.resource.json_schema_extra is not None
and type(self.resource.json_schema_extra) is dict
and "plugin" in self.resource.json_schema_extra
):
plugin_resource: ResourcePlugin_RestResourceBase = self.resource.json_schema_extra["plugin"]()
if not isinstance(plugin_resource, ResourcePlugin_RestResourceBase):
raise RuntimeError("Wrong plugin signature provided")
self.parent.annotation._plugins_[self.resource_name] = plugin_resource
# print("ADD RESOURCE PLUGIN")
class RestResourceWalker_Root__tree_init(RestResourceWalker_Root):
cls_RestResourceWalker_Sub = [
RestResourceWalker_Sub_T_Dict__tree_init,
RestResourceWalker_Sub_RestFields__tree_init,
RestResourceWalker_Sub_RestResourceBase__tree_init,
]
def register_rest_rootpoint(klass: type[RestResourceBase]):
RestResourceWalker_Root__tree_init(klass).process()
return klass
class RestResourceBase(ABC, BaseModel, validate_assignment=True):
_dict_key_type_: ClassVar[dict[str, T_T_DictKey]] = {}
_dict_value_type_: ClassVar[dict[str, T_T_DictValues]] = {}
_model_dump_excluded_: ClassVar[dict[str, bool]] = {}
_primary_key_: ClassVar[Optional[str]] = None
_plugins_: ClassVar[
dict[
str,
ResourcePlugin_field | ResourcePlugin_RestResourceBase | ResourcePlugin_dict,
]
] = {}
def update(self, **new_data):
for field, value in new_data.items():
setattr(self, field, value)
async def read_body(self, receive):
"""
Read and return the entire body from an incoming ASGI message.
"""
body = b""
more_body = True
while more_body:
message = await receive()
body += message.get("body", b"")
more_body = message.get("more_body", False)
return body
async def __call__(self, scope, receive, send):
assert scope["type"] == "http"
method = scope["method"]
assert method in ["GET", "DELETE", "PUT", "POST"]
if b"content-type" in scope["headers"]:
assert scope["headers"][b"content-type"] == b"application/json"
body = await self.read_body(receive)
verb = rsrc_verb[scope["method"]]
result = self.process_request(
scope["path"], rsrc_verb[scope["method"]], body.decode("utf-8"), scope["query_string"].decode("utf-8")
)
status = 200
if verb in (rsrc_verb.POST, rsrc_verb.PUT):
status = 201
await send(
{
"type": "http.response.start",
"status": status,
"headers": [
[b"content-type", b"application/json"],
],
}
)
body = None
if result:
body = result.encode("utf-8")
await send(
{
"type": "http.response.body",
"body": body,
}
)
def process_request(
self, url: str, verb: rsrc_verb = rsrc_verb.GET, data_json: Optional[str] = None, query_string: Optional[str] = None
) -> Optional[str]:
from .rest_resource_handler import (
ResourceHandler,
ResourceHandler_RestResourceBase,
)
data: dict = {}
if data_json:
data = json.loads(data_json)
ressource: ResourceHandler = ResourceHandler_RestResourceBase(self, url, verb, data, query_string)
result = ressource.process_verb()
if isinstance(result, RestResourceBase):
exclude: Optional[dict[str, bool]] = None
raw_exclude = RestResourceWalker_Root__tree_exclude(result).process()
exclude = next(iter(raw_exclude.values()))
return json.dumps(result.model_dump(mode="json", exclude=exclude))
if result is not None:
return json.dumps(result, cls=_JSONEncoder)
return None

631
src/pyrestresource/rest_resource_handler.py Normal file
View File

@@ -0,0 +1,631 @@
from __future__ import annotations
import abc
from typing import Optional, cast, TypeVar, Generic, Self, TYPE_CHECKING
from .rest_types import (
rsrc_verb,
T_SupportedRESTFields,
T_DictKey,
_T_SupportedRESTFields,
T_Dict,
T_T_DictValues,
T_DictValues,
)
from .rest_resource import RestResourceBase
from .rest_request import RequestFactory, RestRequest
from .rest_resource_plugin import (
ResourcePlugin_field,
ResourcePlugin_RestResourceBase,
)
from .rest_request_opt import (
RestRequestParams_POST,
RestRequestParams_DELETE,
RestRequestParams_GET,
RestRequestParams_PUT,
RestRequestParams_RestResourceBase_PUT,
RestRequestParams_RestResourceBase_GET,
RestRequestParams_Dict_POST,
RestRequestParams_Dict_DELETE,
RestRequestParams_Dict_GET,
_T_RestRequestParams_POST,
_T_RestRequestParams_DELETE,
_T_RestRequestParams_GET,
_T_RestRequestParams_PUT,
)
from .rest_resource_handler_walker import RestResourceWalker_Root__handler
if TYPE_CHECKING:
from .rest_types import (
T_ListIndex,
T_ListSize,
T_T_DictKey,
T_FieldValue,
)
_T_Resource = TypeVar("_T_Resource", T_DictValues, T_Dict, T_SupportedRESTFields, RestResourceBase)
class ResourceHandler(
abc.ABC,
Generic[
_T_Resource,
_T_RestRequestParams_POST,
_T_RestRequestParams_DELETE,
_T_RestRequestParams_GET,
_T_RestRequestParams_PUT,
],
):
_ar_resource_handler_cls_: list[type[ResourceHandler]] = []
_nb_url_element_to_consume_ = 1
_request_factory: RequestFactory[
_T_RestRequestParams_POST,
_T_RestRequestParams_DELETE,
_T_RestRequestParams_GET,
_T_RestRequestParams_PUT,
] = RequestFactory[
_T_RestRequestParams_POST,
_T_RestRequestParams_DELETE,
_T_RestRequestParams_GET,
_T_RestRequestParams_PUT,
](
cls_RestRequestParams_GET=RestRequestParams_GET,
cls_RestRequestParams_PUT=RestRequestParams_PUT,
cls_RestRequestParams_POST=RestRequestParams_POST,
cls_RestRequestParams_DELETE=RestRequestParams_DELETE,
)
def __init__(
self,
resource: _T_Resource,
url: Optional[str] = None,
verb: Optional[rsrc_verb] = None,
data: Optional[dict] = None,
query_string: Optional[str] = None,
prev_handler: Optional[ResourceHandler] = None,
) -> None:
self.prev_handler: Optional[ResourceHandler] = None
self.next_handler: Optional[ResourceHandler] = None
self.saved_url: list[str] = []
self.resource: _T_Resource = resource
self.req: RestRequest
if prev_handler is not None:
self.prev_handler = prev_handler
self.req = self._request_factory.update_RestRequest(self.prev_handler.req)
elif None in [url, verb]:
raise RuntimeError("if req not set, url,verb must be setted")
else:
if url is None or verb is None:
raise RuntimeError("url and verb must be set")
if data is None:
data = {}
self.req = self._request_factory.get_RestRequest(url, verb, data, query_string)
# print(f"[TRACE] creating {type(self).__name__}() with url={self.req.get_url_stack()}")
@classmethod
def create_chained_handler(cls, other: ResourceHandler, resource: _T_Resource) -> Self:
return cls(resource, None, None, None, None, other)
@classmethod
@abc.abstractmethod
def _check_resource_handler(cls, resource: _T_Resource, req: RestRequest) -> bool:
return False
@classmethod
def _get_resource_handler(cls, resource: _T_Resource, req: RestRequest) -> type[ResourceHandler]:
for resource_handler_cls in cls._ar_resource_handler_cls_:
if resource_handler_cls._check_resource_handler(resource, req):
# print(f"[DEBUG] match ResourceHandler: {resource_handler_cls.__name__}")
return resource_handler_cls
raise RuntimeError(f"Unsupported Resource Type {type(resource).__name__}")
@classmethod
def register_resource_handler(cls, other_cls) -> None:
cls._ar_resource_handler_cls_.append(other_cls)
return other_cls
def process_verb(
self,
) -> Optional[_T_Resource | T_DictKey | list[T_DictKey]]:
# print(f"[TRACE] {type(self).__name__}->process_verb()")
self._reset_context()
resource_handler = self._find_resource()
return resource_handler._process_verb()
def access_resource(
self,
) -> _T_Resource:
# print(f"[TRACE] {type(self).__name__}->access_resource()")
self._reset_context()
resource_handler = self._find_resource()
return resource_handler.resource
def _reset_context(self) -> None:
self.req.reset_url_stack()
def _find_resource(
self,
) -> ResourceHandler[
_T_Resource,
_T_RestRequestParams_POST,
_T_RestRequestParams_DELETE,
_T_RestRequestParams_GET,
_T_RestRequestParams_PUT,
]:
# print(f"[TRACE] {type(self).__name__}->_find_resource()")
# print(f"[DEBUG] {type(self).__name__}->resource = {type(self.resource).__name__}")
if len(self.req.get_url_stack()) == 0:
return self
self._check_access_rights()
next_resource = self._process_get()
# reveal_type(next_resource)
_next_resource = cast(_T_Resource, next_resource)
# reveal_type(_next_resource)
# print(f"[DEBUG] next_resource = {type(next_resource).__name__}")
if (
isinstance(_next_resource, RestResourceBase)
or isinstance(_next_resource, dict)
or type(_next_resource) in _T_SupportedRESTFields
):
next_resource_handler_cls: type[ResourceHandler] = self._get_resource_handler(_next_resource, self.req)
self.saved_url = self.req.consume_url_stack(self._nb_url_element_to_consume_)
# we always create a new ResourceHandler context because we might want to access
# previous saved/chained ones (for exemple to put/post/delete values in containers)
# if next_resource_handler_cls != type(self):
# print(f"[DEBUG] CHANGING HANDLER to {next_resource_handler_cls.__name__}")
next_resource_handler: ResourceHandler = next_resource_handler_cls.create_chained_handler(self, _next_resource)._find_resource()
self.next_handler = next_resource_handler
return next_resource_handler
# in the context of _find_resource, only resource real values can be retrieved
raise RuntimeError("Wrong request")
def _check_access_rights(self):
pass
def _process_verb(
self,
) -> Optional[_T_Resource | T_DictKey | list[T_DictKey]]:
# print(f"[TRACE] {type(self).__name__}->_process_verb()")
verb = self.req.get_verb()
if verb is rsrc_verb.GET:
return self._process_get()
if verb is rsrc_verb.PUT:
self._process_put()
return None
if verb is rsrc_verb.POST:
return self._process_post()
if verb is rsrc_verb.DELETE:
self._process_delete()
return None
raise RuntimeError("Invalid Verb")
def _process_get(
self,
) -> _T_Resource | list[T_DictKey]:
return self._handle_process_get(self.req.get_req_params())
def _process_put(self) -> None:
self._handle_process_put(self.req.get_req_params())
def _process_post(
self,
) -> Optional[T_DictKey]:
return self._handle_process_post(self.req.get_req_params())
def _process_delete(
self,
) -> None:
self._handle_process_delete(self.req.get_req_params())
def _handle_process_get(self, params: _T_RestRequestParams_GET) -> _T_Resource | list[T_DictKey]:
raise RuntimeError(f"GET method not implemented for {type(self).__name__}")
def _handle_process_put(self, params: _T_RestRequestParams_PUT) -> None:
raise RuntimeError(f"PUT method not implemented for {type(self).__name__}")
def _handle_process_post(self, params: _T_RestRequestParams_POST) -> Optional[T_DictKey]:
raise RuntimeError(f"POST method not implemented for {type(self).__name__}")
def _handle_process_delete(self, params: _T_RestRequestParams_DELETE) -> None:
raise RuntimeError(f"DELETE method not implemented for {type(self).__name__}")
@ResourceHandler.register_resource_handler
class ResourceHandler_dict(
ResourceHandler[
T_Dict,
RestRequestParams_Dict_POST,
RestRequestParams_Dict_DELETE,
RestRequestParams_Dict_GET,
_T_RestRequestParams_PUT,
]
):
_nb_url_element_to_consume_ = 0
_request_factory: RequestFactory[
RestRequestParams_Dict_POST,
RestRequestParams_Dict_DELETE,
RestRequestParams_Dict_GET,
_T_RestRequestParams_PUT,
] = RequestFactory[
RestRequestParams_Dict_POST,
RestRequestParams_Dict_DELETE,
RestRequestParams_Dict_GET,
_T_RestRequestParams_PUT,
](
cls_RestRequestParams_GET=RestRequestParams_Dict_GET,
cls_RestRequestParams_POST=RestRequestParams_Dict_POST,
cls_RestRequestParams_DELETE=RestRequestParams_Dict_DELETE,
)
@classmethod
def _check_resource_handler(cls, resource: _T_Resource, req: RestRequest) -> bool:
# print(f"{cls.__name__}->_check_resource_handler()")
return isinstance(resource, dict) and len(req.get_url_stack()) == 1
def _handle_process_get(self, params) -> list[T_DictKey]:
# print(f"{type(self).__name__}->_process_get()")
# print(f"{type(self).__name__}->resource = {type(self.resource).__name__}")
_dict: dict[T_DictKey, T_DictValues] = cast(dict[T_DictKey, T_DictValues], self.resource)
return list(_dict.keys())
def _handle_process_delete(self, params) -> None:
# print(f"{type(self).__name__}->_handle_process_delete()")
# print(f"{type(self).__name__}->resource = {type(self.resource).__name__}")
if self.prev_handler is None:
raise RuntimeError("Wrong command")
dict_key_type: T_T_DictKey = cast(RestResourceBase, 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)
if params.API_key is not None:
del _dict[dict_key_type(params.API_key)]
else:
_dict.clear()
return
def _handle_process_post(self, params) -> Optional[T_DictKey]:
# pylint: disable=protected-access
# print(f"{type(self).__name__}->_handle_process_post()")
# print(f"{type(self).__name__}->resource = {type(self.resource).__name__}")
if self.prev_handler is None:
raise RuntimeError("Wrong command")
dict_key_type: T_T_DictKey = cast(RestResourceBase, self.prev_handler.resource)._dict_key_type_[self.req.get_resource_origin(1)]
dict_value_type: T_T_DictValues = cast(RestResourceBase, self.prev_handler.resource)._dict_value_type_[
self.req.get_resource_origin(1)
]
_obj = dict_value_type(**self.req.get_data())
_dict: dict[T_DictKey, "T_DictValues"] = cast(dict[T_DictKey, "T_DictValues"], self.resource)
# 1st try/ using request param provided dict API_key
if params.API_key is not None:
# if a primary key is set for the resource, updating it
if isinstance(_obj, RestResourceBase):
if _obj._primary_key_ is not None:
_pri: T_DictKey = dict_key_type(params.API_key)
setattr(_obj, _obj._primary_key_, _pri)
# storing resource
_dict[dict_key_type(params.API_key)] = _obj
return dict_key_type(params.API_key)
# 2nd try/ using provided resource internal primary key
# & 3rd try/ using resource internal auto-generated primary key
# => this case is automatic because if self.req.get_data() doesn't contain the key, it should be automatically created
if isinstance(_obj, RestResourceBase):
if _obj._primary_key_ is not None:
_obj_primary_key: Optional[T_DictKey] = getattr(_obj, _obj._primary_key_)
if _obj_primary_key is not None:
_dict[_obj_primary_key] = _obj
return _obj_primary_key
RuntimeError("Either the object needs defined primary key or the request must contain an API_key param to process this command")
return None # for mypy....
@ResourceHandler.register_resource_handler
class ResourceHandler_dict_elem(
ResourceHandler[
T_DictValues,
_T_RestRequestParams_POST,
_T_RestRequestParams_DELETE,
RestRequestParams_RestResourceBase_GET,
_T_RestRequestParams_PUT,
]
):
_nb_url_element_to_consume_ = 1
_request_factory: RequestFactory[
_T_RestRequestParams_POST,
_T_RestRequestParams_DELETE,
RestRequestParams_RestResourceBase_GET,
_T_RestRequestParams_PUT,
] = RequestFactory[
_T_RestRequestParams_POST,
_T_RestRequestParams_DELETE,
RestRequestParams_RestResourceBase_GET,
_T_RestRequestParams_PUT,
](
cls_RestRequestParams_GET=RestRequestParams_RestResourceBase_GET
)
@classmethod
def _check_resource_handler(cls, resource: _T_Resource, req: RestRequest) -> bool:
# print(f"{cls.__name__}->_check_resource_handler()")
return isinstance(resource, dict) and len(req.get_url_stack()) > 1
def _handle_process_get(self, params) -> T_DictValues:
# print(f"{type(self).__name__}->_process_get()")
# print(f"{type(self).__name__}->resource = {type(self.resource).__name__}")
if self.prev_handler is None:
raise RuntimeError("Wrong command")
dict_key_type: T_T_DictKey = cast(RestResourceBase, self.prev_handler.resource)._dict_key_type_[self.req.get_resource_origin(1)]
if issubclass(dict_key_type, bytes):
key_byte = dict_key_type(self.req.get_resource_origin(0), "utf-8")
return cast(dict[T_DictKey, T_DictValues], self.resource)[key_byte]
else:
key = dict_key_type(self.req.get_resource_origin(0))
return cast(dict[T_DictKey, T_DictValues], self.resource)[key]
def _handle_process_delete(self, params) -> None:
# print(f"{type(self).__name__}->_handle_process_delete()")
# print(f"{type(self).__name__}->resource = {type(self.resource).__name__}")
# this is a litle bit tricky because this call comes from a next resource, so get_resource_origin(2)
# instead of expected get_resource_origin(1) because we need to go backward
# because self.req is another context that is not saved to improve performances
if self.prev_handler is None:
raise RuntimeError("Wrong command")
dict_key_type: T_T_DictKey = cast(RestResourceBase, self.prev_handler.resource)._dict_key_type_[self.req.get_resource_origin(2)]
if issubclass(dict_key_type, bytes):
key_byte = dict_key_type(self.req.get_resource_origin(1), "utf-8")
del cast(dict[T_DictKey, T_DictValues], self.resource)[key_byte]
else:
key = dict_key_type(self.req.get_resource_origin(1))
del cast(dict[T_DictKey, T_DictValues], self.resource)[key]
@ResourceHandler.register_resource_handler
class ResourceHandler_RestResourceBase(
ResourceHandler[
RestResourceBase,
_T_RestRequestParams_POST,
_T_RestRequestParams_DELETE,
RestRequestParams_RestResourceBase_GET,
RestRequestParams_RestResourceBase_PUT,
]
):
_request_factory: RequestFactory[
_T_RestRequestParams_POST,
_T_RestRequestParams_DELETE,
RestRequestParams_RestResourceBase_GET,
RestRequestParams_RestResourceBase_PUT,
] = RequestFactory[
_T_RestRequestParams_POST,
_T_RestRequestParams_DELETE,
RestRequestParams_RestResourceBase_GET,
RestRequestParams_RestResourceBase_PUT,
](
cls_RestRequestParams_GET=RestRequestParams_RestResourceBase_GET,
cls_RestRequestParams_PUT=RestRequestParams_RestResourceBase_PUT,
)
@classmethod
def _check_resource_handler(cls, resource: _T_Resource, req: RestRequest) -> bool:
# print(f"{cls.__name__}->_check_resource_handler()")
return isinstance(resource, RestResourceBase)
def _check_access_rights(self) -> None:
super()._check_access_rights()
# print(f"{type(self).__name__}->_check_access_rights()")
if self.req.get_resource_origin(0) == "/":
return
if (
self.req.get_resource_origin(0) not in self.resource.model_fields
or self.resource.model_fields[self.req.get_resource_origin(0)].exclude is True
):
raise RuntimeError(f"Unknown or not allowed field access detected: {self.req.get_url_stack()}")
def _handle_process_get(self, params) -> RestResourceBase:
# print(f"{type(self).__name__}->_process_get()")
# print(f"{type(self).__name__}->resource = {type(self.resource).__name__}")
# CASE 1: no more item in url_stack => we reached the endpoint
# So we are in a RestResourceBase instance and must return the content
if len(self.req.get_url_stack()) == 0:
for key, attr in self.resource.model_fields.items():
if key in self.resource._plugins_:
if isinstance(self.resource._plugins_[key], ResourcePlugin_field):
plugin_field: ResourcePlugin_field = cast(ResourcePlugin_field, self.resource._plugins_[key])
value = getattr(self.resource, key)
setattr(self.resource, key, plugin_field.handle_field_get(value, params))
elif isinstance(self.resource._plugins_[key], ResourcePlugin_RestResourceBase):
plugin_field: ResourcePlugin_field = cast(ResourcePlugin_RestResourceBase, self.resource._plugins_[key])
value = getattr(self.resource, key)
setattr(self.resource, key, plugin_field.handle_resource_get(value, params))
# result = RestResourceWalker_Root__handler(self.resource).process()
# print(result)
return self.resource
# CASE 2: specific case for root Node
# TODO: this must probably be merged with the previous bloc
if self.req.get_resource_origin(0) == "/":
return self.resource
# CASE 3: in between
value = getattr(self.resource, self.req.get_resource_origin(0))
key = self.req.get_resource_origin(0)
if key in self.resource._plugins_:
if isinstance(self.resource._plugins_[key], ResourcePlugin_field):
plugin_rsrc: ResourcePlugin_RestResourceBase = cast(
ResourcePlugin_RestResourceBase,
self.resource._plugins_[key],
)
value = plugin_rsrc.handle_field_get(value, params)
elif isinstance(self.resource._plugins_[key], ResourcePlugin_RestResourceBase):
plugin_rsrc: ResourcePlugin_RestResourceBase = cast(
ResourcePlugin_RestResourceBase,
self.resource._plugins_[key],
)
value = plugin_rsrc.handle_resource_get(value, params)
return value
def _handle_process_put(self, params) -> None:
# print(f"{type(self).__name__}->_process_put()")
# print(f"{type(self).__name__}->resource = {type(self.resource).__name__}")
# creating a copy of the current resource
_new_resrc = self.resource.copy()
# updating values based on nex data
_new_resrc.update(**self.req.get_data())
# applying plugins (to nested element)
if isinstance(_new_resrc, RestResourceBase):
for key, attr in _new_resrc.model_fields.items():
if key in _new_resrc._plugins_:
if isinstance(_new_resrc._plugins_[key], ResourcePlugin_field):
plugin_field: ResourcePlugin_field = cast(ResourcePlugin_field, _new_resrc._plugins_[key])
value = getattr(_new_resrc, key)
setattr(_new_resrc, key, plugin_field.handle_field_put(value, params))
# applying plugins (from parent element)
if self.prev_handler is not None:
# element is within a dict
if (
isinstance(self.prev_handler.resource, dict)
and self.prev_handler.prev_handler is not None
and isinstance(self.prev_handler.prev_handler.resource, RestResourceBase)
):
key = self.req.get_resource_origin(2)
if key in self.prev_handler.prev_handler.resource._plugins_:
plugin_rsrc: ResourcePlugin_RestResourceBase = cast(
ResourcePlugin_RestResourceBase,
self.prev_handler.prev_handler.resource._plugins_[key],
)
_new_resrc = plugin_rsrc.handle_dict_elem_put(_new_resrc, params)
# element is within a RestResourceBase
elif isinstance(self.prev_handler.resource, RestResourceBase):
key = self.req.get_resource_origin(1)
if key in self.prev_handler.resource._plugins_:
plugin_rsrc: ResourcePlugin_RestResourceBase = cast(
ResourcePlugin_RestResourceBase,
self.prev_handler.resource._plugins_[key],
)
_new_resrc = plugin_rsrc.handle_resource_put(_new_resrc, params)
self.resource.update(**_new_resrc.dict())
return
def _handle_process_delete(self, params) -> None:
# print(f"{type(self).__name__}->_handle_process_delete()")
# print(f"{type(self).__name__}->resource = {type(self.resource).__name__}")
# DELETING an element can only be done from a dict => checking and forwarding
if (
self.prev_handler is not None
and isinstance(self.prev_handler.resource, dict)
and self.prev_handler.prev_handler is not None
and isinstance(self.prev_handler.prev_handler.resource, RestResourceBase)
):
self.prev_handler._process_delete()
else:
raise RuntimeError("cannot delete an element outside a dict")
@ResourceHandler.register_resource_handler
class ResourceHandler_simple(
ResourceHandler[
T_SupportedRESTFields,
_T_RestRequestParams_POST,
_T_RestRequestParams_DELETE,
_T_RestRequestParams_GET,
_T_RestRequestParams_PUT,
]
):
@classmethod
def _check_resource_handler(cls, resource: _T_Resource, req: RestRequest) -> bool:
# print(f"{cls.__name__}->_check_resource_handler()")
return type(resource) in _T_SupportedRESTFields
def _handle_process_get(self, params) -> T_SupportedRESTFields:
# print(f"{type(self).__name__}->_process_get()")
# print(f"{type(self).__name__}->resource = {type(self.resource).__name__}")
assert self.prev_handler is not None
assert isinstance(self.prev_handler.resource, RestResourceBase)
if self.req.get_resource_origin(1) in self.prev_handler.resource._plugins_:
plugin_simple: ResourcePlugin_field = cast(
ResourcePlugin_field,
self.prev_handler.resource._plugins_[self.req.get_resource_origin(1)],
)
return plugin_simple.handle_field_get(self.resource, params)
return self.resource
def _handle_process_put(self, params) -> None:
# print(f"{type(self).__name__}->_process_put()")
# print(f"{type(self).__name__}->resource = {type(self.resource).__name__}")
assert self.prev_handler is not None
assert isinstance(self.prev_handler.resource, RestResourceBase)
value = self.req.get_data()
if self.req.get_resource_origin(1) in self.prev_handler.resource._plugins_:
print("PLUGIN FOUND")
plugin_simple: ResourcePlugin_field = cast(
ResourcePlugin_field,
self.prev_handler.resource._plugins_[self.req.get_resource_origin(1)],
)
print(value)
value = plugin_simple.handle_field_put(value, params)
print(value)
print(self.req.get_resource_origin(1))
setattr(
self.prev_handler.resource,
self.req.get_resource_origin(1),
value,
)
print(self.prev_handler.resource)

84
src/pyrestresource/rest_resource_handler_walker.py Normal file
View File

@@ -0,0 +1,84 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# pyrestresource(c) by chacha
#
# pyrestresource is licensed under a
# Creative Commons Attribution-NonCommercial-ShareAlike 4.0 International Unported License.
#
# You should have received a copy of the license along with this
# work. If not, see <https://creativecommons.org/licenses/by-nc-sa/4.0/>.
# pylint: disable=missing-module-docstring,missing-class-docstring,missing-function-docstring
"""CLI interface module"""
from __future__ import annotations
from typing import (
ClassVar,
get_args,
get_origin,
Optional,
TYPE_CHECKING,
)
from .rest_resource_walker import (
RestResourceWalkerFutureResult,
RestResourceWalker_Root,
RestResourceWalker_Sub_T_Dict,
RestResourceWalker_Sub_RestFields,
RestResourceWalker_Sub_RestResourceBase,
)
class RestResourceWalkerFutureResult_RestResourceBase_handler(RestResourceWalkerFutureResult[dict]):
def process_future(self, result: Optional[list[dict]]) -> Optional[dict]:
print(f"RestResourceWalkerFutureResult_RestResourceBase_handler {result}")
res = {}
res[self.source.resource_name] = dict()
for subres in result:
key = next(iter(subres))
print(key)
res[self.source.resource_name] = res[self.source.resource_name] | subres
return res
class RestResourceWalkerFutureResult_Dict_handler(RestResourceWalkerFutureResult[dict]):
def process_future(self, result: Optional[list[dict]]) -> Optional[dict]:
print(f"RestResourceWalkerFutureResult_Dict_handler {result}")
res = {}
for subres in result:
res = res | subres
return res
class RestResourceWalkerFutureResult_RestFields_handler(RestResourceWalkerFutureResult[dict]):
def process_future(self, result: Optional[list[dict]]) -> Optional[dict]:
print(f"RestResourceWalkerFutureResult_RestFields_handler {result}")
print(self.source.resource)
res = {}
res[self.source.resource_name] = dict()
for subres in result:
key = next(iter(subres))
print(key)
res[self.source.resource_name] = res[self.source.resource_name] | subres
return res
class RestResourceWalker_Sub_T_Dict__handler(RestResourceWalker_Sub_T_Dict):
cls_RestResourceWalkerFutureResult = RestResourceWalkerFutureResult_Dict_handler
class RestResourceWalker_Sub_RestResourceBase__handler(RestResourceWalker_Sub_RestResourceBase):
cls_RestResourceWalkerFutureResult = RestResourceWalkerFutureResult_RestResourceBase_handler
class RestResourceWalker_Sub_RestResourceFields__handler(RestResourceWalker_Sub_RestFields):
cls_RestResourceWalkerFutureResult = RestResourceWalkerFutureResult_RestFields_handler
class RestResourceWalker_Root__handler(RestResourceWalker_Root):
cls_RestResourceWalker_Sub = [
RestResourceWalker_Sub_T_Dict__handler,
RestResourceWalker_Sub_RestResourceFields__handler,
RestResourceWalker_Sub_RestResourceBase__handler,
]

170
src/pyrestresource/rest_resource_plugin.py Normal file
View File

@@ -0,0 +1,170 @@
from __future__ import annotations
from typing import Optional, Protocol, runtime_checkable, TYPE_CHECKING
from abc import abstractmethod
from .rest_types import (
_T_DictValues,
_T_DictKey,
TV_SupportedRESTFields,
TV_RestResourceBase,
)
if TYPE_CHECKING or True:
from .rest_request_opt import (
RestRequestParams_GET,
RestRequestParams_PUT,
RestRequestParams_RestResourceBase_PUT,
RestRequestParams_RestResourceBase_GET,
RestRequestParams_Dict_POST,
RestRequestParams_Dict_DELETE,
RestRequestParams_Dict_GET,
RestRequestParams_Dict_elem_GET,
RestRequestParams_Dict_elem_PUT,
)
@runtime_checkable
class ResourcePlugin_field(Protocol[TV_SupportedRESTFields]):
@abstractmethod
def handle_field_get(self, resource: TV_SupportedRESTFields, params: RestRequestParams_GET) -> TV_SupportedRESTFields:
...
@abstractmethod
def handle_field_put(self, resource: TV_SupportedRESTFields, params: RestRequestParams_PUT) -> TV_SupportedRESTFields:
...
class ResourcePlugin_field_default(ResourcePlugin_field[TV_SupportedRESTFields]):
"""default implementation of RestResourcePlugin_simple"""
def handle_field_get(self, resource: TV_SupportedRESTFields, params: RestRequestParams_GET) -> TV_SupportedRESTFields:
return resource
def handle_field_put(self, resource: TV_SupportedRESTFields, params: RestRequestParams_PUT) -> TV_SupportedRESTFields:
return resource
@runtime_checkable
class ResourcePlugin_RestResourceBase(Protocol[TV_RestResourceBase]):
@abstractmethod
def handle_resource_get(
self,
resource: TV_RestResourceBase,
params: RestRequestParams_RestResourceBase_GET,
) -> TV_RestResourceBase:
...
@abstractmethod
def handle_resource_put(
self,
resource: TV_RestResourceBase,
params: RestRequestParams_RestResourceBase_PUT,
) -> TV_RestResourceBase:
...
class ResourcePlugin_RestResourceBase_default(ResourcePlugin_RestResourceBase[TV_RestResourceBase]):
"""default implementation of RestResourcePlugin_RestResourceBase"""
def handle_resource_get(
self,
resource: TV_RestResourceBase,
params: RestRequestParams_RestResourceBase_GET,
) -> TV_RestResourceBase:
return resource
def handle_resource_put(
self,
resource: TV_RestResourceBase,
params: RestRequestParams_RestResourceBase_PUT,
) -> TV_RestResourceBase:
return resource
@runtime_checkable
class ResourcePlugin_dict(Protocol[_T_DictKey, _T_DictValues]):
@abstractmethod
def handle_dict_get_keys(
self,
resource_dict: dict[_T_DictKey, _T_DictValues],
params: RestRequestParams_Dict_GET,
) -> list[_T_DictKey]:
...
@abstractmethod
def handle_dict_post(
self,
resource_dict: dict[_T_DictKey, _T_DictValues],
resource: _T_DictValues,
params: RestRequestParams_Dict_POST[_T_DictKey],
) -> Optional[_T_DictKey]:
...
@abstractmethod
def handle_dict_delete(
self,
resource_dict: dict[_T_DictKey, _T_DictValues],
params: RestRequestParams_Dict_DELETE[_T_DictKey],
) -> None:
...
@abstractmethod
def handle_dict_elem_get(
self,
resource: TV_RestResourceBase,
params: RestRequestParams_Dict_elem_GET,
) -> TV_RestResourceBase:
...
@abstractmethod
def handle_dict_elem_put(
self,
resource: TV_RestResourceBase,
params: RestRequestParams_Dict_elem_PUT,
) -> TV_RestResourceBase:
...
class ResourcePlugin_dict_default(ResourcePlugin_dict[_T_DictKey, _T_DictValues]):
"""default implementation of RestResourcePlugin_dict"""
def handle_dict_get_keys(
self,
resource_dict: dict[_T_DictKey, _T_DictValues],
params: RestRequestParams_Dict_GET,
) -> list[_T_DictKey]:
return list(resource_dict.keys())
def handle_dict_post(
self,
resource_dict: dict[_T_DictKey, _T_DictValues],
resource: _T_DictValues,
params: RestRequestParams_Dict_POST[_T_DictKey],
) -> Optional[_T_DictKey]:
if params.API_key is not None:
resource_dict[params.API_key] = resource
return params.API_key
def handle_dict_delete(
self,
resource_dict: dict[_T_DictKey, _T_DictValues],
params: RestRequestParams_Dict_DELETE[_T_DictKey],
) -> None:
if params.API_key is not None:
del resource_dict[params.API_key]
def handle_dict_elem_get(
self,
resource: TV_RestResourceBase,
params: RestRequestParams_Dict_elem_GET,
) -> TV_RestResourceBase:
return resource
def handle_dict_elem_put(
self,
resource: TV_RestResourceBase,
params: RestRequestParams_Dict_elem_PUT,
) -> TV_RestResourceBase:
return resource

291
src/pyrestresource/rest_resource_walker.py Normal file
View File

@@ -0,0 +1,291 @@
from __future__ import annotations
from typing import (
cast,
Any,
Optional,
Union,
get_args,
get_origin,
TypeVar,
Generic,
TYPE_CHECKING,
)
from typing import Type
from abc import ABC, abstractmethod
from pydantic.fields import FieldInfo
from .rest_types import _T_SupportedRESTFields
if TYPE_CHECKING:
from .rest_resource import RestResourceBase
TV_RestResourceWalkerFutureResult = TypeVar("TV_RestResourceWalkerFutureResult")
class RestResourceWalkerFutureResult(ABC, Generic[TV_RestResourceWalkerFutureResult]):
def __init__(self, source: RestResourceWalker_Sub):
self.source: RestResourceWalker_Sub = source
def chain_process_future(self) -> Optional[TV_RestResourceWalkerFutureResult]:
return self.source.chain_process_future()
@abstractmethod
def process_future(self, result: Optional[list[TV_RestResourceWalkerFutureResult]]) -> Optional[TV_RestResourceWalkerFutureResult]:
pass
class RestResourceWalker_Sub(ABC, Generic[TV_RestResourceWalkerFutureResult]):
cls_RestResourceWalkerFutureResult: Optional[type[RestResourceWalkerFutureResult[TV_RestResourceWalkerFutureResult]]] = None
@classmethod
@abstractmethod
def check_type(cls, resource: FieldInfo | Type["RestResourceBase"]) -> tuple[bool, Type[Any], bool]:
"""implementation interface to Factory.
The factory will call this specialized method on each implementation to find a supported one.
"""
...
@classmethod
def get(
self,
subs: list[type[RestResourceWalker_Sub]],
resource_name: str,
resource: FieldInfo | Type["RestResourceBase"],
parent: Optional[RestResourceWalker_Sub] = None,
) -> Optional[RestResourceWalker_Sub]:
for sub in subs:
_is_valid, _anno, _optional = sub.check_type(resource)
if _is_valid is True:
return sub(resource_name, resource, parent, _anno, _optional)
raise RuntimeError(f"Incompatible Field Found: {type(resource).__name__}")
return None
def __init__(
self,
resource_name: str,
resource: FieldInfo | Type["RestResourceBase"],
parent: Optional[RestResourceWalker_Sub] = None,
annotation: Optional[type["RestResourceBase"]] = None,
optional: Optional[bool] = None,
):
self.resource_name: str = resource_name
self.resource: FieldInfo | Type["RestResourceBase"] = resource
self.parent: Optional[RestResourceWalker_Sub] = parent
self.future_results_subs: Optional[list[RestResourceWalkerFutureResult[TV_RestResourceWalkerFutureResult]]] = None
self.future_result: Optional[RestResourceWalkerFutureResult[TV_RestResourceWalkerFutureResult]] = None
if self.cls_RestResourceWalkerFutureResult is not None:
self.future_results_subs = []
self.future_result = self.cls_RestResourceWalkerFutureResult(self)
self.annotation: type["RestResourceBase"]
self.optional: bool
if annotation is None or optional is None:
self.annotation, self.optional = self.ProcessAnnotation(resource)
else:
self.annotation = annotation
self.optional = optional
if self.annotation is None:
raise RuntimeError("Only annotated types are allowed in RestResourceBase derived classes")
self.subdatatype = get_args(self.annotation)
# self.info()
def info(self) -> None:
print(f"{type(self).__name__}->info()")
print("==========================")
print(f"resource_name: {self.resource_name}")
print(f"resource: {type(self.resource).__name__}")
print(f"resource: {self.resource}")
print(f"parent: {self.parent}")
print(f"annotation: {self.annotation}")
print(f"optional: {self.optional}")
print(f"subdatatype: {self.subdatatype}")
# -> cannot do that on dicts
# if self.parent is not None:
# print(f"_model_dump_excluded_: {self.parent.annotation._model_dump_excluded_}")
if False:
print("------ STACK ------")
_rsrc = self.parent
while _rsrc is not None:
print(f"{id(_rsrc.annotation)}:{_rsrc.annotation}")
_rsrc = _rsrc.parent
print("-------------------")
@classmethod
def init_sub(cls, walker: RestResourceWalker_Root) -> None:
pass
@abstractmethod
def get_future(self) -> Optional[RestResourceWalkerFutureResult]:
return self.future_result
def chain_process_future(self) -> Optional[TV_RestResourceWalkerFutureResult]:
if self.future_results_subs is not None and self.future_result is not None:
return_future_results_subs: list[Any] = [] # TODO: use typevar
for future_result in self.future_results_subs:
return_future_results_subs.append(future_result.chain_process_future())
return self.future_result.process_future(return_future_results_subs)
return None
def collect_future_result(
self,
process_future_result: Optional[RestResourceWalkerFutureResult[TV_RestResourceWalkerFutureResult]],
) -> None:
if process_future_result is not None and self.future_results_subs is not None and self.future_result is not None:
self.future_results_subs.append(process_future_result)
# @abstractmethod
def get_sub_resources(self) -> list[tuple[str, FieldInfo]]:
return []
def process(self):
pass
@staticmethod
def ProcessAnnotation(
resource: FieldInfo | Type["RestResourceBase"],
) -> tuple[type[Any], bool]:
from .rest_resource import RestResourceBase
_anno: Type[Any]
# print("!!!!!!!!!!!!!!!!!!!!!!!")
# print(resource)
# print(type(resource))
if isinstance(resource, FieldInfo) and resource.annotation is not None:
_anno = resource.annotation
elif not isinstance(resource, FieldInfo) and issubclass(resource, RestResourceBase):
_anno = resource
else:
raise RuntimeError("Incompatible resource type")
_datatype = get_args(_anno)
_optional: bool = False
if get_origin(_anno) is Union:
if len(_datatype) == 2:
if _datatype[0] is type(None):
_anno = _datatype[1]
_optional = True
elif _datatype[1] is type(None):
_anno = _datatype[0]
_optional = True
else:
raise RuntimeError("Union is only allowed to describe Optional (e.g. Union[XXX,None])")
return _anno, _optional
class RestResourceWalker_Sub_T_Dict(RestResourceWalker_Sub):
@classmethod
def check_type(cls, resource: FieldInfo | Type["RestResourceBase"]) -> tuple[bool, Type[Any], bool]:
_anno, _optional = cls.ProcessAnnotation(resource)
_type_resource = get_origin(_anno)
return (_type_resource is dict), _anno, _optional
def get_sub_resources(self) -> list[tuple[str, FieldInfo]]:
return [(self.resource_name, self.subdatatype[1])]
def get_future(self) -> Optional[RestResourceWalkerFutureResult]:
return self.future_result
class RestResourceWalker_Sub_RestFields(RestResourceWalker_Sub):
@classmethod
def check_type(cls, resource: FieldInfo | Type["RestResourceBase"]) -> tuple[bool, Type[Any], bool]:
_anno, _optional = cls.ProcessAnnotation(resource)
return (_anno in _T_SupportedRESTFields), _anno, _optional
def get_future(self) -> Optional[RestResourceWalkerFutureResult]:
return self.future_result
class RestResourceWalker_Sub_RestResourceBase(RestResourceWalker_Sub):
@classmethod
def check_type(cls, resource: FieldInfo | Type["RestResourceBase"]) -> tuple[bool, Type[Any], bool]:
from .rest_resource import RestResourceBase
_anno, _optional = cls.ProcessAnnotation(resource)
return (
((get_origin(_anno) is None) and issubclass(_anno, RestResourceBase)),
_anno,
_optional,
)
def get_sub_resources(self) -> list[tuple[str, FieldInfo]]:
return [(cast(str, key), attr) for key, attr in self.annotation.model_fields.items()]
def get_future(self) -> Optional[RestResourceWalkerFutureResult]:
return self.future_result
class RestResourceWalker_Root:
cls_RestResourceWalker_Sub: list[Type[RestResourceWalker_Sub]] = [
RestResourceWalker_Sub_T_Dict,
RestResourceWalker_Sub_RestFields,
RestResourceWalker_Sub_RestResourceBase,
]
def __init__(self, resource: "RestResourceBase" | Type["RestResourceBase"]) -> None:
from .rest_resource import RestResourceBase
self.resource: Type["RestResourceBase"]
if isinstance(resource, RestResourceBase):
self.resource = type(resource)
else:
self.resource = resource
def process(self, deep_limit: Optional[int] = None) -> Optional[TV_RestResourceWalkerFutureResult]:
current_deep: int = 0
for cls_Sub in self.cls_RestResourceWalker_Sub:
_self = self
cls_Sub.init_sub(_self)
sub_walker_initial: Optional[RestResourceWalker_Sub] = RestResourceWalker_Sub.get(
self.cls_RestResourceWalker_Sub, "/", self.resource, None
)
if sub_walker_initial is not None:
sub_walker_initial.get_future()
resource_list: list[tuple[str, FieldInfo | Type["RestResourceBase"], RestResourceWalker_Sub]] = [
(subresource_name, subresource, sub_walker_initial)
for subresource_name, subresource in sub_walker_initial.get_sub_resources()
]
new_resource_list: list[tuple[str, FieldInfo, RestResourceWalker_Sub]]
sub_walker: Optional[RestResourceWalker_Sub]
while len(resource_list) > 0 and (deep_limit is None or current_deep < deep_limit):
new_resource_list = []
for resource_name, resource, parent_sub_walker in resource_list:
sub_walker = RestResourceWalker_Sub.get(
self.cls_RestResourceWalker_Sub,
resource_name,
resource,
parent_sub_walker,
)
if sub_walker is not None:
sub_walker.process()
process_future_result: Optional[RestResourceWalkerFutureResult] = sub_walker.get_future()
parent_sub_walker.collect_future_result(process_future_result)
new_resource_list.extend(
[
(subresource_name, subresource, sub_walker)
for subresource_name, subresource in sub_walker.get_sub_resources()
]
)
resource_list = list(new_resource_list)
current_deep = current_deep + 1
return sub_walker_initial.chain_process_future()
else:
raise RuntimeError("Invalid Rootpoint")
return None

105
src/pyrestresource/rest_types.py Normal file
View File

@@ -0,0 +1,105 @@
# pylint: disable=missing-module-docstring,missing-class-docstring,missing-function-docstring
from __future__ import annotations
from enum import Enum, auto
from typing import Union, get_origin, NewType, TypeVar, TYPE_CHECKING
from datetime import datetime
from pathlib import Path
from uuid import UUID
from ipaddress import IPv4Address, IPv4Network
if TYPE_CHECKING:
from .rest_resource import RestResourceBase
T_Gen_DictKeys: type = type({}.keys())
NoneType = type(None)
class rsrc_verb(Enum):
GET = auto()
PUT = auto()
POST = auto()
DELETE = auto()
class rsrc_type(Enum):
resource = auto()
dict = auto()
list = auto()
field = auto()
_T_SupportedRESTFields = [
UUID,
str,
int,
float,
bool,
bytes,
datetime,
Path,
IPv4Address,
IPv4Network,
]
T_SupportedRESTFields = Union[
UUID, str, int, float, bool, bytes, datetime, Path, IPv4Address, IPv4Network
]
TV_SupportedRESTFields = TypeVar(
"TV_SupportedRESTFields",
UUID,
str,
int,
float,
bool,
bytes,
datetime,
Path,
IPv4Address,
IPv4Network,
)
if get_origin(T_SupportedRESTFields) is not Union:
raise RuntimeError("wrong T_SupportedRESTFields (must be flat Union)")
TV_RestResourceBase = TypeVar("TV_RestResourceBase", bound="RestResourceBase")
T_FieldValue = Union[T_SupportedRESTFields, "RestResourceBase"]
T_ListIndex = NewType("T_ListIndex", int)
T_ListSize = NewType("T_ListSize", int)
T_DictKey = Union[
UUID, str, int, float, bool, bytes, Path, IPv4Address, IPv4Network
] # datetime is removed because non-hashable
_T_DictKey = TypeVar(
"_T_DictKey", UUID, str, int, float, bool, bytes, Path, IPv4Address, IPv4Network
)
T_T_DictKey = type[T_DictKey]
T_DictValues = T_FieldValue
_T_DictValues = TypeVar(
"_T_DictValues",
UUID,
str,
int,
float,
bool,
bytes,
datetime,
Path,
IPv4Address,
IPv4Network,
"RestResourceBase",
)
T_T_FieldValue = type(T_FieldValue)
T_T_DictValues = type[T_DictValues]
T_Dict = dict[T_DictKey, T_DictValues]
_T_Dict = dict[_T_DictKey, _T_DictValues]
T_AllSupportedFields = T_Dict | T_FieldValue
T_AllSupportedFiels = T_Dict | T_FieldValue
T_AllSupportedContainers = Union[T_Dict, "RestResourceBase"]

43
src/pyrestresource/test_module.py
View File

@@ -1,43 +0,0 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# pyrestresource (c) by chacha
#
# pyrestresource is licensed under a
# Creative Commons Attribution-NonCommercial-ShareAlike 4.0 International Unported License.
#
# You should have received a copy of the license along with this
# work. If not, see <https://creativecommons.org/licenses/by-nc-sa/4.0/>.
"""Phasellus tellus lectus, volutpat eu dapibus ut, suscipit vel augue.
Tips:
Aliquam non leo vel libero sagittis viverra. Quisque lobortis nunc sit amet augue euismod laoreet.
Note:
Maecenas volutpat porttitor pretium. Aliquam suscipit quis nisi non imperdiet.
Note:
Vivamus et efficitur lorem, eget imperdiet tortor. Integer vel interdum sem.
"""
from __future__ import annotations
from typing import TYPE_CHECKING
if TYPE_CHECKING: # Only imports the below statements during type checking
pass
def test_function(testvar: int) -> int:
""" A test function that return testvar+1 and print "Hello world !"
Proin eget sapien eget ipsum efficitur mollis nec ac nibh.
Note:
Morbi id lectus maximus, condimentum nunc eget, porta felis. In tristique velit tortor.
Args:
testvar: any integer
Returns:
testvar+1
"""
print("Hello world !")
return testvar+1

2
test/__init__.py
View File

@@ -4,4 +4,4 @@
# Creative Commons Attribution-NonCommercial-ShareAlike 4.0 International Unported License.
#
# You should have received a copy of the license along with this
# work. If not, see <https://creativecommons.org/licenses/by-nc-sa/4.0/>.
# work. If not, see <https://creativecommons.org/licenses/by-nc-sa/4.0/>.

80
test/test_rest_login.py Normal file
View File

@@ -0,0 +1,80 @@
from __future__ import annotations
import unittest
from unittest.mock import patch
from os import chdir
from pathlib import Path
from typing import Optional, Annotated
from pydantic import Field
from uuid import UUID, uuid4
from time import time
import json
print(__name__)
print(__package__)
from src.pyrestresource import (
register_rest_rootpoint,
RestResourceBase,
rsrc_verb,
RestRequestParams_GET,
RestRequestParams_POST,
RestRequestParams_Dict_GET,
RestRequestParams_PUT,
T_SupportedRESTFields,
ResourcePlugin_field_default,
ResourcePlugin_RestResourceBase_default,
)
from pprint import pprint
testdir_path = Path(__file__).parent.resolve()
chdir(testdir_path.parent.resolve())
# to allow mock-ing, all the tested classes are in a function
def init_classes():
class ResourcePlugin_Login(ResourcePlugin_RestResourceBase_default):
def handle_resource_get(self, resource: Login, params: RestRequestParams_GET) -> Login:
print("hook GET")
print(resource)
print(params)
return resource
def handle_resource_put(self, resource: Login, params: RestRequestParams_GET) -> Login:
print("hook PUT")
print(resource)
print(params)
return resource
class Login(RestResourceBase):
username: Optional[str] = Field(None, exclude=True)
# username: Optional[str] = Field(None)
secret: Optional[str] = Field(None, exclude=True)
@register_rest_rootpoint
class RootApp(RestResourceBase):
login: Login = Field(
default=Login(),
plugin=ResourcePlugin_Login,
)
# this add the classes to globals to allow using them later on
# => this is only for uinit-testing purpose and is not needed in real use
globals()[Login.__name__] = Login
globals()[RootApp.__name__] = RootApp
class Test_RestAPI_LOGIN(unittest.TestCase):
def setUp(self) -> None:
chdir(testdir_path.parent.resolve())
init_classes()
self.testapp = RootApp()
def test_login(self):
result = self.testapp.process_request("/login", rsrc_verb.GET)
print(result)
result = self.testapp.process_request("/login", rsrc_verb.PUT, '{"username":"toto","secret":"123456"}')
print(result)
result = self.testapp.process_request("/login", rsrc_verb.GET)
print(result)

557
test/test_rest_resource.py Normal file
View File

@@ -0,0 +1,557 @@
from __future__ import annotations
import unittest
from unittest.mock import patch
from os import chdir
from pathlib import Path
from typing import Optional
from pydantic import Field
from uuid import UUID, uuid4
from time import time
import json
print(__name__)
print(__package__)
from src.pyrestresource import (
register_rest_rootpoint,
RestResourceBase,
rsrc_verb,
RestRequestParams_GET,
RestRequestParams_POST,
RestRequestParams_Dict_GET,
T_SupportedRESTFields,
)
from pprint import pprint
testdir_path = Path(__file__).parent.resolve()
chdir(testdir_path.parent.resolve())
# to allow mock-ing, all the tested classes are in a function
def init_classes():
class Info(RestResourceBase):
version: str
api_version: str
class Patch(RestResourceBase):
uuid: UUID = Field(default_factory=uuid4, primary_key=True)
shortname: str
name: Optional[str] = None
description: Optional[str] = None
class Profile(RestResourceBase):
uuid: UUID = Field(default_factory=uuid4, primary_key=True)
shortname: str
name: Optional[str] = None
description: Optional[str] = None
class Game(RestResourceBase):
uuid: UUID = Field(default_factory=uuid4, primary_key=True)
shortname: str
name: Optional[str] = None
description: Optional[str] = None
profiles: dict[UUID, Profile] = {}
patchs: dict[UUID, Patch] = {}
Patch_1 = Patch(uuid="cee1e870-65fa-11ee-8c99-0242ac120002", shortname="testPatch1")
Patch_2 = Patch(uuid="d385a1d2-65fa-11ee-8c99-0242ac120002", shortname="testPatch2")
class User(RestResourceBase):
uuid: UUID = Field(default_factory=uuid4, primary_key=True)
name: str
secret: str = Field(..., exclude=True)
User1 = User(
uuid="8da57a3c-661f-11ee-8c99-0242ac120002",
name="chacha",
secret="la blanquette est bonne",
)
ext_patchs: dict[UUID, Patch] = {}
class Patch2(RestResourceBase):
uuid: UUID = Field(default_factory=uuid4, primary_key=True)
shortname: str
name: Optional[str] = None
description: Optional[str] = None
@register_rest_rootpoint
class RootApp(RestResourceBase):
testValueRoot: float = 3.14
info: Info = Info(version="0.0.1", api_version="0.0.2")
games: dict[UUID, Game] = {
UUID("9b0381d4-65f6-11ee-8c99-0242ac120002"): Game(
uuid="9b0381d4-65f6-11ee-8c99-0242ac120002",
shortname="testGame",
patchs={Patch_1.uuid: Patch_1},
profiles={
UUID("aee1e870-65fa-11ee-8c99-0242ac120002"): Profile(
uuid="aee1e870-65fa-11ee-8c99-0242ac120002",
shortname="testprofile",
)
},
)
}
patchs: dict[UUID, Patch] = {Patch_1.uuid: Patch_1, Patch_2.uuid: Patch_2}
users: dict[UUID, User] = {User1.uuid: User1}
patchs2: dict[UUID, Patch2] = {}
# this add the classes to globals to allow using them later on
# => this is only for uinit-testing purpose and is not needed in real use
globals()[Info.__name__] = Info
globals()[Game.__name__] = Game
globals()[User.__name__] = User
globals()[Profile.__name__] = Profile
globals()[Patch.__name__] = Patch
globals()[Patch2.__name__] = Patch2
globals()[RootApp.__name__] = RootApp
class Test_RestAPI_GET(unittest.TestCase):
def setUp(self) -> None:
chdir(testdir_path.parent.resolve())
init_classes()
self.testapp = RootApp()
def test_get_root(self):
result = self.testapp.process_request("/", rsrc_verb.GET)
self.assertEqual(result, '{"testValueRoot": 3.14}')
def test_get_root__multiple_slash(self):
result = self.testapp.process_request("/////", rsrc_verb.GET)
self.assertEqual(result, '{"testValueRoot": 3.14}')
result = self.testapp.process_request("////", rsrc_verb.GET)
self.assertEqual(result, '{"testValueRoot": 3.14}')
def test_get_root__nested_value(self):
result = self.testapp.process_request("/testValueRoot", rsrc_verb.GET)
self.assertEqual(result, "3.14")
def test_get_root__nested_value__trailing_slash(self):
result = self.testapp.process_request("/testValueRoot/", rsrc_verb.GET)
self.assertEqual(result, "3.14")
result = self.testapp.process_request("/testValueRoot//", rsrc_verb.GET)
self.assertEqual(result, "3.14")
result = self.testapp.process_request("/testValueRoot///", rsrc_verb.GET)
self.assertEqual(result, "3.14")
def test_get_root__nested_value__multiple_slash(self):
result = self.testapp.process_request("//testValueRoot", rsrc_verb.GET)
self.assertEqual(result, "3.14")
result = self.testapp.process_request("///testValueRoot", rsrc_verb.GET)
self.assertEqual(result, "3.14")
def test_get_version(self):
result = self.testapp.process_request("/info", rsrc_verb.GET)
self.assertEqual(result, '{"version": "0.0.1", "api_version": "0.0.2"}')
def test_get_version__trailing_slash(self):
result = self.testapp.process_request("/info/", rsrc_verb.GET)
self.assertEqual(result, '{"version": "0.0.1", "api_version": "0.0.2"}')
result = self.testapp.process_request("/info//", rsrc_verb.GET)
self.assertEqual(result, '{"version": "0.0.1", "api_version": "0.0.2"}')
result = self.testapp.process_request("/info///", rsrc_verb.GET)
self.assertEqual(result, '{"version": "0.0.1", "api_version": "0.0.2"}')
def test_get_version__multiple_slash(self):
result = self.testapp.process_request("//info", rsrc_verb.GET)
self.assertEqual(result, '{"version": "0.0.1", "api_version": "0.0.2"}')
result = self.testapp.process_request("///info", rsrc_verb.GET)
self.assertEqual(result, '{"version": "0.0.1", "api_version": "0.0.2"}')
def test_get_version__nested_value(self):
result = self.testapp.process_request("/info/api_version", rsrc_verb.GET)
self.assertEqual(result, '"0.0.2"')
result = self.testapp.process_request("/info/version", rsrc_verb.GET)
self.assertEqual(result, '"0.0.1"')
def test_get_dict_games(self):
result = self.testapp.process_request("/games", rsrc_verb.GET)
self.assertEqual(result, '["9b0381d4-65f6-11ee-8c99-0242ac120002"]')
def test_get_dict_patchs(self):
result = self.testapp.process_request("/patchs", rsrc_verb.GET)
self.assertEqual(
result,
'["cee1e870-65fa-11ee-8c99-0242ac120002", "d385a1d2-65fa-11ee-8c99-0242ac120002"]',
)
def test_get_dict_patch_element(self):
result = self.testapp.process_request("/patchs/cee1e870-65fa-11ee-8c99-0242ac120002", rsrc_verb.GET)
self.assertEqual(
result,
'{"uuid": "cee1e870-65fa-11ee-8c99-0242ac120002", "shortname": "testPatch1", "name": null, "description": null}',
)
def test_get_dict_game_element(self):
result = self.testapp.process_request("/games/9b0381d4-65f6-11ee-8c99-0242ac120002", rsrc_verb.GET)
expected = '{"uuid": "9b0381d4-65f6-11ee-8c99-0242ac120002", "shortname": "testGame", "name": null, "description": null}'
self.assertEqual(result, expected)
def test_get_dict_game_element__nested_value(self):
result = self.testapp.process_request("/games/9b0381d4-65f6-11ee-8c99-0242ac120002/shortname", rsrc_verb.GET)
expected = '"testGame"'
self.assertEqual(result, expected)
def test_get_dict_game_element__nested_value2(self):
result = self.testapp.process_request("/games/9b0381d4-65f6-11ee-8c99-0242ac120002/uuid", rsrc_verb.GET)
expected = '"9b0381d4-65f6-11ee-8c99-0242ac120002"'
self.assertEqual(result, expected)
def test_get_nested_dict_games_patchs(self):
result = self.testapp.process_request("/games/9b0381d4-65f6-11ee-8c99-0242ac120002/patchs", rsrc_verb.GET)
self.assertEqual(result, '["cee1e870-65fa-11ee-8c99-0242ac120002"]')
def test_get_nested_dict_games_patch_element(self):
result = self.testapp.process_request(
"/games/9b0381d4-65f6-11ee-8c99-0242ac120002/patchs/cee1e870-65fa-11ee-8c99-0242ac120002",
rsrc_verb.GET,
)
expected = '{"uuid": "cee1e870-65fa-11ee-8c99-0242ac120002", "shortname": "testPatch1", "name": null, "description": null}'
self.assertEqual(result, expected)
def test_get_nested_dict_games_patch_element__nested_value(self):
result = self.testapp.process_request(
"/games/9b0381d4-65f6-11ee-8c99-0242ac120002/patchs/cee1e870-65fa-11ee-8c99-0242ac120002/uuid",
rsrc_verb.GET,
)
self.assertEqual(result, '"cee1e870-65fa-11ee-8c99-0242ac120002"')
def test_get_dict_game_element__API_nested(self):
result = self.testapp.process_request("/games/9b0381d4-65f6-11ee-8c99-0242ac120002?API_nested=True", rsrc_verb.GET)
expected = '{"uuid": "9b0381d4-65f6-11ee-8c99-0242ac120002", "shortname": "testGame", "name": null, "description": null}'
self.assertEqual(result, expected)
def test_get_dict_users(self):
result = self.testapp.process_request("/users", rsrc_verb.GET)
self.assertEqual(result, '["8da57a3c-661f-11ee-8c99-0242ac120002"]')
def test_get_dict_user_element(self):
result = self.testapp.process_request("/users/8da57a3c-661f-11ee-8c99-0242ac120002", rsrc_verb.GET)
self.assertEqual(
result,
'{"uuid": "8da57a3c-661f-11ee-8c99-0242ac120002", "name": "chacha"}',
"no secret seen",
)
def test_get_dict_user_element2(self):
result = self.testapp.process_request("/users/8da57a3c-661f-11ee-8c99-0242ac120002?API_nested=True", rsrc_verb.GET)
self.assertEqual(
result,
'{"uuid": "8da57a3c-661f-11ee-8c99-0242ac120002", "name": "chacha"}',
"no secret seen",
)
def test_get_dict_user_element__nested_value(self):
result = self.testapp.process_request("/users/8da57a3c-661f-11ee-8c99-0242ac120002/name", rsrc_verb.GET)
self.assertEqual(result, '"chacha"')
def test_get_dict_user_element__nested_value__forbiden(self):
with self.assertRaises(RuntimeError): # TODO: custom exception
self.testapp.process_request("/users/8da57a3c-661f-11ee-8c99-0242ac120002/secret", rsrc_verb.GET)
def test_get_dict_user_element__nested_value__forbiden2(self):
with self.assertRaises(RuntimeError): # TODO: custom exception
self.testapp.process_request(
"/users/8da57a3c-661f-11ee-8c99-0242ac120002/secret?API_nested=True",
rsrc_verb.GET,
)
class Test_RestAPI_PUT(unittest.TestCase):
def setUp(self) -> None:
chdir(testdir_path.parent.resolve())
init_classes()
self.testapp = RootApp()
def test_put_info(self):
self.testapp.process_request("/info", rsrc_verb.PUT, '{"version": "1.2.3", "api_version": "3.2.1"}')
result = self.testapp.process_request("/info", rsrc_verb.GET)
self.assertEqual(result, '{"version": "1.2.3", "api_version": "3.2.1"}')
def test_put_dict_user_nested_value(self):
self.testapp.process_request(
"/users/8da57a3c-661f-11ee-8c99-0242ac120002/name",
rsrc_verb.PUT,
'"chacha2"',
)
result = self.testapp.process_request("/users/8da57a3c-661f-11ee-8c99-0242ac120002/name", rsrc_verb.GET)
self.assertEqual(result, '"chacha2"')
def test_put_user_nested_value__forbiden(self):
with self.assertRaises(RuntimeError): # TODO: custom exception
self.testapp.process_request(
"/users/8da57a3c-661f-11ee-8c99-0242ac120002/secret",
rsrc_verb.PUT,
'"test"',
)
def test_put_dict_user_element(self):
self.testapp.process_request(
"/users/8da57a3c-661f-11ee-8c99-0242ac120002",
rsrc_verb.PUT,
'{"name": "testUser4", "secret": "test5"}',
)
result = self.testapp.process_request("/users", rsrc_verb.GET)
expected = '["8da57a3c-661f-11ee-8c99-0242ac120002"]'
self.assertEqual(result, expected)
result = self.testapp.process_request("/users/8da57a3c-661f-11ee-8c99-0242ac120002", rsrc_verb.GET)
expected = '{"uuid": "8da57a3c-661f-11ee-8c99-0242ac120002", "name": "testUser4"}'
self.assertEqual(result, expected)
def test_put_dict_patch__nested(self):
self.testapp.process_request(
"/games/9b0381d4-65f6-11ee-8c99-0242ac120002/patchs/cee1e870-65fa-11ee-8c99-0242ac120002",
rsrc_verb.PUT,
'{"shortname": "testPatch998", "name": "MyPatch", "description": "MyDescription123"}',
)
result = self.testapp.process_request(
"/games/9b0381d4-65f6-11ee-8c99-0242ac120002/patchs/cee1e870-65fa-11ee-8c99-0242ac120002",
rsrc_verb.GET,
)
expected = '{"uuid": "cee1e870-65fa-11ee-8c99-0242ac120002", "shortname": "testPatch998", "name": "MyPatch", "description": "MyDescription123"}'
self.assertEqual(result, expected)
class Test_RestAPI_POST(unittest.TestCase):
def setUp(self) -> None:
chdir(testdir_path.parent.resolve())
init_classes()
self.testapp = RootApp()
def test_post_dict_user__API_key(self):
result = self.testapp.process_request(
"/users?API_key=e5e87d32-662b-11ee-8c99-0242ac120002",
rsrc_verb.POST,
'{"name": "testUser", "secret": "test"}',
)
self.assertEqual(result, '"e5e87d32-662b-11ee-8c99-0242ac120002"')
result = self.testapp.process_request("/users", rsrc_verb.GET)
expected = '["8da57a3c-661f-11ee-8c99-0242ac120002", "e5e87d32-662b-11ee-8c99-0242ac120002"]'
self.assertEqual(result, expected)
result = self.testapp.process_request("/users/e5e87d32-662b-11ee-8c99-0242ac120002", rsrc_verb.GET)
expected = '{"uuid": "e5e87d32-662b-11ee-8c99-0242ac120002", "name": "testUser"}'
self.assertEqual(result, expected)
def test_post_dict_user__nested_key(self):
result = self.testapp.process_request(
"/users",
rsrc_verb.POST,
'{"name": "testUser2", "secret": "test", "uuid":"e7e86d32-662b-11ee-8c99-0242ac120002"}',
)
self.assertEqual(result, '"e7e86d32-662b-11ee-8c99-0242ac120002"')
result = self.testapp.process_request("/users", rsrc_verb.GET)
expected = '["8da57a3c-661f-11ee-8c99-0242ac120002", "e7e86d32-662b-11ee-8c99-0242ac120002"]'
self.assertEqual(result, expected)
result = self.testapp.process_request("/users/e7e86d32-662b-11ee-8c99-0242ac120002", rsrc_verb.GET)
expected = '{"uuid": "e7e86d32-662b-11ee-8c99-0242ac120002", "name": "testUser2"}'
self.assertEqual(result, expected)
@patch(f"{__loader__.name }.uuid4")
def test_post_dict_user__auto_key(self, mock_uuid4):
mock_uuid4.return_value = UUID("5faccb2e-69aa-11ee-8c99-0242ac120002")
# recreating classes & objects to force using the Mock-ed uuid4
init_classes()
self.testapp = RootApp()
result = self.testapp.process_request("/users", rsrc_verb.POST, '{"name": "testUser3", "secret": "test"}')
self.assertEqual(result, '"5faccb2e-69aa-11ee-8c99-0242ac120002"')
result = self.testapp.process_request("/users", rsrc_verb.GET)
expected = '["8da57a3c-661f-11ee-8c99-0242ac120002", "5faccb2e-69aa-11ee-8c99-0242ac120002"]'
self.assertEqual(result, expected)
result = self.testapp.process_request("/users/5faccb2e-69aa-11ee-8c99-0242ac120002", rsrc_verb.GET)
expected = '{"uuid": "5faccb2e-69aa-11ee-8c99-0242ac120002", "name": "testUser3"}'
self.assertEqual(result, expected)
def test_post_dict_patch__nested_API_key(self):
self.testapp.process_request(
"/games/9b0381d4-65f6-11ee-8c99-0242ac120002/patchs?API_key=cee1e971-65fa-11ee-8c99-0242ac120002",
rsrc_verb.POST,
'{"shortname": "testPatch99", "name": "MyPatch", "description": "MyDescription"}',
)
result = self.testapp.process_request(
"/games/9b0381d4-65f6-11ee-8c99-0242ac120002/patchs/cee1e971-65fa-11ee-8c99-0242ac120002",
rsrc_verb.GET,
)
expected = '{"uuid": "cee1e971-65fa-11ee-8c99-0242ac120002", "shortname": "testPatch99", "name": "MyPatch", "description": "MyDescription"}'
self.assertEqual(result, expected)
class Test_RestAPI_DELETE(unittest.TestCase):
def setUp(self) -> None:
chdir(testdir_path.parent.resolve())
init_classes()
self.testapp = RootApp()
def test_delete_dict_user__API_key(self):
self.testapp.process_request("/users?API_key=8da57a3c-661f-11ee-8c99-0242ac120002", rsrc_verb.DELETE)
result = self.testapp.process_request("/users", rsrc_verb.GET)
expected = "[]"
self.assertEqual(result, expected)
def test_delete_dict_user__All(self):
result = self.testapp.process_request(
"/users?API_key=e5e87d32-662b-11ee-8c99-0242ac120002",
rsrc_verb.POST,
'{"name": "testUser", "secret": "test"}',
)
self.assertEqual(result, '"e5e87d32-662b-11ee-8c99-0242ac120002"')
result = self.testapp.process_request("/users", rsrc_verb.GET)
expected = '["8da57a3c-661f-11ee-8c99-0242ac120002", "e5e87d32-662b-11ee-8c99-0242ac120002"]'
self.assertEqual(result, expected)
self.testapp.process_request("/users", rsrc_verb.DELETE)
result = self.testapp.process_request("/users", rsrc_verb.GET)
expected = "[]"
self.assertEqual(result, expected)
def test_delete_dict_user_element(self):
self.testapp.process_request("/users/8da57a3c-661f-11ee-8c99-0242ac120002", rsrc_verb.DELETE)
result = self.testapp.process_request("/users", rsrc_verb.GET)
expected = "[]"
self.assertEqual(result, expected)
def test_delete_nested_dict_games_patch_element(self):
self.testapp.process_request(
"/games/9b0381d4-65f6-11ee-8c99-0242ac120002/patchs/cee1e870-65fa-11ee-8c99-0242ac120002",
rsrc_verb.DELETE,
)
result = self.testapp.process_request("/games/9b0381d4-65f6-11ee-8c99-0242ac120002/patchs", rsrc_verb.GET)
expected = "[]"
self.assertEqual(result, expected)
def test_delete_nested_dict_games_patch_API_key(self):
self.testapp.process_request(
"/games/9b0381d4-65f6-11ee-8c99-0242ac120002/patchs?API_key=cee1e870-65fa-11ee-8c99-0242ac120002",
rsrc_verb.DELETE,
)
result = self.testapp.process_request("/games/9b0381d4-65f6-11ee-8c99-0242ac120002/patchs", rsrc_verb.GET)
expected = "[]"
self.assertEqual(result, expected)
def test_delete_nested_dict_games_patch_All(self):
self.testapp.process_request("/games/9b0381d4-65f6-11ee-8c99-0242ac120002/patchs", rsrc_verb.DELETE)
result = self.testapp.process_request("/games/9b0381d4-65f6-11ee-8c99-0242ac120002/patchs", rsrc_verb.GET)
expected = "[]"
self.assertEqual(result, expected)
class Test_RestAPI_PERFO(unittest.TestCase):
def setUp(self) -> None:
chdir(testdir_path.parent.resolve())
init_classes()
self.testapp = RootApp()
@unittest.skip
def test_perf_dict(self):
print(f"LIB INTERNAL PERF TEST")
n_loop = 10000
start = time()
for i in range(n_loop):
self.testapp.process_request(f"/users/8da57a3c-661f-11ee-8c99-0242ac120002", rsrc_verb.GET)
end = time()
print(f"GET 1st level dict: {int(n_loop/(end-start))} Req/s")
start = time()
for i in range(n_loop):
newUUID = uuid4()
self.testapp.process_request(
f"/users?API_key={newUUID}",
rsrc_verb.POST,
'{"name": "testUser", "secret": "test"}',
)
end = time()
print(f"POST 1st level dict (API_key): {int(n_loop/(end-start))} Req/s")
start = time()
for i in range(n_loop):
newUUID = uuid4()
self.testapp.process_request(
f"/users?API_key={newUUID}",
rsrc_verb.POST,
'{"name": "testUser", "secret": "test"}',
)
self.testapp.process_request(f"/users/{newUUID}", rsrc_verb.GET)
end = time()
print(f"POST/GET 1st level dict (API_key): {int(n_loop/(end-start))} Req/s")
start = time()
for i in range(n_loop):
result = self.testapp.process_request(f"/users", rsrc_verb.POST, '{"name": "testUser", "secret": "test"}')
self.testapp.process_request(f"/users/{json.loads(result)}", rsrc_verb.GET)
end = time()
print(f"POST/GET 1st level dict (autokey): {int(n_loop/(end-start))} Req/s")
start = time()
for i in range(n_loop):
self.testapp.process_request(
f"/games/9b0381d4-65f6-11ee-8c99-0242ac120002/shortname",
rsrc_verb.PUT,
'"TestValue!!"',
)
self.testapp.process_request(f"/games/9b0381d4-65f6-11ee-8c99-0242ac120002/shortname", rsrc_verb.GET)
end = time()
print(f"PUT/GET 1st level (value) dict: {int(n_loop/(end-start))} Req/s")
start = time()
for i in range(n_loop):
self.testapp.process_request(
f"/games/9b0381d4-65f6-11ee-8c99-0242ac120002/patchs/cee1e870-65fa-11ee-8c99-0242ac120002",
rsrc_verb.GET,
)
end = time()
print(f"GET 2nd level dict: {int(n_loop/(end-start))} Req/s")
start = time()
for i in range(n_loop):
self.testapp.process_request(
f"/games/9b0381d4-65f6-11ee-8c99-0242ac120002/patchs/cee1e870-65fa-11ee-8c99-0242ac120002/shortname",
rsrc_verb.GET,
)
end = time()
print(f"GET 2nd level (value) dict: {int(n_loop/(end-start))} Req/s")
start = time()
for i in range(n_loop):
self.testapp.process_request(
f"/games/9b0381d4-65f6-11ee-8c99-0242ac120002/patchs/cee1e870-65fa-11ee-8c99-0242ac120002/shortname",
rsrc_verb.PUT,
'"TestValue!!"',
)
self.testapp.process_request(
f"/games/9b0381d4-65f6-11ee-8c99-0242ac120002/patchs/cee1e870-65fa-11ee-8c99-0242ac120002/shortname",
rsrc_verb.GET,
)
end = time()
print(f"PUT/GET 2nd level (value) dict: {int(n_loop/(end-start))} Req/s")

216
test/test_rest_resource_plugins.py Normal file
View File

@@ -0,0 +1,216 @@
from __future__ import annotations
import unittest
from unittest.mock import patch
from os import chdir
from pathlib import Path
from typing import Optional, Annotated
from pydantic import Field
from uuid import UUID, uuid4
from time import time
import json
print(__name__)
print(__package__)
from src.pyrestresource import (
register_rest_rootpoint,
RestResourceBase,
rsrc_verb,
RestRequestParams_GET,
RestRequestParams_POST,
RestRequestParams_Dict_GET,
RestRequestParams_PUT,
T_SupportedRESTFields,
ResourcePlugin_field_default,
ResourcePlugin_RestResourceBase_default,
)
from pprint import pprint
testdir_path = Path(__file__).parent.resolve()
chdir(testdir_path.parent.resolve())
# to allow mock-ing, all the tested classes are in a function
def init_classes():
class ResourcePlugin_version_get(ResourcePlugin_field_default):
def handle_field_get(self, resource: Info_get, params: RestRequestParams_GET) -> Info_get:
return "1.5.6"
class ResourcePlugin_version_put(ResourcePlugin_field_default):
def handle_field_put(self, resource: Info_put, params: RestRequestParams_PUT) -> Info_put:
return "42"
class ResourcePlugin_Info(ResourcePlugin_RestResourceBase_default):
def handle_resource_get(self, resource: Info_get, params: RestRequestParams_GET) -> Info_get:
return Info_get(version="65.45", api_version="98.321")
class Info_get(RestResourceBase):
# test plugin injection within annotation
# + test plugin on a simple field
version: Annotated[str, Field(plugin=ResourcePlugin_version_get)]
api_version: str
class Info_put(RestResourceBase):
# test plugin injection within annotation
# + test plugin on a simple field
version: Annotated[str, Field(plugin=ResourcePlugin_version_put)]
api_version: str
@register_rest_rootpoint
class RootApp(RestResourceBase):
# test plugin injection within Field value
# + test plugin on a RestResourceBase field
info: Info_get = Field(
default=Info_get(version="0.0.1", api_version="0.0.2"),
plugin=ResourcePlugin_Info,
)
info_put: Info_put = Field(
default=Info_put(version="0.0.1", api_version="0.0.2"),
)
info2: Info_get = Field(default=Info_get(version="0.0.2", api_version="0.0.3"))
# this add the classes to globals to allow using them later on
# => this is only for uinit-testing purpose and is not needed in real use
globals()[Info_get.__name__] = Info_get
globals()[Info_put.__name__] = Info_put
globals()[RootApp.__name__] = RootApp
def init_bad_plugin1():
# plugin with missing handle_resource_put() method
class ResourcePlugin_TestResource:
def handle_field_get(self, resource: TestResource, params: RestRequestParams_GET) -> TestResource:
return resource
class TestResource(RestResourceBase):
tetvaluestr: Annotated[str, Field(plugin=ResourcePlugin_TestResource)]
@register_rest_rootpoint
class RootApp2(RestResourceBase):
test: TestResource = Field(default=TestResource(tetvaluestr="testvalue"))
RootApp2()
def init_bad_plugin2():
# plugin with missing handle_resource_get() method
class ResourcePlugin_TestResource:
def handle_field_put(self, resource: TestResource, params: RestRequestParams_PUT) -> TestResource:
return resource
class TestResource(RestResourceBase):
tetvaluestr: Annotated[str, Field(plugin=ResourcePlugin_TestResource)]
@register_rest_rootpoint
class RootApp2(RestResourceBase):
test: TestResource = Field(default=TestResource(tetvaluestr="testvalue"))
RootApp2()
def init_bad_plugin3():
# wrong plugin
class ResourcePlugin_TestResource(ResourcePlugin_RestResourceBase_default):
pass
class TestResource(RestResourceBase):
tetvaluestr: Annotated[str, Field(plugin=ResourcePlugin_TestResource)]
@register_rest_rootpoint
class RootApp2(RestResourceBase):
test: TestResource = Field(default=TestResource(tetvaluestr="testvalue"))
RootApp2()
class Test_RestAPI_Plugin_PUT(unittest.TestCase):
def setUp(self) -> None:
chdir(testdir_path.parent.resolve())
init_classes()
self.testapp = RootApp()
def test_put_field_version_fieldplugin(self):
self.testapp.process_request("/info_put/version", rsrc_verb.PUT, '"1.5.6"')
result = self.testapp.process_request("/info_put", rsrc_verb.GET)
print(result)
result = self.testapp.process_request("/info_put/version", rsrc_verb.GET)
print(result)
self.assertEqual(result, '"42"')
def test_put_field_version_resourceplugin(self):
self.testapp.process_request("/info_put", rsrc_verb.PUT, '{"version": "1.5.6", "api_version": "98.321"}')
result = self.testapp.process_request("/info_put", rsrc_verb.GET)
self.assertEqual(result, '{"version": "42", "api_version": "98.321"}')
class Test_RestAPI_Plugin_GET(unittest.TestCase):
def setUp(self) -> None:
chdir(testdir_path.parent.resolve())
init_classes()
self.testapp = RootApp()
def test_get_root(self):
result = self.testapp.process_request("/", rsrc_verb.GET)
self.assertEqual(result, "{}")
def test_get_version(self):
result = self.testapp.process_request("/info", rsrc_verb.GET)
self.assertEqual(result, '{"version": "1.5.6", "api_version": "98.321"}')
result = self.testapp.process_request("/info2", rsrc_verb.GET)
self.assertEqual(result, '{"version": "1.5.6", "api_version": "0.0.3"}')
def test_get_version__trailing_slash(self):
result = self.testapp.process_request("/info/", rsrc_verb.GET)
self.assertEqual(result, '{"version": "1.5.6", "api_version": "98.321"}')
result = self.testapp.process_request("/info//", rsrc_verb.GET)
self.assertEqual(result, '{"version": "1.5.6", "api_version": "98.321"}')
result = self.testapp.process_request("/info///", rsrc_verb.GET)
self.assertEqual(result, '{"version": "1.5.6", "api_version": "98.321"}')
result = self.testapp.process_request("/info2/", rsrc_verb.GET)
self.assertEqual(result, '{"version": "1.5.6", "api_version": "0.0.3"}')
result = self.testapp.process_request("/info2//", rsrc_verb.GET)
self.assertEqual(result, '{"version": "1.5.6", "api_version": "0.0.3"}')
result = self.testapp.process_request("/info2///", rsrc_verb.GET)
self.assertEqual(result, '{"version": "1.5.6", "api_version": "0.0.3"}')
def test_get_version__multiple_slash(self):
result = self.testapp.process_request("//info", rsrc_verb.GET)
self.assertEqual(result, '{"version": "1.5.6", "api_version": "98.321"}')
result = self.testapp.process_request("///info", rsrc_verb.GET)
self.assertEqual(result, '{"version": "1.5.6", "api_version": "98.321"}')
result = self.testapp.process_request("//info2", rsrc_verb.GET)
self.assertEqual(result, '{"version": "1.5.6", "api_version": "0.0.3"}')
result = self.testapp.process_request("///info2", rsrc_verb.GET)
self.assertEqual(result, '{"version": "1.5.6", "api_version": "0.0.3"}')
def test_get_version__nested_value(self):
result = self.testapp.process_request("/info/api_version", rsrc_verb.GET)
self.assertEqual(result, '"98.321"')
result = self.testapp.process_request("/info/version", rsrc_verb.GET)
self.assertEqual(result, '"1.5.6"')
result = self.testapp.process_request("/info2/api_version", rsrc_verb.GET)
self.assertEqual(result, '"0.0.3"')
result = self.testapp.process_request("/info2/version", rsrc_verb.GET)
self.assertEqual(result, '"1.5.6"')
def test_defect_plugin_field(self):
with self.assertRaises(RuntimeError):
init_bad_plugin1()
with self.assertRaises(RuntimeError):
init_bad_plugin2()
with self.assertRaises(RuntimeError):
init_bad_plugin3()

163
test/test_rest_resource_walker.py Normal file
View File

@@ -0,0 +1,163 @@
from __future__ import annotations
import unittest
from typing import Annotated, Optional
from os import chdir
from pathlib import Path
from pydantic import Field
from io import StringIO
from contextlib import redirect_stdout, redirect_stderr
print(__name__)
print(__package__)
from src.pyrestresource import (
RestResourceBase,
)
from src.pyrestresource.rest_resource_walker import (
RestResourceWalker_Root,
RestResourceWalker_Sub_T_Dict,
RestResourceWalker_Sub_RestFields,
RestResourceWalker_Sub_RestResourceBase,
)
testdir_path = Path(__file__).parent.resolve()
chdir(testdir_path.parent.resolve())
class RestResourceWalker_Sub_T_Dict_TEST_Print(RestResourceWalker_Sub_T_Dict):
counter: dict[str, int] = {}
@classmethod
def init_sub(cls, walker: RestResourceWalker_Root) -> None:
cls.counter = {}
def process(self) -> None:
if self.resource_name not in self.counter:
self.counter[self.resource_name] = 0
self.counter[self.resource_name] = self.counter[self.resource_name] + 1
print(f"DICT {self.resource_name} {self.counter[self.resource_name]}")
class RestResourceWalker_Sub_RestFields_TEST_Print(RestResourceWalker_Sub_RestFields):
counter: dict[str, int] = {}
@classmethod
def init_sub(cls, walker: RestResourceWalker_Root) -> None:
cls.counter = {}
def process(self) -> None:
if self.resource_name not in self.counter:
self.counter[self.resource_name] = 0
self.counter[self.resource_name] = self.counter[self.resource_name] + 1
print(f"FIELD {self.resource_name} {self.counter[self.resource_name]}")
class RestResourceWalker_Sub_RestResourceBase_TEST_Print(
RestResourceWalker_Sub_RestResourceBase
):
counter: dict[str, int] = {}
@classmethod
def init_sub(cls, walker: RestResourceWalker_Root) -> None:
cls.counter = {}
def process(self) -> None:
if self.resource_name not in self.counter:
self.counter[self.resource_name] = 0
self.counter[self.resource_name] = self.counter[self.resource_name] + 1
print(f"RestResource {self.resource_name} {self.counter[self.resource_name]}")
class RestResourceWalker_Root_TEST_Print(RestResourceWalker_Root):
cls_RestResourceWalker_Sub = [
RestResourceWalker_Sub_T_Dict_TEST_Print,
RestResourceWalker_Sub_RestFields_TEST_Print,
RestResourceWalker_Sub_RestResourceBase_TEST_Print,
]
def init_classes():
class Info(RestResourceBase):
version: str
api_version: str
class People(RestResourceBase):
last_name: str
class RootApp(RestResourceBase):
info: Info = Field(default=Info(version="0.0.1", api_version="0.0.2"))
info2: Info = Info(version="0.0.2", api_version="0.0.3")
peoples: dict[str, People] = {
"john": People(last_name="Doe"),
"jane": People(last_name="Roe"),
}
test_string: str = "test value"
test_string_opt: Optional[str] = None
test_int: int = 42
# this add the classes to globals to allow using them later on
# => this is only for uinit-testing purpose and is not needed in real use
globals()[Info.__name__] = Info
globals()[People.__name__] = People
globals()[RootApp.__name__] = RootApp
class Test_Walker(unittest.TestCase):
def setUp(self) -> None:
chdir(testdir_path.parent.resolve())
init_classes()
def test_walk_class(self):
test = RestResourceWalker_Root_TEST_Print(RootApp)
with redirect_stdout(StringIO()) as capted_stdout, redirect_stderr(
StringIO()
) as capted_stderr:
test.process()
self.assertIn("RestResource info 1", capted_stdout.getvalue())
self.assertIn("RestResource info2 1", capted_stdout.getvalue())
self.assertIn("DICT peoples 1", capted_stdout.getvalue())
self.assertIn("FIELD test_string 1", capted_stdout.getvalue())
self.assertIn("FIELD test_string_opt 1", capted_stdout.getvalue())
self.assertIn("FIELD test_int 1", capted_stdout.getvalue())
self.assertIn("FIELD version 1", capted_stdout.getvalue())
self.assertIn("FIELD version 2", capted_stdout.getvalue())
self.assertIn("FIELD api_version 1", capted_stdout.getvalue())
self.assertIn("FIELD api_version 2", capted_stdout.getvalue())
self.assertIn("RestResource peoples 1", capted_stdout.getvalue())
self.assertIn("FIELD last_name 1", capted_stdout.getvalue())
def test_walk_obj(self):
instRootApp = RootApp()
test = RestResourceWalker_Root_TEST_Print(instRootApp)
with redirect_stdout(StringIO()) as capted_stdout, redirect_stderr(
StringIO()
) as capted_stderr:
test.process()
self.assertIn("RestResource info 1", capted_stdout.getvalue())
self.assertIn("RestResource info2 1", capted_stdout.getvalue())
self.assertIn("DICT peoples 1", capted_stdout.getvalue())
self.assertIn("FIELD test_string 1", capted_stdout.getvalue())
self.assertIn("FIELD test_string_opt 1", capted_stdout.getvalue())
self.assertIn("FIELD test_int 1", capted_stdout.getvalue())
self.assertIn("FIELD version 1", capted_stdout.getvalue())
self.assertIn("FIELD version 2", capted_stdout.getvalue())
self.assertIn("FIELD api_version 1", capted_stdout.getvalue())
self.assertIn("FIELD api_version 2", capted_stdout.getvalue())
self.assertIn("RestResource peoples 1", capted_stdout.getvalue())
self.assertIn("FIELD last_name 1", capted_stdout.getvalue())
def test_walk_obj_nested_RestResource(self):
instRootApp = RootApp()
test = RestResourceWalker_Root_TEST_Print(instRootApp.info)
with redirect_stdout(StringIO()) as capted_stdout, redirect_stderr(
StringIO()
) as capted_stderr:
test.process()
self.assertIn("FIELD version 1", capted_stdout.getvalue())
self.assertIn("FIELD api_version 1", capted_stdout.getvalue())

150
test/test_rest_resource_walker_tree.py Normal file
View File

@@ -0,0 +1,150 @@
from __future__ import annotations
import unittest
from typing import Annotated, Optional
from os import chdir
from pathlib import Path
from pydantic import Field
from io import StringIO
from contextlib import redirect_stdout, redirect_stderr
print(__name__)
print(__package__)
from src.pyrestresource import (
RestResourceBase,
)
from src.pyrestresource.rest_resource_walker import (
RestResourceWalkerFutureResult,
RestResourceWalker_Root,
RestResourceWalker_Sub_T_Dict,
RestResourceWalker_Sub_RestFields,
RestResourceWalker_Sub_RestResourceBase,
)
testdir_path = Path(__file__).parent.resolve()
chdir(testdir_path.parent.resolve())
class RestResourceWalkerFutureResult_RestFields_Test(RestResourceWalkerFutureResult[dict]):
def process_future(self, result: Optional[list[dict]]) -> Optional[dict]:
res = dict()
res[self.source.resource_name] = False
return res
class RestResourceWalker_Sub_RestFields_TEST_Print(RestResourceWalker_Sub_RestFields):
cls_RestResourceWalkerFutureResult = RestResourceWalkerFutureResult_RestFields_Test
class RestResourceWalkerFutureResult_RestResourceBase_Test(RestResourceWalkerFutureResult[dict]):
def process_future(self, result: Optional[list[dict]]) -> Optional[dict]:
res = dict()
res[self.source.resource_name] = dict()
for subres in result:
res[self.source.resource_name] = res[self.source.resource_name] | subres
return res
class RestResourceWalker_Sub_RestResourceBase_TEST_Print(RestResourceWalker_Sub_RestResourceBase):
cls_RestResourceWalkerFutureResult = RestResourceWalkerFutureResult_RestResourceBase_Test
class RestResourceWalkerFutureResult_Dict_Test(RestResourceWalkerFutureResult[dict]):
def process_future(self, result: Optional[list[dict]]) -> Optional[dict]:
res = dict()
for subres in result:
res = res | subres
return res
class RestResourceWalker_Sub_T_Dict_TEST_Print(RestResourceWalker_Sub_T_Dict):
cls_RestResourceWalkerFutureResult = RestResourceWalkerFutureResult_Dict_Test
class RestResourceWalker_Root_TEST_Print(RestResourceWalker_Root):
cls_RestResourceWalker_Sub = [
RestResourceWalker_Sub_T_Dict_TEST_Print,
RestResourceWalker_Sub_RestFields_TEST_Print,
RestResourceWalker_Sub_RestResourceBase_TEST_Print,
]
def init_classes():
class Info(RestResourceBase):
version: str
api_version: str
class People(RestResourceBase):
last_name: str
class RootApp(RestResourceBase):
info: Info = Field(default=Info(version="0.0.1", api_version="0.0.2"))
info2: Info = Info(version="0.0.2", api_version="0.0.3")
peoples: dict[str, People] = {
"john": People(last_name="Doe"),
"jane": People(last_name="Roe"),
}
test_string: str = "test value"
test_string_opt: Optional[str] = None
test_int: int = 42
# this add the classes to globals to allow using them later on
# => this is only for uinit-testing purpose and is not needed in real use
globals()[Info.__name__] = Info
globals()[People.__name__] = People
globals()[RootApp.__name__] = RootApp
class Test_Walker_tree(unittest.TestCase):
def setUp(self) -> None:
chdir(testdir_path.parent.resolve())
init_classes()
def test_walk_class(self):
test = RestResourceWalker_Root_TEST_Print(RootApp)
res = test.process()
self.assertDictEqual(
res,
{
"/": {
"info": {"version": False, "api_version": False},
"info2": {"version": False, "api_version": False},
"peoples": {"last_name": False},
"test_string": False,
"test_string_opt": False,
"test_int": False,
}
},
)
def test_walk_obj(self):
instRootApp = RootApp()
test = RestResourceWalker_Root_TEST_Print(instRootApp)
res = test.process()
self.assertDictEqual(
res,
{
"/": {
"info": {"version": False, "api_version": False},
"info2": {"version": False, "api_version": False},
"peoples": {"last_name": False},
"test_string": False,
"test_string_opt": False,
"test_int": False,
}
},
)
def test_walk_obj_nested_RestResource(self):
instRootApp = RootApp()
test = RestResourceWalker_Root_TEST_Print(instRootApp.info)
res = test.process()
self.assertDictEqual(
res,
{
"/": {"version": False, "api_version": False},
},
)

382
test/test_rest_webserver.py Normal file
View File

@@ -0,0 +1,382 @@
from __future__ import annotations
import unittest
from unittest.mock import patch
from os import chdir
from pathlib import Path
from typing import Optional
from pydantic import Field
from uuid import UUID, uuid4
from time import time, sleep
import json
import uvicorn
import socket
import requests
from contextlib import closing
from multiprocessing import Process
print(__name__)
print(__package__)
from src.pyrestresource import (
register_rest_rootpoint,
RestResourceBase,
rsrc_verb,
RestRequestParams_GET,
RestRequestParams_POST,
RestRequestParams_Dict_GET,
T_SupportedRESTFields,
)
from pprint import pprint
testdir_path = Path(__file__).parent.resolve()
chdir(testdir_path.parent.resolve())
# to allow mock-ing, all the tested classes are in a function
def init_classes():
class Info(RestResourceBase):
version: str
api_version: str
class Patch(RestResourceBase):
uuid: UUID = Field(default_factory=uuid4, primary_key=True)
shortname: str
name: Optional[str] = None
description: Optional[str] = None
class Profile(RestResourceBase):
uuid: UUID = Field(default_factory=uuid4, primary_key=True)
shortname: str
name: Optional[str] = None
description: Optional[str] = None
class Game(RestResourceBase):
uuid: UUID = Field(default_factory=uuid4, primary_key=True)
shortname: str
name: Optional[str] = None
description: Optional[str] = None
profiles: dict[UUID, Profile] = {}
patchs: dict[UUID, Patch] = {}
Patch_1 = Patch(uuid="cee1e870-65fa-11ee-8c99-0242ac120002", shortname="testPatch1")
Patch_2 = Patch(uuid="d385a1d2-65fa-11ee-8c99-0242ac120002", shortname="testPatch2")
class User(RestResourceBase):
uuid: UUID = Field(default_factory=uuid4, primary_key=True)
name: str
secret: str = Field(..., exclude=True)
User1 = User(
uuid="8da57a3c-661f-11ee-8c99-0242ac120002",
name="chacha",
secret="la blanquette est bonne",
)
class Patch2(RestResourceBase):
uuid: UUID = Field(default_factory=uuid4, primary_key=True)
shortname: str
name: Optional[str] = None
description: Optional[str] = None
@register_rest_rootpoint
class RootApp(RestResourceBase):
testValueRoot: float = 3.14
info: Info = Info(version="0.0.1", api_version="0.0.2")
games: dict[UUID, Game] = {
UUID("9b0381d4-65f6-11ee-8c99-0242ac120002"): Game(
uuid="9b0381d4-65f6-11ee-8c99-0242ac120002",
shortname="testGame Origin",
description="test Game Desc Origin",
patchs={Patch_1.uuid: Patch_1},
profiles={
UUID("aee1e870-65fa-11ee-8c99-0242ac120002"): Profile(
uuid="aee1e870-65fa-11ee-8c99-0242ac120002",
shortname="testprofile",
)
},
)
}
patchs: dict[UUID, Patch] = {Patch_1.uuid: Patch_1, Patch_2.uuid: Patch_2}
users: dict[UUID, User] = {User1.uuid: User1}
patchs2: dict[UUID, Patch2] = {}
# this add the classes to globals to allow using them later on
# => this is only for uinit-testing purpose and is not needed in real use
globals()[Info.__name__] = Info
globals()[Game.__name__] = Game
globals()[User.__name__] = User
globals()[Profile.__name__] = Profile
globals()[Patch.__name__] = Patch
globals()[Patch2.__name__] = Patch2
globals()[RootApp.__name__] = RootApp
def find_free_port():
with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as s:
s.bind(("", 0))
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
hostname = socket.gethostname()
IPAddr = socket.gethostbyname(hostname)
return "localhost", s.getsockname()[1]
def launch_server(ip, port):
print(f"port2={port}")
init_classes()
uvicorn.run(f"{__loader__.name}:RootApp", port=port, host="0.0.0.0", log_level="warning", factory=True)
class Test_RestAPI_WebServer(unittest.TestCase):
def setUp(self) -> None:
chdir(testdir_path.parent.resolve())
def test_nomal_AllCmd_games(self):
ip, port = find_free_port()
print(f"ip1={ip}")
print(f"port1={port}")
proc = Process(
target=launch_server,
args=(
ip,
port,
),
)
proc.start()
sleep(1)
s = requests.Session()
try:
# Fetching games
response = s.get(f"http://{ip}:{port}/games")
self.assertEqual(response.status_code, 200)
data = response.json()
self.assertIsInstance(data, list)
self.assertEqual(
data,
["9b0381d4-65f6-11ee-8c99-0242ac120002"],
)
# Login in
"""
response = s.post(
f"http://{ip}:{port}/login",
params={"username": "test", "password": "test"},
)
self.assertEqual(response.status_code, 200)
"""
# Add a new one (with all values setted)
response = s.post(
f"http://{ip}:{port}/games",
json={
"shortname": "test",
"name": "nametest",
"description": "test Game Desc",
},
)
self.assertEqual(response.status_code, 201)
data = response.json()
NEW_GAME_UUID = UUID(data)
# Fetching games again
response = s.get(f"http://{ip}:{port}/games")
self.assertEqual(response.status_code, 200)
data = response.json()
self.assertIsInstance(data, list)
self.assertEqual(
data,
["9b0381d4-65f6-11ee-8c99-0242ac120002", str(NEW_GAME_UUID)],
)
# Getting accurate values of created element
response = s.get(f"http://{ip}:{port}/games/{str(NEW_GAME_UUID)}")
self.assertEqual(response.status_code, 200)
data = response.json()
self.assertIsInstance(data, dict)
self.assertIsInstance(data, dict)
self.assertIn("shortname", data)
self.assertIn("name", data)
self.assertIn("description", data)
self.assertIn("uuid", data)
NEW_GAME_UUID = UUID(data["uuid"])
del data["uuid"]
self.assertDictEqual(
data,
{
"name": "nametest",
"shortname": "test",
"description": "test Game Desc",
},
)
# removing the new one
response = s.delete(f"http://{ip}:{port}/games/{str(NEW_GAME_UUID)}")
self.assertEqual(response.status_code, 200)
# Fetching games again
response = s.get(f"http://{ip}:{port}/games")
self.assertEqual(response.status_code, 200)
data = response.json()
self.assertEqual(
data,
["9b0381d4-65f6-11ee-8c99-0242ac120002"],
)
# Getting accurate values
response = s.get(f"http://{ip}:{port}/games/9b0381d4-65f6-11ee-8c99-0242ac120002")
self.assertEqual(response.status_code, 200)
data = response.json()
self.assertIsInstance(data, dict)
self.assertIsInstance(data, dict)
self.assertIn("shortname", data)
self.assertIn("name", data)
self.assertIn("description", data)
self.assertIn("uuid", data)
NEW_GAME_UUID = UUID(data["uuid"])
del data["uuid"]
self.assertDictEqual(
data,
{
"name": None,
"shortname": "testGame Origin",
"description": "test Game Desc Origin",
},
)
# Update values
response = s.put(
f"http://{ip}:{port}/games/9b0381d4-65f6-11ee-8c99-0242ac120002",
json={
"name": "MyName",
},
)
self.assertEqual(response.status_code, 201)
# Getting accurate values
response = s.get(f"http://{ip}:{port}/games/9b0381d4-65f6-11ee-8c99-0242ac120002")
self.assertEqual(response.status_code, 200)
data = response.json()
self.assertIsInstance(data, dict)
self.assertIsInstance(data, dict)
self.assertIn("shortname", data)
self.assertIn("name", data)
self.assertIn("description", data)
self.assertIn("uuid", data)
NEW_GAME_UUID = UUID(data["uuid"])
del data["uuid"]
self.assertDictEqual(
data,
{
"name": "MyName",
"shortname": "testGame Origin",
"description": "test Game Desc Origin",
},
)
# removing original element
response = s.delete(f"http://{ip}:{port}/games?API_key={str(NEW_GAME_UUID)}")
self.assertEqual(response.status_code, 200)
# Fetching games again
response = s.get(f"http://{ip}:{port}/games")
self.assertEqual(response.status_code, 200)
data = response.json()
self.assertTrue(len(data) == 0)
finally:
proc.terminate()
s.close()
@unittest.skip
def test_perf_dict(self):
print(f"SOCKET PERF TEST")
n_loop = 10000
ip, port = find_free_port()
print(f"ip1={ip}")
print(f"port1={port}")
proc = Process(
target=launch_server,
args=(
ip,
port,
),
)
proc.start()
sleep(1)
s = requests.Session()
try:
start = time()
for _ in range(n_loop):
s.get(f"http://{ip}:{port}/users/8da57a3c-661f-11ee-8c99-0242ac120002")
end = time()
print(f"GET 1st level dict: {int(n_loop/(end-start))} Req/s")
start = time()
for _ in range(n_loop):
newUUID = uuid4()
s.post(
f"http://{ip}:{port}/users?API_key={newUUID}",
json={"name": "testUser", "secret": "test"},
)
end = time()
print(f"POST 1st level dict (API_key): {int(n_loop/(end-start))} Req/s")
start = time()
for _ in range(n_loop):
newUUID = uuid4()
s.post(
f"http://{ip}:{port}/users?API_key={str(newUUID)}",
json={"name": "testUser", "secret": "test"},
)
s.get(f"http://{ip}:{port}/users/{newUUID}")
end = time()
print(f"POST/GET 1st level dict (API_key): {int(n_loop/(end-start))} Req/s")
start = time()
for _ in range(n_loop):
response = s.post(f"http://{ip}:{port}/users", '{"name": "testUser", "secret": "test"}')
s.get(f"http://{ip}:{port}/users/{response.json()}")
end = time()
print(f"POST/GET 1st level dict (autokey): {int(n_loop/(end-start))} Req/s")
start = time()
for _ in range(n_loop):
s.put(
f"http://{ip}:{port}/games/9b0381d4-65f6-11ee-8c99-0242ac120002/shortname",
json="TestValue!!",
)
s.get(f"http://{ip}:{port}/games/9b0381d4-65f6-11ee-8c99-0242ac120002/shortname")
end = time()
print(f"PUT/GET 1st level (value) dict: {int(n_loop/(end-start))} Req/s")
start = time()
for _ in range(n_loop):
s.get(f"http://{ip}:{port}/games/9b0381d4-65f6-11ee-8c99-0242ac120002/patchs/cee1e870-65fa-11ee-8c99-0242ac120002")
end = time()
print(f"GET 2nd level dict: {int(n_loop/(end-start))} Req/s")
start = time()
for _ in range(n_loop):
s.get(
f"http://{ip}:{port}/games/9b0381d4-65f6-11ee-8c99-0242ac120002/patchs/cee1e870-65fa-11ee-8c99-0242ac120002/shortname",
)
end = time()
print(f"GET 2nd level (value) dict: {int(n_loop/(end-start))} Req/s")
start = time()
for _ in range(n_loop):
s.put(
f"http://{ip}:{port}/games/9b0381d4-65f6-11ee-8c99-0242ac120002/patchs/cee1e870-65fa-11ee-8c99-0242ac120002/shortname",
json="TestValue!!",
)
s.get(
f"http://{ip}:{port}/games/9b0381d4-65f6-11ee-8c99-0242ac120002/patchs/cee1e870-65fa-11ee-8c99-0242ac120002/shortname",
)
end = time()
print(f"PUT/GET 2nd level (value) dict: {int(n_loop/(end-start))} Req/s")
finally:
proc.terminate()
s.close()

35
test/test_test_module.py
View File

@@ -1,35 +0,0 @@
# pyrestresource (c) by chacha
#
# pyrestresource is licensed under a
# Creative Commons Attribution-NonCommercial-ShareAlike 4.0 International Unported License.
#
# You should have received a copy of the license along with this
# work. If not, see <https://creativecommons.org/licenses/by-nc-sa/4.0/>.
import unittest
from os import chdir
from io import StringIO
from contextlib import redirect_stdout,redirect_stderr
from pathlib import Path
print(__name__)
print(__package__)
from src import pyrestresource
testdir_path = Path(__file__).parent.resolve()
chdir(testdir_path.parent.resolve())
class Testtest_module(unittest.TestCase):
def setUp(self) -> None:
chdir(testdir_path.parent.resolve())
def test_version(self):
self.assertNotEqual(pyrestresource.__version__,"?.?.?")
def test_test_module(self):
with redirect_stdout(StringIO()) as capted_stdout, redirect_stderr(StringIO()) as capted_stderr:
self.assertEqual(pyrestresource.test_function(41),42)
self.assertEqual(len(capted_stderr.getvalue()),0)
self.assertEqual(capted_stdout.getvalue().strip(),"Hello world !")