Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 12 additions & 1 deletion integration_tests/base_routes.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
import os
import pathlib
from collections import defaultdict
from typing import Optional
from typing import Optional, List

from integration_tests.subroutes import di_subrouter, sub_router
from robyn import Headers, Request, Response, Robyn, WebSocket, WebSocketConnector, jsonify, serve_file, serve_html
Expand Down Expand Up @@ -561,6 +561,11 @@ async def async_dict_post():
# Body


class TestMyRequest(Body):
items: List[str]
numbers: list[int]


@app.post("/sync/body")
def sync_body_post(request: Request):
return request.body
Expand All @@ -576,6 +581,12 @@ def sync_form_data(request: Request):
return request.headers["Content-Type"]


@app.post("/sync/body/typed")
def sync_body_typed(body: TestMyRequest):
# the server should just start
return "This is not tested with a request"


# JSON Request


Expand Down
3 changes: 3 additions & 0 deletions robyn/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -607,6 +607,9 @@ def ALLOW_CORS(app: Robyn, origins: Union[List[str], str]):

@app.before_request()
def cors_middleware(request):
if request is None:
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The CORS middleware's handling of None request is incorrect. Middlewares in Robyn should either return the original request object to continue processing or a Response object to short-circuit. Returning None (from line 611) will likely cause undefined behavior or errors in the request processing pipeline. Instead, it should return an appropriate Response object with an error status code if the request is invalid.


React with 👍 to tell me that this comment was useful, or 👎 if not (and I'll stop posting more comments like this in the future)

return None

origin = request.headers.get("Origin")

# If specific origins are set, validate the request origin
Expand Down
36 changes: 25 additions & 11 deletions robyn/openapi.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@
from importlib import resources
from inspect import Signature
from pathlib import Path
from typing import Any, Callable, Dict, List, Optional, Tuple, TypedDict
from typing import Any, Callable, Dict, List, Optional, Tuple, TypedDict, Union

from robyn.responses import html
from robyn.robyn import QueryParams, Response
Expand Down Expand Up @@ -389,25 +389,39 @@ def get_schema_object(self, parameter: str, param_type: Any) -> dict:
list: "array",
}

# Handle basic types
for type_name in type_mapping:
if param_type is type_name:
properties["type"] = type_mapping[type_name]
return properties

# check for Optional type
if param_type.__module__ == "typing":
properties["anyOf"] = [{"type": self.get_openapi_type(param_type.__args__[0])}, {"type": "null"}]
return properties
# check for custom classes and TypedDicts
# Handle typing module types (Optional, List, etc)
if hasattr(param_type, "__module__") and param_type.__module__ == "typing":
origin = typing.get_origin(param_type)
args = typing.get_args(param_type)

# Handle Optional types
if origin is Union and type(None) in args:
non_none_type = next(t for t in args if t is not type(None))
properties["anyOf"] = [{"type": self.get_openapi_type(non_none_type)}, {"type": "null"}]
return properties

# Handle List types
elif origin in (list, List):
properties["type"] = "array"
if args:
item_type = args[0]
properties["items"] = self.get_schema_object("item", item_type)
return properties

# Handle custom classes and TypedDicts
elif inspect.isclass(param_type):
properties["type"] = "object"

properties["properties"] = {}

for e in param_type.__annotations__:
properties["properties"][e] = self.get_schema_object(e, param_type.__annotations__[e])

properties["type"] = "object"
if hasattr(param_type, "__annotations__"):
for e in param_type.__annotations__:
properties["properties"][e] = self.get_schema_object(e, param_type.__annotations__[e])

return properties

Expand Down
Loading