-
Notifications
You must be signed in to change notification settings - Fork 0
fix(typing): Typing fixes #18
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
base: main
Are you sure you want to change the base?
Changes from all commits
5c09a9f
eae7ad4
f70bbe7
170da2c
37ccdcd
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -1,5 +1,9 @@ | ||
from __future__ import annotations | ||
from typing import Union, Optional | ||
|
||
from typing import Optional, Any | ||
from typing_extensions import Literal | ||
|
||
|
||
import importlib | ||
import os | ||
import pathlib | ||
|
@@ -44,8 +48,8 @@ def __init__(self, | |
summary: str = None, | ||
# description: str = "Description", | ||
# desc: str = "Description", | ||
types: Union[list, str] = "application/octet-stream", | ||
example: Union[dict, str, int, float, bool, any] = None, | ||
types: list | str = "application/octet-stream", | ||
example: Any = None, | ||
security: Optional[dict] = None, | ||
responses: Optional[list] = None, | ||
tags: Optional[list] = None, | ||
|
@@ -66,10 +70,10 @@ def __init__(self, | |
|
||
def http(method: str, | ||
require_auth: bool = True, | ||
args: Union[tuple, list, Argument] = (), | ||
args: tuple | list | Argument = (), | ||
docs: Optional[Document] = None): | ||
def _context(handler): | ||
path = None | ||
path: Optional[str] = None | ||
file = handler.__globals__["__file__"] | ||
if "___" in os.path.normpath(file).split(os.path.sep): | ||
raise IsADirectoryError("Path-argument like directory found.") | ||
|
@@ -148,21 +152,27 @@ def __init__(self, document: Optional[Document] = None): | |
class Undefined: pass | ||
|
||
|
||
ArgumentTypes = Literal["str", "string", "bool", "boolean", "number", "int", | ||
"long", "double", "decimal", "float", "other"] | ||
|
||
|
||
class Argument(Documented): | ||
type: ArgumentTypes | ||
|
||
def __init__(self, | ||
name: str, | ||
arg_type: str, | ||
arg_type: ArgumentTypes, | ||
arg_in: str, | ||
required: bool = True, | ||
auto_cast: bool = True, | ||
minimum: int = -1, | ||
maximum: int = -1, | ||
must_be: Union[tuple, list] = (), | ||
must_be: tuple | list = (), | ||
doc: Optional[Document] = None, | ||
format_type: Optional[str] = None, | ||
ignore_check_expect100: bool = False, | ||
enum: Union[tuple, list] = (), | ||
default: Optional[any] = Undefined): | ||
enum: tuple | list = (), | ||
Potato1682 marked this conversation as resolved.
Show resolved
Hide resolved
|
||
default: Any = Undefined): | ||
super().__init__(doc) | ||
if arg_type not in ["str", "string", "bool", "boolean", "number", "int", "long", | ||
"double", "decimal", "float", "other"]: | ||
|
@@ -182,14 +192,14 @@ def __init__(self, | |
self.ignore_check_expect100 = ignore_check_expect100 | ||
self.default = default | ||
|
||
def norm_type(self, val: Optional[any] = None) -> Optional[any]: | ||
def norm_type(self, val: Any = None) -> Any: | ||
if "str" in self.type: | ||
return "string" if val is None else str(val) | ||
elif "bool" in self.type: | ||
return "boolean" if val is None else bool(val) | ||
elif self.type is "number" or "int" in self.type: | ||
elif self.type == "number" or "int" in self.type: | ||
return "integer" if val is None else int(val) | ||
elif self.type is "long": | ||
elif self.type == "long": | ||
return "integer" if val is None else int(val) | ||
else: | ||
return "number" if val is None else float(val) | ||
|
@@ -217,20 +227,20 @@ def validate(self, param_dict: dict) -> int: | |
value = param_dict[name] | ||
|
||
if "str" in typ: | ||
if len(must_be) is not 0 and value not in must_be: | ||
if len(must_be) != 0 and value not in must_be: | ||
return 1 | ||
|
||
if min_val is not -1 and len(value) < min_val: | ||
if min_val != -1 and len(value) < min_val: | ||
return 3 | ||
|
||
if max_val is not -1 and len(value) > max_val: | ||
if max_val != -1 and len(value) > max_val: | ||
return 4 | ||
|
||
if cast: | ||
param_dict[name] = str(value) | ||
|
||
elif "bool" in typ: | ||
if value not in ("true", "false") + self.must_be: | ||
if value not in ("true", "false") + tuple(self.must_be): | ||
return 1 | ||
|
||
if cast: | ||
|
@@ -246,13 +256,13 @@ def validate(self, param_dict: dict) -> int: | |
except ValueError: | ||
return 2 | ||
|
||
if len(must_be) is not 0 and val not in must_be: | ||
if len(must_be) != 0 and val not in must_be: | ||
return 1 | ||
|
||
if min_val is not -1 and val < min_val: | ||
if min_val != -1 and val < min_val: | ||
return 3 | ||
|
||
if max_val is not -1 and val > max_val: | ||
if max_val != -1 and val > max_val: | ||
return 4 | ||
|
||
if cast: | ||
|
@@ -280,7 +290,7 @@ def __init__(self, | |
self.args = () if args is None else args | ||
self.path_arg = path_arg | ||
|
||
def handle(self, handler, params: dict, queries: dict, path_param: dict) -> Union[Response, any]: | ||
def handle(self, handler, params: dict, queries: dict, path_param: dict) -> Any: | ||
if self.auth_required and handler.do_auth(): | ||
return | ||
|
||
|
@@ -309,7 +319,7 @@ def validate_arg(self, handler, params: dict, queries: dict, path_param: dict) - | |
continue | ||
elif code == 1: | ||
if "bool" in arg.type: | ||
quick_invalid(handler, arg.name, "[" + ", ".join(("true", "false") + arg.must_be) + "]") | ||
quick_invalid(handler, arg.name, "[" + ", ".join(("true", "false") + tuple(arg.must_be)) + "]") | ||
return False | ||
else: | ||
quick_invalid(handler, arg.name, "[" + ", ".join(arg.must_be) + "]") | ||
|
@@ -339,7 +349,7 @@ def validate_arg(self, handler, params: dict, queries: dict, path_param: dict) - | |
val = arg.norm_type(path_param[arg.name]) if arg.auto_cast else path_param[arg.name] | ||
params[arg.name] = val | ||
|
||
if len(missing) is not 0: | ||
if len(missing) != 0: | ||
write(handler, 400, e(Cause.MISSING_FIELD, Cause.MISSING_FIELD[2] | ||
.replace("%0", str(len(missing))) | ||
.replace("%1", ", ".join(missing)))) | ||
|
@@ -350,9 +360,9 @@ def validate_arg(self, handler, params: dict, queries: dict, path_param: dict) - | |
class Response(Documented): | ||
def __init__(self, | ||
code: int = 0, | ||
body: Optional[any] = None, | ||
body: Any = None, | ||
raw_body: bool = False, | ||
content_type: Union[str, list] = None, | ||
content_type: str | list = None, | ||
headers: Optional[dict] = None, | ||
doc: Optional[Document] = None): | ||
super().__init__(doc) | ||
|
@@ -367,7 +377,7 @@ def header(self, name: str, value: str) -> Response: | |
self.headers[name] = value | ||
return self | ||
|
||
def body(self, value: any, raw: bool = False) -> Response: | ||
def body(self, value: Any, raw: bool = False) -> Response: | ||
self.body_data = value | ||
self.raw = raw | ||
return self | ||
|
@@ -390,13 +400,13 @@ def __init__(self, | |
cause: Optional[Cause] = None, | ||
code: int = 0, | ||
headers: Optional[dict] = None, | ||
body: Optional[any] = None, | ||
content_type: Optional[Union[str, list]] = None, | ||
body: Any = None, | ||
content_type: Optional[str | list] = None, | ||
doc: Optional[Document] = None): | ||
if cause is not None: | ||
super().__init__(cause[0], headers, cause[2], content_type, doc) | ||
super().__init__(cause[0], headers, cause[2], content_type, headers, doc) | ||
else: | ||
super().__init__(code, headers, body, content_type, doc) | ||
super().__init__(code, body, False, content_type, headers, doc) | ||
|
||
self.cause = cause | ||
|
||
|
@@ -415,6 +425,8 @@ def error(cause: Optional[Cause] = None, code: int = 0, message: Optional[str] = | |
|
||
|
||
class EPManager: | ||
known_source: list[str] | ||
Potato1682 marked this conversation as resolved.
Show resolved
Hide resolved
|
||
|
||
def __init__(self): | ||
global loader | ||
self.signals = [] | ||
|
@@ -501,7 +513,7 @@ def make_cache(self) -> None: | |
cursor[method] = EndPoint(method, rt, path, function, auth, args, bool(paths), docs) | ||
self.count += 1 | ||
|
||
def get_endpoint(self, method: str, path: str, params: Optional[dict] = None) -> Optional[EndPoint]: | ||
def get_endpoint(self, method: str, path: str, params: dict = {}) -> Optional[EndPoint]: | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. 引数のデフォルト値が mutable です。 |
||
|
||
cursor = self.index_tree | ||
|
||
|
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -1,3 +1,7 @@ | ||
from __future__ import annotations | ||
|
||
from typing import Optional | ||
|
||
import cgi | ||
import json | ||
import mimetypes | ||
|
@@ -16,6 +20,7 @@ | |
|
||
|
||
class Handler(ServerHandler): | ||
request: Optional[HTTPRequest] | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. requestがNoneの場合はパース時に切り捨てられます。 There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. アノテーション用。少なくともNoneが入る場合はNoneとのunionになります。 |
||
|
||
def __init__(self, request, client_address, server): | ||
self.logger = server.logger | ||
|
@@ -76,7 +81,10 @@ def call_handler(self, path: str, params, queries): | |
def dynamic_handle(self, path, params, queries): | ||
path_param = {} | ||
|
||
ep = endpoint.loader.get_endpoint(self.request.method, path, path_param) | ||
ep: Optional[endpoint.EndPoint] = None | ||
|
||
if self.request is not None and self.request.method is not None: | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
|
||
ep = endpoint.loader.get_endpoint(self.request.method, path, path_param) | ||
|
||
if ep is None: | ||
return False | ||
|
@@ -128,7 +136,10 @@ def _send_body(self, body, raw=False, content_types=None): | |
return | ||
|
||
default = self.config["system"]["request"]["default_content_type"] | ||
accept = self.request.headers["Accept"] if "Accept" in self.request.headers else "" | ||
accept = "" | ||
|
||
if self.request is not None: | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
|
||
self.request.headers["Accept"] if self.request.headers is not None and "Accept" in self.request.headers else "" | ||
Potato1682 marked this conversation as resolved.
Show resolved
Hide resolved
|
||
|
||
if content_types is not None: | ||
if isinstance(content_types, str): | ||
|
@@ -149,18 +160,24 @@ def log_request(self, **kwargs): | |
if not no_req_log: | ||
self.logger.info(get_log_name(), '%s -- %s %s -- "%s %s"' % | ||
(kwargs["client"], kwargs["code"], "" if kwargs["message"] is None else kwargs["message"], | ||
self.request.method, kwargs["path"])) | ||
self.request.method if self.request is not None else "<no method>", kwargs["path"])) | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
|
||
|
||
def handle_switch(self): | ||
try: | ||
if self.request is None: | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
|
||
raise TypeError("Request instance is None") | ||
|
||
if self.request.path is None: | ||
raise TypeError("Request path is None") | ||
|
||
path = parse.urlparse(self.request.path) | ||
queries = dict(parse.parse_qsl(path.query)) | ||
|
||
if self.request.method in ["GET", "HEAD", "TRACE", "OPTIONS"]: | ||
|
||
self.call_handler(path.path, {}, queries) | ||
else: | ||
if "Content-Type" in self.request.headers: | ||
if self.request.headers is not None and "Content-Type" in self.request.headers: | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
|
||
content_len = int(self.request.headers.get("content-length").value) | ||
content_type = str(self.request.headers["Content-Type"]) | ||
|
||
|
@@ -196,7 +213,6 @@ def handle_switch(self): | |
self.logger.warn(get_log_name(), get_stack_trace("server", *sys.exc_info())) | ||
|
||
def do_auth(self): | ||
self.request: HTTPRequest | ||
if "Authorization" not in self.request.headers: | ||
route.post_error(self, route.Cause.AUTH_REQUIRED) | ||
|
||
|
Uh oh!
There was an error while loading. Please reload this page.