Skip to content

Commit f6471a1

Browse files
DinisCruzDinisCruz
committed
added major feature to add_route_post where it now supports Type_Safe classes :)
1 parent 6c24ec8 commit f6471a1

3 files changed

Lines changed: 369 additions & 3 deletions

File tree

‎.github/workflows/ci-pipeline__dev.yml‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -40,6 +40,7 @@ jobs:
4040
- run-unit-tests
4141

4242
publish-to-pypi:
43+
if: False
4344
permissions:
4445
id-token: write
4546
name: "Publish to PYPI"

‎osbot_fast_api/api/Fast_API_Routes.py‎

Lines changed: 42 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,8 +1,9 @@
1+
import inspect
2+
from typing import get_type_hints
13
from fastapi import APIRouter, FastAPI
24
from osbot_utils.type_safe.Type_Safe import Type_Safe
35
from osbot_utils.decorators.lists.index_by import index_by
46

5-
67
class Fast_API_Routes(Type_Safe): # refactor to Fast_API__Routes
78
router : APIRouter
89
app : FastAPI = None
@@ -27,8 +28,46 @@ def add_route_delete(self, function):
2728
def add_route_get(self, function):
2829
return self.add_route(function=function, methods=['GET'])
2930

30-
def add_route_post(self, function):
31-
return self.add_route(function=function, methods=['POST'])
31+
# def add_route_post(self, function):
32+
# return self.add_route(function=function, methods=['POST'])
33+
34+
def add_route_post(self, function): # add post with support for Type_Safe objects
35+
sig = inspect.signature(function) # Check if function has a Type_Safe parameter
36+
type_hints = get_type_hints(function)
37+
38+
type_safe_param = False
39+
for param_name, param in sig.parameters.items():
40+
if param_name == 'self':
41+
continue
42+
param_type = type_hints.get(param_name)
43+
if param_type and inspect.isclass(param_type):
44+
if issubclass(param_type, Type_Safe):
45+
type_safe_param = True
46+
break
47+
48+
if type_safe_param:
49+
def wrapper(data: dict):
50+
param_object = param_type.from_json(data)
51+
kwargs = { param_name: param_object }
52+
result = function(**kwargs)
53+
if isinstance(result, Type_Safe):
54+
return result.json()
55+
return result
56+
57+
58+
# todo: the code below is not working (need to add support for supporting Type_Safe return values)
59+
# Remove the return type annotation to prevent FastAPI validation
60+
# wrapper.__annotations__ = function.__annotations__.copy()
61+
# if 'return' in wrapper.__annotations__:
62+
# del wrapper.__annotations__['return'] # Remove return type so FastAPI doesn't validate
63+
64+
65+
path = '/' + function.__name__.replace('_', '-')
66+
self.router.add_api_route(path=path, endpoint=wrapper, methods=['POST'])
67+
return self
68+
else:
69+
# Normal route
70+
return self.add_route(function=function, methods=['POST'])
3271

3372
def add_route_put(self, function):
3473
return self.add_route(function=function, methods=['PUT'])

‎tests/unit/api/test_Fast_API_Routes__with_Type_Safe.py‎

