forked from turner-townsend/flask-pydantic-spec
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathclass_view.py
121 lines (93 loc) · 3.08 KB
/
class_view.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
from flask import Flask, request, jsonify
from pydantic import BaseModel, Field, Extra
from flask_pydantic_spec import FlaskPydanticSpec, Response, Request
import time
from flask.views import MethodView
class ExampleQueryGetRequest(BaseModel):
text: str = Field(default="default query strings", description="blabla")
class Config:
extra = Extra.forbid
class ExampleQueryPostRequest(BaseModel):
query_text: str = Field(default="default query strings", description="post req blabla")
class Config:
extra = Extra.forbid
class ExampleQueryGetResponse(BaseModel):
my_get_time: int = Field(description="current time")
class Config:
extra = Extra.forbid
class ExamplePathResponse(BaseModel):
id: str = Field(description="id")
current_time: int = Field(description="current time")
query_text: str = Field(default="", description="query text")
class Config:
extra = Extra.forbid
app = Flask(__name__)
api = FlaskPydanticSpec("flask")
class TestQueryView(MethodView):
"""
Test view class
"""
@api.validate(
query=ExampleQueryGetRequest,
resp=Response(HTTP_200=ExampleQueryGetResponse),
tags=["test query"],
)
def get(self):
"""
Test query get view
:return:
"""
return jsonify({"my_get_time": int(time.time())})
@api.validate(query=ExampleQueryPostRequest, tags=["test query"])
def post(self):
"""
Test query post view
:return:
"""
return jsonify(
{"post_time": int(time.time()), "post_text": request.context.query.query_text}
)
class TestPathView(MethodView):
@api.validate(
query=ExampleQueryPostRequest,
resp=Response(HTTP_200=ExamplePathResponse),
tags=["test path"],
)
def post(self, id):
"""
Test path post view
:return:
"""
print(f"query: {request.context.query}")
return jsonify(
ExamplePathResponse(
id=id, current_time=int(time.time()), query_text=request.args["text"]
).dict()
)
class TestBodyView(MethodView):
@api.validate(
body=Request(ExampleQueryPostRequest),
resp=Response(HTTP_200=ExamplePathResponse),
tags=["test body"],
)
def post(self, id):
"""
Test body post view
:return:
"""
return jsonify({"id": id, "current_time": int(time.time())})
if __name__ == "__main__":
api.register(app) # if you don't register in api init step
app.add_url_rule(
"/api/testquery", methods=["GET", "POST"], view_func=TestQueryView.as_view("TestQueryView"),
)
api.register_class_view_apidoc(TestQueryView)
app.add_url_rule(
"/api/testpath/<id>", methods=["POST"], view_func=TestPathView.as_view("TestPathView"),
)
api.register_class_view_apidoc(TestPathView)
app.add_url_rule(
"/api/testbody/<id>", methods=["POST"], view_func=TestBodyView.as_view("TestBodyView"),
)
api.register_class_view_apidoc(TestBodyView)
app.run(port=8000)