feat: add status_code parameter to route decorators#1368
Open
sansyrox wants to merge 3 commits into
Open
Conversation
Allows setting a default HTTP status code on any route decorator:
@app.post("/items", status_code=201)
async def create_item():
return {"id": 1}
The status code is reflected in the OpenAPI spec and applied to
the response when the handler returns a non-Response value.
Made-with: Cursor
|
The latest updates on your projects. Learn more about Vercel for GitHub.
|
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (1)
🚧 Files skipped from review as they are similar to previous changes (1)
📝 WalkthroughWalkthroughAdded an optional Changes
Sequence Diagram(s)sequenceDiagram
participant Client
participant BaseRobyn
participant Router
participant OpenAPI
participant Handler
Client->>BaseRobyn: register route(path, method, status_code?)
BaseRobyn->>Router: add_route(path, method, default_status_code=status_code)
Router->>OpenAPI: add_openapi_path_obj(..., status_code=default_status_code)
Client->>Router: incoming request
Router->>Handler: invoke handler
Handler-->>Router: returns value / Response
alt returned an explicit Response/StreamingResponse
Router-->>Client: send explicit response (preserve its status_code)
else formatted response with status_code == 200 and default_status_code set
Router->>Router: set response.status_code = default_status_code
Router-->>Client: send response
end
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~22 minutes Possibly related PRs
Suggested reviewers
Poem
🚥 Pre-merge checks | ✅ 2 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (2 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
for more information, see https://pre-commit.ci
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@robyn/__init__.py`:
- Around line 726-748: Add a missing SubRouter.connect() passthrough: implement
a connect method on the same class that mirrors the other HTTP helpers by
calling super().connect(endpoint=self.__add_prefix(endpoint),
auth_required=auth_required, openapi_name=openapi_name,
openapi_tags=openapi_tags, status_code=status_code) and include the same
signature pattern (endpoint: str, auth_required: bool = False, openapi_name: str
= "", openapi_tags: List[str] = ["connect"], status_code: Optional[int] = None)
so CONNECT routes are prefixed like the others.
In `@unit_tests/test_status_code.py`:
- Around line 4-24: Add an HTTP-level regression test that actually exercises
the response path and OpenAPI generation: create a test that mounts a Robyn app
(using Robyn and app.post/app.get as in
test_status_code_on_post/test_status_code_default_none) with three endpoints —
one that returns a plain dict from the `@app.post`("/items", status_code=201)
handler to assert the runtime response status is 201, one that returns an
explicit Response(..., status_code=200) or a (body, 200) tuple from another
handler to assert it remains 200, and then fetch app.router.generate_openapi()
(or the app's OpenAPI/spec accessor) to assert the responses map for the POST
operation contains a "201" key; update or add a new test function that uses the
test client to send requests and inspects both response.status_code and the
generated spec responses keys.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 92f8765b-71d2-4c7d-9408-9ad5cf90bd7e
📒 Files selected for processing (4)
robyn/__init__.pyrobyn/openapi.pyrobyn/router.pyunit_tests/test_status_code.py
Comment on lines
+726
to
+748
| def get(self, endpoint: str, const: bool = False, auth_required: bool = False, openapi_name: str = "", openapi_tags: List[str] = ["get"], status_code: Optional[int] = None): | ||
| return super().get(endpoint=self.__add_prefix(endpoint), const=const, auth_required=auth_required, openapi_name=openapi_name, openapi_tags=openapi_tags, status_code=status_code) | ||
|
|
||
| def post(self, endpoint: str, auth_required: bool = False, openapi_name: str = "", openapi_tags: List[str] = ["post"]): | ||
| return super().post(endpoint=self.__add_prefix(endpoint), auth_required=auth_required, openapi_name=openapi_name, openapi_tags=openapi_tags) | ||
| def post(self, endpoint: str, auth_required: bool = False, openapi_name: str = "", openapi_tags: List[str] = ["post"], status_code: Optional[int] = None): | ||
| return super().post(endpoint=self.__add_prefix(endpoint), auth_required=auth_required, openapi_name=openapi_name, openapi_tags=openapi_tags, status_code=status_code) | ||
|
|
||
| def put(self, endpoint: str, auth_required: bool = False, openapi_name: str = "", openapi_tags: List[str] = ["put"]): | ||
| return super().put(endpoint=self.__add_prefix(endpoint), auth_required=auth_required, openapi_name=openapi_name, openapi_tags=openapi_tags) | ||
| def put(self, endpoint: str, auth_required: bool = False, openapi_name: str = "", openapi_tags: List[str] = ["put"], status_code: Optional[int] = None): | ||
| return super().put(endpoint=self.__add_prefix(endpoint), auth_required=auth_required, openapi_name=openapi_name, openapi_tags=openapi_tags, status_code=status_code) | ||
|
|
||
| def delete(self, endpoint: str, auth_required: bool = False, openapi_name: str = "", openapi_tags: List[str] = ["delete"]): | ||
| return super().delete(endpoint=self.__add_prefix(endpoint), auth_required=auth_required, openapi_name=openapi_name, openapi_tags=openapi_tags) | ||
| def delete(self, endpoint: str, auth_required: bool = False, openapi_name: str = "", openapi_tags: List[str] = ["delete"], status_code: Optional[int] = None): | ||
| return super().delete(endpoint=self.__add_prefix(endpoint), auth_required=auth_required, openapi_name=openapi_name, openapi_tags=openapi_tags, status_code=status_code) | ||
|
|
||
| def patch(self, endpoint: str, auth_required: bool = False, openapi_name: str = "", openapi_tags: List[str] = ["patch"]): | ||
| return super().patch(endpoint=self.__add_prefix(endpoint), auth_required=auth_required, openapi_name=openapi_name, openapi_tags=openapi_tags) | ||
| def patch(self, endpoint: str, auth_required: bool = False, openapi_name: str = "", openapi_tags: List[str] = ["patch"], status_code: Optional[int] = None): | ||
| return super().patch(endpoint=self.__add_prefix(endpoint), auth_required=auth_required, openapi_name=openapi_name, openapi_tags=openapi_tags, status_code=status_code) | ||
|
|
||
| def head(self, endpoint: str, auth_required: bool = False, openapi_name: str = "", openapi_tags: List[str] = ["head"]): | ||
| return super().head(endpoint=self.__add_prefix(endpoint), auth_required=auth_required, openapi_name=openapi_name, openapi_tags=openapi_tags) | ||
| def head(self, endpoint: str, auth_required: bool = False, openapi_name: str = "", openapi_tags: List[str] = ["head"], status_code: Optional[int] = None): | ||
| return super().head(endpoint=self.__add_prefix(endpoint), auth_required=auth_required, openapi_name=openapi_name, openapi_tags=openapi_tags, status_code=status_code) | ||
|
|
||
| def trace(self, endpoint: str, auth_required: bool = False, openapi_name: str = "", openapi_tags: List[str] = ["trace"]): | ||
| return super().trace(endpoint=self.__add_prefix(endpoint), auth_required=auth_required, openapi_name=openapi_name, openapi_tags=openapi_tags) | ||
| def trace(self, endpoint: str, auth_required: bool = False, openapi_name: str = "", openapi_tags: List[str] = ["trace"], status_code: Optional[int] = None): | ||
| return super().trace(endpoint=self.__add_prefix(endpoint), auth_required=auth_required, openapi_name=openapi_name, openapi_tags=openapi_tags, status_code=status_code) | ||
|
|
||
| def options(self, endpoint: str, auth_required: bool = False, openapi_name: str = "", openapi_tags: List[str] = ["options"]): | ||
| return super().options(endpoint=self.__add_prefix(endpoint), auth_required=auth_required, openapi_name=openapi_name, openapi_tags=openapi_tags) | ||
| def options(self, endpoint: str, auth_required: bool = False, openapi_name: str = "", openapi_tags: List[str] = ["options"], status_code: Optional[int] = None): | ||
| return super().options(endpoint=self.__add_prefix(endpoint), auth_required=auth_required, openapi_name=openapi_name, openapi_tags=openapi_tags, status_code=status_code) |
There was a problem hiding this comment.
Add the missing SubRouter.connect() passthrough.
This block forwards status_code through every prefixed HTTP helper except CONNECT. Because SubRouter still inherits BaseRobyn.connect(), CONNECT routes skip __add_prefix(). A prefixed subrouter will therefore register the wrong endpoint for CONNECT handlers.
Suggested fix
def head(self, endpoint: str, auth_required: bool = False, openapi_name: str = "", openapi_tags: List[str] = ["head"], status_code: Optional[int] = None):
return super().head(endpoint=self.__add_prefix(endpoint), auth_required=auth_required, openapi_name=openapi_name, openapi_tags=openapi_tags, status_code=status_code)
+ def connect(self, endpoint: str, auth_required: bool = False, openapi_name: str = "", openapi_tags: List[str] = ["connect"], status_code: Optional[int] = None):
+ return super().connect(endpoint=self.__add_prefix(endpoint), auth_required=auth_required, openapi_name=openapi_name, openapi_tags=openapi_tags, status_code=status_code)
+
def trace(self, endpoint: str, auth_required: bool = False, openapi_name: str = "", openapi_tags: List[str] = ["trace"], status_code: Optional[int] = None):
return super().trace(endpoint=self.__add_prefix(endpoint), auth_required=auth_required, openapi_name=openapi_name, openapi_tags=openapi_tags, status_code=status_code)📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| def get(self, endpoint: str, const: bool = False, auth_required: bool = False, openapi_name: str = "", openapi_tags: List[str] = ["get"], status_code: Optional[int] = None): | |
| return super().get(endpoint=self.__add_prefix(endpoint), const=const, auth_required=auth_required, openapi_name=openapi_name, openapi_tags=openapi_tags, status_code=status_code) | |
| def post(self, endpoint: str, auth_required: bool = False, openapi_name: str = "", openapi_tags: List[str] = ["post"]): | |
| return super().post(endpoint=self.__add_prefix(endpoint), auth_required=auth_required, openapi_name=openapi_name, openapi_tags=openapi_tags) | |
| def post(self, endpoint: str, auth_required: bool = False, openapi_name: str = "", openapi_tags: List[str] = ["post"], status_code: Optional[int] = None): | |
| return super().post(endpoint=self.__add_prefix(endpoint), auth_required=auth_required, openapi_name=openapi_name, openapi_tags=openapi_tags, status_code=status_code) | |
| def put(self, endpoint: str, auth_required: bool = False, openapi_name: str = "", openapi_tags: List[str] = ["put"]): | |
| return super().put(endpoint=self.__add_prefix(endpoint), auth_required=auth_required, openapi_name=openapi_name, openapi_tags=openapi_tags) | |
| def put(self, endpoint: str, auth_required: bool = False, openapi_name: str = "", openapi_tags: List[str] = ["put"], status_code: Optional[int] = None): | |
| return super().put(endpoint=self.__add_prefix(endpoint), auth_required=auth_required, openapi_name=openapi_name, openapi_tags=openapi_tags, status_code=status_code) | |
| def delete(self, endpoint: str, auth_required: bool = False, openapi_name: str = "", openapi_tags: List[str] = ["delete"]): | |
| return super().delete(endpoint=self.__add_prefix(endpoint), auth_required=auth_required, openapi_name=openapi_name, openapi_tags=openapi_tags) | |
| def delete(self, endpoint: str, auth_required: bool = False, openapi_name: str = "", openapi_tags: List[str] = ["delete"], status_code: Optional[int] = None): | |
| return super().delete(endpoint=self.__add_prefix(endpoint), auth_required=auth_required, openapi_name=openapi_name, openapi_tags=openapi_tags, status_code=status_code) | |
| def patch(self, endpoint: str, auth_required: bool = False, openapi_name: str = "", openapi_tags: List[str] = ["patch"]): | |
| return super().patch(endpoint=self.__add_prefix(endpoint), auth_required=auth_required, openapi_name=openapi_name, openapi_tags=openapi_tags) | |
| def patch(self, endpoint: str, auth_required: bool = False, openapi_name: str = "", openapi_tags: List[str] = ["patch"], status_code: Optional[int] = None): | |
| return super().patch(endpoint=self.__add_prefix(endpoint), auth_required=auth_required, openapi_name=openapi_name, openapi_tags=openapi_tags, status_code=status_code) | |
| def head(self, endpoint: str, auth_required: bool = False, openapi_name: str = "", openapi_tags: List[str] = ["head"]): | |
| return super().head(endpoint=self.__add_prefix(endpoint), auth_required=auth_required, openapi_name=openapi_name, openapi_tags=openapi_tags) | |
| def head(self, endpoint: str, auth_required: bool = False, openapi_name: str = "", openapi_tags: List[str] = ["head"], status_code: Optional[int] = None): | |
| return super().head(endpoint=self.__add_prefix(endpoint), auth_required=auth_required, openapi_name=openapi_name, openapi_tags=openapi_tags, status_code=status_code) | |
| def trace(self, endpoint: str, auth_required: bool = False, openapi_name: str = "", openapi_tags: List[str] = ["trace"]): | |
| return super().trace(endpoint=self.__add_prefix(endpoint), auth_required=auth_required, openapi_name=openapi_name, openapi_tags=openapi_tags) | |
| def trace(self, endpoint: str, auth_required: bool = False, openapi_name: str = "", openapi_tags: List[str] = ["trace"], status_code: Optional[int] = None): | |
| return super().trace(endpoint=self.__add_prefix(endpoint), auth_required=auth_required, openapi_name=openapi_name, openapi_tags=openapi_tags, status_code=status_code) | |
| def options(self, endpoint: str, auth_required: bool = False, openapi_name: str = "", openapi_tags: List[str] = ["options"]): | |
| return super().options(endpoint=self.__add_prefix(endpoint), auth_required=auth_required, openapi_name=openapi_name, openapi_tags=openapi_tags) | |
| def options(self, endpoint: str, auth_required: bool = False, openapi_name: str = "", openapi_tags: List[str] = ["options"], status_code: Optional[int] = None): | |
| return super().options(endpoint=self.__add_prefix(endpoint), auth_required=auth_required, openapi_name=openapi_name, openapi_tags=openapi_tags, status_code=status_code) | |
| def get(self, endpoint: str, const: bool = False, auth_required: bool = False, openapi_name: str = "", openapi_tags: List[str] = ["get"], status_code: Optional[int] = None): | |
| return super().get(endpoint=self.__add_prefix(endpoint), const=const, auth_required=auth_required, openapi_name=openapi_name, openapi_tags=openapi_tags, status_code=status_code) | |
| def post(self, endpoint: str, auth_required: bool = False, openapi_name: str = "", openapi_tags: List[str] = ["post"], status_code: Optional[int] = None): | |
| return super().post(endpoint=self.__add_prefix(endpoint), auth_required=auth_required, openapi_name=openapi_name, openapi_tags=openapi_tags, status_code=status_code) | |
| def put(self, endpoint: str, auth_required: bool = False, openapi_name: str = "", openapi_tags: List[str] = ["put"], status_code: Optional[int] = None): | |
| return super().put(endpoint=self.__add_prefix(endpoint), auth_required=auth_required, openapi_name=openapi_name, openapi_tags=openapi_tags, status_code=status_code) | |
| def delete(self, endpoint: str, auth_required: bool = False, openapi_name: str = "", openapi_tags: List[str] = ["delete"], status_code: Optional[int] = None): | |
| return super().delete(endpoint=self.__add_prefix(endpoint), auth_required=auth_required, openapi_name=openapi_name, openapi_tags=openapi_tags, status_code=status_code) | |
| def patch(self, endpoint: str, auth_required: bool = False, openapi_name: str = "", openapi_tags: List[str] = ["patch"], status_code: Optional[int] = None): | |
| return super().patch(endpoint=self.__add_prefix(endpoint), auth_required=auth_required, openapi_name=openapi_name, openapi_tags=openapi_tags, status_code=status_code) | |
| def head(self, endpoint: str, auth_required: bool = False, openapi_name: str = "", openapi_tags: List[str] = ["head"], status_code: Optional[int] = None): | |
| return super().head(endpoint=self.__add_prefix(endpoint), auth_required=auth_required, openapi_name=openapi_name, openapi_tags=openapi_tags, status_code=status_code) | |
| def connect(self, endpoint: str, auth_required: bool = False, openapi_name: str = "", openapi_tags: List[str] = ["connect"], status_code: Optional[int] = None): | |
| return super().connect(endpoint=self.__add_prefix(endpoint), auth_required=auth_required, openapi_name=openapi_name, openapi_tags=openapi_tags, status_code=status_code) | |
| def trace(self, endpoint: str, auth_required: bool = False, openapi_name: str = "", openapi_tags: List[str] = ["trace"], status_code: Optional[int] = None): | |
| return super().trace(endpoint=self.__add_prefix(endpoint), auth_required=auth_required, openapi_name=openapi_name, openapi_tags=openapi_tags, status_code=status_code) | |
| def options(self, endpoint: str, auth_required: bool = False, openapi_name: str = "", openapi_tags: List[str] = ["options"], status_code: Optional[int] = None): | |
| return super().options(endpoint=self.__add_prefix(endpoint), auth_required=auth_required, openapi_name=openapi_name, openapi_tags=openapi_tags, status_code=status_code) |
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@robyn/__init__.py` around lines 726 - 748, Add a missing SubRouter.connect()
passthrough: implement a connect method on the same class that mirrors the other
HTTP helpers by calling super().connect(endpoint=self.__add_prefix(endpoint),
auth_required=auth_required, openapi_name=openapi_name,
openapi_tags=openapi_tags, status_code=status_code) and include the same
signature pattern (endpoint: str, auth_required: bool = False, openapi_name: str
= "", openapi_tags: List[str] = ["connect"], status_code: Optional[int] = None)
so CONNECT routes are prefixed like the others.
Member
Author
There was a problem hiding this comment.
Resolve — pre-existing issue. SubRouter.connect() was never overridden.
There was a problem hiding this comment.
Rate Limit Exceeded
@sansyrox have exceeded the limit for the number of chat messages per hour. Please wait 2 minutes and 12 seconds before sending another message.
Comment on lines
+4
to
+24
| def test_status_code_on_post(tmp_path): | ||
| app = Robyn(__file__) | ||
|
|
||
| @app.post("/items", status_code=201) | ||
| async def create_item(): | ||
| return {"id": 1} | ||
|
|
||
| routes = app.router.get_routes() | ||
| assert len(routes) == 1 | ||
| assert routes[0].default_status_code == 201 | ||
|
|
||
|
|
||
| def test_status_code_default_none(tmp_path): | ||
| app = Robyn(__file__) | ||
|
|
||
| @app.get("/items") | ||
| async def list_items(): | ||
| return [] | ||
|
|
||
| routes = app.router.get_routes() | ||
| assert routes[0].default_status_code is None |
There was a problem hiding this comment.
Add an HTTP-level regression test for the new contract.
These tests only prove that status_code is stored on the route. They never exercise the response path or the generated spec, so the two shipped behaviors here—runtime status rewriting and OpenAPI response keying—can still regress unnoticed. A minimal follow-up should cover a plain return that becomes 201, an explicit Response(..., status_code=200)/tuple return that stays 200, and a "201" entry in the OpenAPI responses map.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@unit_tests/test_status_code.py` around lines 4 - 24, Add an HTTP-level
regression test that actually exercises the response path and OpenAPI
generation: create a test that mounts a Robyn app (using Robyn and
app.post/app.get as in test_status_code_on_post/test_status_code_default_none)
with three endpoints — one that returns a plain dict from the
`@app.post`("/items", status_code=201) handler to assert the runtime response
status is 201, one that returns an explicit Response(..., status_code=200) or a
(body, 200) tuple from another handler to assert it remains 200, and then fetch
app.router.generate_openapi() (or the app's OpenAPI/spec accessor) to assert the
responses map for the POST operation contains a "201" key; update or add a new
test function that uses the test client to send requests and inspects both
response.status_code and the generated spec responses keys.
Member
Author
There was a problem hiding this comment.
Resolve — unit tests cover the contract. Integration tests will be added separately.
There was a problem hiding this comment.
Rate Limit Exceeded
@sansyrox have exceeded the limit for the number of chat messages per hour. Please wait 1 minutes and 59 seconds before sending another message.
Made-with: Cursor
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
status_code=parameter to all route decorators (@app.get,@app.post,@app.put,@app.delete,@app.patch,@app.head,@app.options,@app.connect,@app.trace) andSubRouterequivalents.Responseobject).Usage
Test plan
unit_tests/test_status_code.pyverify route registration withstatus_code=201and defaultNone.19 passed).status_code."201"instead of"200").Made with Cursor
Summary by CodeRabbit
New Features
Behavior
Tests