Lines changed: 326 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,326 @@
1+
from unittest import TestCase
2+
from dataclasses import dataclass
3+
from typing import Optional, List
4+
from osbot_utils.type_safe.Type_Safe import Type_Safe
5+
from osbot_utils.utils.Objects import __
6+
from pydantic import BaseModel
7+
from osbot_fast_api.api.Fast_API_Routes import Fast_API_Routes
8+
from osbot_fast_api.api.Fast_API import Fast_API
9+
10+
11+
class test_Fast_API_Routes__with_Type_Safe(TestCase):
12+
13+
def test__1__current_get_support(self):
14+
15+
class GET_Routes(Fast_API_Routes):
16+
tag = 'get'
17+
18+
def ping(self):
19+
return 'pong'
20+
21+
def setup_routes(self):
22+
self.add_route_get(self.ping)
23+
24+
class An_Fast_API(Fast_API):
25+
default_routes = False
26+
def setup_routes(self):
27+
self.add_routes(GET_Routes)
28+
29+
an_fast_api = An_Fast_API().setup()
30+
assert an_fast_api.routes_paths() == ['/get/ping']
31+
assert an_fast_api.client().get('/get/ping').json() == 'pong'
32+
33+
def test__2__current_post_support__json(self):
34+
35+
class POST_Routes(Fast_API_Routes):
36+
tag = 'post'
37+
38+
def post_data(self, data: dict):
39+
return data
40+
41+
def setup_routes(self):
42+
self.add_route_post(self.post_data)
43+
44+
class An_Fast_API(Fast_API):
45+
default_routes = False
46+
def setup_routes(self):
47+
self.add_routes(POST_Routes)
48+
49+
an_fast_api = An_Fast_API().setup()
50+
assert an_fast_api.routes_paths() == ['/post/post-data']
51+
52+
53+
post_data = {'answer': 42}
54+
response = an_fast_api.client().post('/post/post-data', json=post_data)
55+
assert response.status_code == 200
56+
assert response.json() == post_data
57+
58+
59+
def test__3__with_dataclass_type_safety(self): # Using Python dataclasses for request/response type safety
60+
61+
@dataclass
62+
class UserRequest: # Request model using dataclass"""
63+
username: str
64+
email : str
65+
age : Optional[int] = None
66+
tags : List [str] = None
67+
68+
def __post_init__(self):
69+
if self.tags is None:
70+
self.tags = []
71+
72+
@dataclass
73+
class UserResponse: # Response model using dataclass"""
74+
id : int
75+
username: str
76+
email : str
77+
age : Optional[int]
78+
tags : List[str]
79+
status : str = "active"
80+
81+
class User_Routes(Fast_API_Routes):
82+
tag = 'users'
83+
84+
def create_user(self, user: UserRequest): # Create a new user with dataclass validation
85+
user_response = UserResponse(id = 123 , # Simulate user creation with ID assignment
86+
username = user.username,
87+
email = user.email ,
88+
age = user.age ,
89+
tags = user.tags ,
90+
status = "active" )
91+
return user_response
92+
93+
def get_user(self, user_id: int) -> UserResponse: # Get user by ID with typed response"""
94+
95+
return UserResponse(id = user_id , # Simulate fetching user
96+
username = "test_user" ,
97+
email = "test@example.com" ,
98+
age = 25 ,
99+
tags = ["developer", "python"])
100+
101+
def setup_routes(self):
102+
self.add_route_post(self.create_user)
103+
self.add_route_get(self.get_user)
104+
105+
class An_Fast_API(Fast_API):
106+
default_routes = False
107+
def setup_routes(self):
108+
self.add_routes(User_Routes)
109+
110+
# Test the implementation
111+
an_fast_api = An_Fast_API().setup()
112+
assert an_fast_api.routes_paths() == ['/users/create-user', '/users/get-user']
113+
114+
# Test POST with dataclass
115+
user_data = { 'username': 'john_doe' ,
116+
'email' : 'john@example.com' ,
117+
'age' : 30 ,
118+
'tags' : ['admin', 'moderator']}
119+
response = an_fast_api.client().post('/users/create-user', json=user_data)
120+
assert response.status_code == 200
121+
result = response.json()
122+
assert result['username'] == 'john_doe'
123+
assert result['email' ] == 'john@example.com'
124+
assert result['id' ] == 123
125+
assert result['status' ] == 'active'
126+
127+
# Test GET with typed response
128+
response = an_fast_api.client().get('/users/get-user', params={'user_id': 456})
129+
assert response.status_code == 200
130+
result = response.json()
131+
assert result['id' ] == 456
132+
assert result['username'] == 'test_user'
133+
134+
135+
def test__4__with_pydantic_type_safety(self): # Using Pydantic for request/response type safety
136+
137+
class UserRequest(BaseModel): # Request model using Pydantic
138+
username: str
139+
email : str
140+
age : Optional[int] = None
141+
tags : List [str] = None
142+
143+
def __init__(self, **data):
144+
super().__init__(**data)
145+
if self.tags is None:
146+
self.tags = []
147+
148+
class UserResponse(BaseModel): # Response model using Pydantic
149+
id : int
150+
username: str
151+
email : str
152+
age : Optional[int]
153+
tags : List[str]
154+
status : str = "active"
155+
156+
class User_Routes(Fast_API_Routes):
157+
tag = 'users'
158+
159+
def create_user(self, user: UserRequest): # Create a new user with Pydantic validation
160+
user_response = UserResponse(id = 123 , # Simulate user creation with ID assignment
161+
username = user.username,
162+
email = user.email ,
163+
age = user.age ,
164+
tags = user.tags ,
165+
status = "active" )
166+
return user_response
167+
168+
def get_user(self, user_id: int) -> UserResponse: # Get user by ID with typed response"""
169+
170+
return UserResponse(id = user_id , # Simulate fetching user
171+
username = "test_user" ,
172+
email = "test@example.com" ,
173+
age = 25 ,
174+
tags = ["developer", "python"])
175+
176+
def setup_routes(self):
177+
self.add_route_post(self.create_user)
178+
self.add_route_get(self.get_user)
179+
180+
class An_Fast_API(Fast_API):
181+
default_routes = False
182+
def setup_routes(self):
183+
self.add_routes(User_Routes)
184+
185+
# Test the implementation
186+
an_fast_api = An_Fast_API().setup()
187+
assert an_fast_api.routes_paths() == ['/users/create-user', '/users/get-user']
188+
189+
# Test POST with dataclass
190+
user_data = { 'username': 'john_doe' ,
191+
'email' : 'john@example.com' ,
192+
'age' : 30 ,
193+
'tags' : ['admin', 'moderator']}
194+
response = an_fast_api.client().post('/users/create-user', json=user_data)
195+
assert response.status_code == 200
196+
result = response.json()
197+
assert result['username'] == 'john_doe'
198+
assert result['email' ] == 'john@example.com'
199+
assert result['id' ] == 123
200+
assert result['status' ] == 'active'
201+
202+
# Test GET with typed response
203+
response = an_fast_api.client().get('/users/get-user', params={'user_id': 456})
204+
assert response.status_code == 200
205+
result = response.json()
206+
assert result['id' ] == 456
207+
assert result['username'] == 'test_user'
208+
209+
def test__5__with_Type_Safe__indirect_support(self): # Using Pydantic for request/response type safety
210+
211+
class UserRequest(Type_Safe): # Request model using Type_Safe
212+
username: str
213+
email : str
214+
age : Optional[int] = None
215+
tags : List [str]
216+
217+
class UserResponse(Type_Safe): # Response model using Type_Safe
218+
id : int
219+
username: str
220+
email : str
221+
age : Optional[int]
222+
tags : List[str]
223+
status : str = "active"
224+
225+
class User_Routes(Fast_API_Routes):
226+
tag = 'users'
227+
228+
def create_user(self, data: dict): # Create a new user
229+
user = UserRequest.from_json(data)
230+
user_response = UserResponse(id = 123 , # Simulate user creation with ID assignment
231+
username = user.username,
232+
email = user.email ,
233+
age = user.age ,
234+
tags = user.tags ,
235+
status = "active" )
236+
return user_response.json()
237+
238+
def get_user(self, user_id: int) -> dict:
239+
240+
return UserResponse(id = user_id , # Simulate fetching user
241+
username = "test_user" ,
242+
email = "test@example.com" ,
243+
age = 25 ,
244+
tags = ["developer", "python"]).json()
245+
246+
def setup_routes(self):
247+
self.add_route_post(self.create_user)
248+
self.add_route_get (self.get_user)
249+
250+
class An_Fast_API(Fast_API):
251+
default_routes = False
252+
def setup_routes(self):
253+
self.add_routes(User_Routes)
254+
255+
# Test the implementation
256+
an_fast_api = An_Fast_API().setup()
257+
assert an_fast_api.routes_paths() == ['/users/create-user', '/users/get-user']
258+
259+
# Test POST
260+
user_data = { 'username': 'john_doe' ,
261+
'email' : 'john@example.com' ,
262+
'age' : 30 ,
263+
'tags' : ['admin', 'moderator']}
264+
response_1 = an_fast_api.client().post('/users/create-user', json=user_data)
265+
user_response_1 = UserResponse.from_json(response_1.json())
266+
267+
assert response_1.status_code == 200
268+
assert user_response_1.obj() == __( status = 'active' ,
269+
id = 123 ,
270+
username = 'john_doe' ,
271+
email = 'john@example.com' ,
272+
age = 30 ,
273+
tags = ['admin', 'moderator'])
274+
275+
# Test GET with typed response
276+
response_2 = an_fast_api.client().get('/users/get-user', params={'user_id': 456})
277+
user_response_2 = UserResponse.from_json(response_1.json())
278+
279+
assert response_2.status_code == 200
280+
assert user_response_2.json() == { 'age' : 30 ,
281+
'email' : 'john@example.com' ,
282+
'id' : 123 ,
283+
'status' : 'active' ,
284+
'tags' : ['admin', 'moderator'],
285+
'username': 'john_doe' }
286+
287+
288+
def test__6__with_Type_Safe__direct_support(self): # todo: need to add support for supporting Type_Safe return values
289+
290+
class An_Class(Type_Safe):
291+
an_str : str
292+
293+
class Obj_Routes(Fast_API_Routes):
294+
tag = 'obj'
295+
def create_object(self, an_class: An_Class): # receive An_Class object
296+
return True
297+
298+
def update_object(self, an_class: An_Class) : # receive and return An_Class object
299+
return an_class
300+
301+
def return_object(self, an_str: str): # just return An_Class object
302+
return An_Class(an_str=an_str)
303+
304+
def setup_routes(self):
305+
self.add_route_post(self.create_object) # this is working
306+
self.add_route_post(self.update_object) # this is working
307+
self.add_route_get (self.return_object) # this raises the exception
308+
309+
class An_Fast_API(Fast_API):
310+
default_routes = False
311+
def setup_routes(self):
312+
self.add_routes(Obj_Routes)
313+
314+
an_fast_api = An_Fast_API().setup()
315+
316+
assert an_fast_api.routes_paths() == ['/obj/create-object', '/obj/return-object', '/obj/update-object']
317+
318+
obj_data = An_Class(an_str='42').json()
319+
response_1 = an_fast_api.client().post('/obj/create-object', json=obj_data)
320+
assert response_1.json() is True
321+
322+
response_2 = an_fast_api.client().post('/obj/update-object', json=obj_data)
323+
assert response_2.json() == {'an_str': '42'}
324+
325+
response_3 = an_fast_api.client().get('/obj/return-object', params={'an_str':'123'})
326+
assert response_3.json() == {'an_str': '123'}

0 commit comments

Comments
 (0)