ClassFilterDone
Some checks failed
K8S Fission Deployment / Deployment fission functions (push) Failing after 11s
Some checks failed
K8S Fission Deployment / Deployment fission functions (push) Failing after 11s
This commit is contained in:
@@ -8,9 +8,9 @@ def main():
|
||||
"""
|
||||
```fission
|
||||
{
|
||||
"name": "phone-admin-get-insert-delete",
|
||||
"name": "phonenumbers-admin-get-insert-delete",
|
||||
"http_triggers": {
|
||||
"phone-admin-get-insert-delete-http": {
|
||||
"phonenumbers-admin-get-insert-delete-http": {
|
||||
"url": "/ailbl/admin/users/{UserId}/phones",
|
||||
"methods": [ "POST", "DELETE", "GET"]
|
||||
}
|
||||
|
||||
@@ -8,9 +8,9 @@ def main():
|
||||
"""
|
||||
```fission
|
||||
{
|
||||
"name": "phone-users-get-insert-delete",
|
||||
"name": "phonenumbers-users-get-insert-delete",
|
||||
"http_triggers": {
|
||||
"phone-users-get-insert-delete-http": {
|
||||
"phonenumbers-users-get-insert-delete-http": {
|
||||
"url": "/ailbl/users/phones",
|
||||
"methods": ["POST", "DELETE", "GET"]
|
||||
}
|
||||
|
||||
54
apps/filters.py
Normal file
54
apps/filters.py
Normal file
@@ -0,0 +1,54 @@
|
||||
import dataclasses
|
||||
from typing import Optional
|
||||
from flask import request
|
||||
from helpers import str_to_bool
|
||||
|
||||
|
||||
@dataclasses.dataclass
|
||||
class PhoneFilter:
|
||||
phone_numbers: Optional[str] = None
|
||||
prefix: Optional[str] = None
|
||||
created_from: Optional[str] = None
|
||||
created_to: Optional[str] = None
|
||||
modified_from: Optional[str] = None
|
||||
modified_to: Optional[str] = None
|
||||
|
||||
@classmethod
|
||||
def from_request_queries(cls) -> "PhoneFilter":
|
||||
return cls(
|
||||
phone_number=request.args.get("filter[phone_number]"),
|
||||
prefix=request.args.get("filter[prefix]"),
|
||||
created_from=request.args.get("filter[created_from]"),
|
||||
created_to=request.args.get("filter[created_to]"),
|
||||
modified_from=request.args.get("filter[modified_from]"),
|
||||
modified_to=request.args.get("filter[modified_to]"),
|
||||
)
|
||||
|
||||
|
||||
@dataclasses.dataclass
|
||||
class Page:
|
||||
page: int = 0
|
||||
size: int = 10
|
||||
asc: bool = False
|
||||
|
||||
@classmethod
|
||||
def from_request_queries(cls) -> "Page":
|
||||
return Page(
|
||||
page=int(request.args.get("page", 0)),
|
||||
size=int(request.args.get("size", 10)),
|
||||
asc=request.args.get("asc", type=str_to_bool) or False
|
||||
)
|
||||
|
||||
|
||||
@dataclasses.dataclass
|
||||
class PhonePage(Page):
|
||||
sortby: Optional[str] = None
|
||||
filter: PhoneFilter = dataclasses.field(
|
||||
default_factory=PhoneFilter.from_request_queries
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def from_request_queries(cls) -> "PhonePage":
|
||||
# return super().from_request_queries()
|
||||
base = Page.from_request_queries()
|
||||
return cls(**dataclasses.asdict(base), sortby=request.args.get("sortby"))
|
||||
@@ -94,3 +94,13 @@ def check_port_open(ip: str, port: int, timeout: int = 30):
|
||||
except Exception as err:
|
||||
current_app.logger.err(f"Check port open error: {err}")
|
||||
return False
|
||||
|
||||
def str_to_bool(value: str | None) -> typing.Optional[bool]:
|
||||
if value is None:
|
||||
return None
|
||||
val = value.strip().lower()
|
||||
if val in ("true", "1", "yes"):
|
||||
return True
|
||||
if val in ("false", "0", "no"):
|
||||
return False
|
||||
return None
|
||||
Reference in New Issue
Block a user