diff --git a/notes.md b/notes.md
new file mode 100644
index 0000000..ab62aad
--- /dev/null
+++ b/notes.md
@@ -0,0 +1,36 @@
+1. Out-of-the-box Real-time Capabilities
+ WebSocket & Server-Sent Events (SSE) Support: Provide built-in support for WebSocket and SSE endpoints with seamless integration into the frameworkâs routing and middleware systems.
+ Event Broadcasting and Pub/Sub: Integrate a simple pub/sub or event broadcasting system (using Redis or another broker) for real-time features like chat applications, live updates, or collaborative tools.
+
+2. First-class Support for Background Tasks and Workers
+ Integrated Worker Queue: Include a built-in task queue system (similar to Celery) for background processing with async task support. Tasks could be defined in the same codebase with decorators and could run in separate worker processes.
+ Job Scheduling: Native support for scheduling tasks with a cron-like syntax.
+
+3. Intuitive Error Handling and Debugging
+ Error Tracing and Categorization: Introduce a built-in error-handling mechanism that categorizes errors, traces them to their origin, and provides detailed context in logs or error responses.
+ Middleware for Debugging: Provide middleware that allows inspecting the internal state, dependencies, and the request lifecycle in real-time.
+
+4. Security Enhancements
+ Built-in Security Middleware: Include robust security middleware by default for features like CSRF protection, CORS handling, rate limiting, and DDoS protection.
+ Advanced Auth Integration: Built-in support for advanced authentication methods, including OAuth2, SAML, JWT, and OpenID Connect, with easy configuration.
+
+5. Optimized for Serverless and Edge Deployments
+ Edge-optimized and Serverless-first: Design the framework to be easily deployable on serverless platforms (like AWS Lambda or Vercel) or edge networks, with minimal configuration and optimized cold-start times.
+ Pre-rendering and Edge Caching: Integrate strategies for pre-rendering content and edge caching, making it more performant for static content delivery.
+
+6. Automatic API Documentation with Extensions
+ Customizable and Dynamic Documentation: Provide a dynamic documentation system like Swagger or Redoc but with more customization options (themes, plugins) and interactive features, such as example requests/responses based on schema.
+ API Versioning and Deprecation Notices: Automatically handle API versioning and provide built-in tools for deprecating old versions gracefully.
+
+7. Declarative Data Validation and Transformation
+ Advanced Validation Framework: Go beyond Pydantic models by offering a more declarative approach for complex data validation, transformation, and coercion, including custom validation rules and error messages.
+
+8. GraphQL and gRPC Integration
+ GraphQL Server with Subscriptions: Provide first-class support for building GraphQL servers with built-in support for subscriptions.
+ gRPC Support: Offer built-in support for gRPC endpoints to enable high-performance RPCs for microservices.
+
+ Schema-first Approach: Focus on a design-first approach where API definitions (using OpenAPI or GraphQL schema) drive the development process.
+ Automatic Code Generation: Enable automatic code generation for client SDKs, server stubs, and documentation based on the schema.
+
+9. Data Streaming and Real-time Data Sync
+ Built-in Data Streaming: Provide support for data streaming (using WebSockets or HTTP/2) and real-time data synchronization, which is particularly useful for collaborative applications.
diff --git a/src/ziplineio/app.py b/src/ziplineio/app.py
index 35020d8..3abe8b6 100644
--- a/src/ziplineio/app.py
+++ b/src/ziplineio/app.py
@@ -4,7 +4,7 @@
from ziplineio.exception import NotFoundHttpException
-from ziplineio.middleware import middleware, run_middleware_stack
+from ziplineio.middleware import run_middleware_stack
from ziplineio.dependency_injector import inject, injector, DependencyInjector
from ziplineio import settings
from ziplineio.handler import Handler
@@ -89,12 +89,12 @@ async def _get_and_call_handler(
if handler is not None:
# If a handler is found, call it with the request
- return await call_handler(handler, req)
+ return await call_handler(handler, req=req)
# If no handler was found, attempt to run middlewares.
# (If a handler was found, middlewares will be run by `call_handler`)
req, ctx, res = await run_middleware_stack(
- self._router._router_level_middelwares, req
+ self._router._router_level_middelwares, req=req
)
# If middleware does not provide a response, return a 404 Not Found
@@ -103,7 +103,7 @@ async def _get_and_call_handler(
return res
if self._router._not_found_handler:
- response = await call_handler(self._router._not_found_handler, req)
+ response = await call_handler(self._router._not_found_handler, req=req)
headers = isinstance(response, Response) and response._headers or {}
return NotFoundResponse(response, headers)
diff --git a/src/ziplineio/html/jinja.py b/src/ziplineio/html/jinja.py
index aa86edc..187c988 100644
--- a/src/ziplineio/html/jinja.py
+++ b/src/ziplineio/html/jinja.py
@@ -14,7 +14,7 @@ async def wrapped_handler(req, **kwargs):
# Filter kwargs to only pass those that the handler expects
# filtered_kwargs = {k: v for k, v in kwargs.items() if k in sig.parameters}
- context = await call_handler(handler, req, **kwargs)
+ context = await call_handler(handler, req=req, **kwargs)
rendered = template.render(context)
return JinjaResponse(rendered)
diff --git a/src/ziplineio/middleware.py b/src/ziplineio/middleware.py
index a318dec..f051438 100644
--- a/src/ziplineio/middleware.py
+++ b/src/ziplineio/middleware.py
@@ -1,6 +1,5 @@
-import inspect
from typing import List, Callable, Tuple
-from ziplineio import response
+
from ziplineio.handler import Handler
from ziplineio.request import Request
from ziplineio.response import Response
@@ -15,7 +14,9 @@ async def wrapped_handler(req: Request, **kwargs):
kwargs.setdefault("ctx", {})
# Run the middleware stack
- req, kwargs, res = await run_middleware_stack(middlewares, req, **kwargs)
+ req, kwargs, res = await run_middleware_stack(
+ middlewares, req=req, **kwargs
+ )
if res is not None:
return res
@@ -28,7 +29,7 @@ async def wrapped_handler(req: Request, **kwargs):
async def run_middleware_stack(
- middlewares: list[Handler], request: Request, **kwargs
+ middlewares: list[Handler], req: Request, **kwargs
) -> Tuple[Request, dict, bytes | str | dict | Response | None]:
for middleware in middlewares:
# if the middleware func takes params, pass them in. Otherwise, just pass req
@@ -36,7 +37,7 @@ async def run_middleware_stack(
if "ctx" not in kwargs:
kwargs["ctx"] = {}
- _res = await call_handler(middleware, request, **kwargs)
+ _res = await call_handler(middleware, req=req, **kwargs)
# regular handlers return a response, but middleware can return a tuple
if not isinstance(_res, tuple):
@@ -51,6 +52,6 @@ async def run_middleware_stack(
if not isinstance(req, Request):
response = req
- return request, kwargs, response
+ return req, kwargs, response
- return request, kwargs, None
+ return req, kwargs, None
diff --git a/src/ziplineio/utils.py b/src/ziplineio/utils.py
index 92b8b41..9bc4c2b 100644
--- a/src/ziplineio/utils.py
+++ b/src/ziplineio/utils.py
@@ -2,6 +2,8 @@
import inspect
import asyncio
from typing import Dict
+
+from httpx import request
from ziplineio.request import Request
from ziplineio.response import Response
from ziplineio.handler import Handler
@@ -18,11 +20,12 @@ def get_class_params(cls):
async def call_handler(
handler: Handler,
- req: Request,
**kwargs,
) -> bytes | str | dict | Response | Exception:
try:
- kwargs = {"req": req, **kwargs}
+ if "req" not in kwargs:
+ raise ValueError("Request object not found in kwargs")
+
params = inspect.signature(handler).parameters
kwargs = {k: v for k, v in kwargs.items() if k in params}
if not inspect.iscoroutinefunction(handler):
diff --git a/src/ziplineio/validation/__init__.py b/src/ziplineio/validation/__init__.py
new file mode 100644
index 0000000..708bacd
--- /dev/null
+++ b/src/ziplineio/validation/__init__.py
@@ -0,0 +1,3 @@
+from .body import BodyParam, validate_body
+from .query import QueryParam, validate_query
+from .index import validate
diff --git a/src/ziplineio/validation/body.py b/src/ziplineio/validation/body.py
new file mode 100644
index 0000000..e4c11b1
--- /dev/null
+++ b/src/ziplineio/validation/body.py
@@ -0,0 +1,46 @@
+from functools import wraps
+from ziplineio.request import Request
+from pydantic import ValidationError
+
+from ziplineio.validation.query import QueryParam
+
+
+class BodyParam(QueryParam):
+ """Alias Query param."""
+
+ pass
+
+
+def validate_body(*body_params):
+ """
+ Decorator to validate body parameters.
+ """
+
+ def decorator(handler):
+ @wraps(handler)
+ async def wrapper(req: Request, *args, **kwargs):
+ errors = {}
+ validated_body = {}
+
+ for param in body_params:
+ param_name = param.param
+ value = req.body.get(param_name)
+ if value is None:
+ if param.required:
+ errors[param_name] = "Missing required body parameter"
+ else:
+ try:
+ validated_body[param_name] = param.validate(value)
+ except (ValueError, ValidationError) as e:
+ errors[param_name] = str(e)
+
+ if errors:
+ return {"errors": errors}, 400
+
+ # Attach validated body parameters to the request
+ req.body = validated_body
+ return await handler(req, *args, **kwargs)
+
+ return wrapper
+
+ return decorator if body_params else lambda req: decorator(lambda *_: None)(req)
diff --git a/src/ziplineio/validation/index.py b/src/ziplineio/validation/index.py
new file mode 100644
index 0000000..9847c38
--- /dev/null
+++ b/src/ziplineio/validation/index.py
@@ -0,0 +1,45 @@
+from ziplineio.request import Request
+
+
+def validate(*validators):
+ def decorator(handler):
+ async def wrapper(req: Request, ctx: dict):
+ errors = {}
+ validated_body = {}
+
+ for validator in validators:
+ validation_result = validator(req)
+ validated_body.update(validation_result.get("validated", {}))
+ errors.update(validation_result.get("errors", {}))
+
+ if errors:
+ return {"errors": errors}, 400
+
+ req.body = validated_body
+ return await handler(req, ctx)
+
+ return wrapper
+
+ return decorator
+
+
+# # Example Pydantic model
+# class UserModel(BaseModel):
+# username: str
+# age: int
+
+
+# # Usage example with Pydantic model
+# @app.post("/")
+# @validate_body(BodyParam("user", UserModel))
+# async def create_user_handler(req: Request, ctx: dict):
+# user = req.validated_body.get("user")
+# return {"user": user.dict()} # Access the validated Pydantic model instance
+
+
+# # Usage example with simple type validation
+# @app.get("/")
+# @validate_query(QueryParam("bar", str))
+# async def test_handler(req: Request, ctx: dict):
+# bar = req.validated_query.get("bar")
+# return {"bar": bar}
diff --git a/src/ziplineio/validation/query.py b/src/ziplineio/validation/query.py
new file mode 100644
index 0000000..92373b6
--- /dev/null
+++ b/src/ziplineio/validation/query.py
@@ -0,0 +1,86 @@
+from functools import wraps
+import inspect
+from typing import Type, Union
+from pydantic import BaseModel, ValidationError
+from ziplineio.request import Request
+
+
+class QueryParam:
+ def __init__(
+ self, param: str, type: Union[Type, BaseModel] = str, required: bool = True
+ ):
+ self.param = param
+ self.type = type
+ self.required = required
+
+ def validate(self, value):
+ if isinstance(self.type, type) and issubclass(self.type, BaseModel):
+ try:
+ # Use Pydantic model for validation
+ model_instance = self.type.parse_obj(value)
+ return model_instance
+ except ValidationError as e:
+ raise ValueError(f"Validation error for {self.param}: {e}")
+
+ try:
+ # Basic type casting
+ return self.type(value)
+ except ValueError:
+ raise ValueError(
+ f"Invalid type for {self.param}, expected {self.type.__name__}"
+ )
+
+
+def validate_query(*query_params):
+ """
+ Decorator to validate query parameters.
+ """
+
+ def decorator(handler):
+ # Use inspect to get the function signature and find the request parameter name
+ signature = inspect.signature(handler)
+ request_param_name = None
+
+ # Find the parameter annotated with the Request type
+ for param_name, param in signature.parameters.items():
+ if param.annotation == Request or isinstance(
+ param_name, Request
+ ): # Look for the parameter with type `Request`
+ request_param_name = param_name
+ break
+
+ if request_param_name is None:
+ raise ValueError(
+ "Handler function must have a parameter of type `Request`."
+ )
+
+ @wraps(handler)
+ async def wrapper(*args, **kwargs):
+ req = kwargs.get(request_param_name)
+
+ errors = {}
+ validated_query = {}
+
+ for param in query_params:
+ param_name = param.param
+ value = req.query_params.get(param_name)
+ if value is None:
+ if param.required:
+ errors[param_name] = "Missing required query parameter"
+ else:
+ try:
+ validated_query[param_name] = param.validate(value)
+ except (ValueError, ValidationError) as e:
+ errors[param_name] = str(e)
+
+ if errors:
+ return {"errors": errors}, 400
+
+ # Attach validated query parameters to the request
+ req.query_params = validated_query
+
+ return await handler(*args, **kwargs)
+
+ return wrapper
+
+ return decorator if query_params else lambda req: decorator(lambda *_: None)(req)
diff --git a/test/test_app.py b/test/test_app.py
index beac9f7..096bff0 100644
--- a/test/test_app.py
+++ b/test/test_app.py
@@ -42,6 +42,23 @@ async def test_handler(req: Request, ctx: dict):
# Assertions
self.assertEqual(response["message"], "Hello, world!")
+ async def test_query_params(self):
+ # Mock request data
+ req = Request(method="GET", path="/")
+ req.query_params = {"bar": "baz"}
+
+ # Define a simple handler for testing
+ @self.app.get("/")
+ async def test_handler(req: Request, ctx: dict):
+ return {"message": req.query_params["bar"]}
+
+ # Call the route
+ handler, params = self.app._router.get_handler("GET", "/")
+ response = await handler(req, {})
+
+ # Assertions
+ self.assertEqual(response["message"], "baz")
+
async def test_middleware_injection(self):
# Mock request data
req = Request(method="GET", path="/with-middleware")
diff --git a/test/test_dependency_injection.py b/test/test_dependency_injection.py
index 110e252..7981293 100644
--- a/test/test_dependency_injection.py
+++ b/test/test_dependency_injection.py
@@ -147,6 +147,6 @@ async def test_handler(req, service1: Service1):
# Call the handler
handler, params = self.ziplineio.get_handler("GET", "/")
- response = await call_handler(handler, {})
+ response = await call_handler(handler, req={})
self.assertEqual(response["message"], "Service 1")
diff --git a/test/test_e2e.py b/test/test_e2e.py
index b6ef4fb..9c956aa 100644
--- a/test/test_e2e.py
+++ b/test/test_e2e.py
@@ -4,7 +4,6 @@
import requests
import uvicorn
import unittest
-import unittest.async_case
from ziplineio.app import App
@@ -81,7 +80,19 @@ async def asyncSetUp(self):
"""Bring server up."""
self.proc = Process(target=run_server, args=(), daemon=False)
self.proc.start()
- await asyncio.sleep(0.2) # time for the server to start
+
+ # Wait for the server to be up
+ await self.wait_for_server()
+
+ async def wait_for_server(self):
+ """Wait for the server to be ready."""
+ while True:
+ try:
+ response = requests.get("http://localhost:5050/bytes")
+ if response.status_code == 200:
+ break
+ except requests.ConnectionError:
+ await asyncio.sleep(0.1) # Short sleep between retries
async def test_handler_returns_bytes(self):
response = requests.get("http://localhost:5050/bytes")
diff --git a/test/test_exceptions.py b/test/test_exceptions.py
index 5719914..defd1e9 100644
--- a/test/test_exceptions.py
+++ b/test/test_exceptions.py
@@ -19,7 +19,7 @@ async def test_handler(req):
raise BaseHttpException("Hey! We messed up", 409)
# Call the handler
- response = await call_handler(test_handler, {})
+ response = await call_handler(test_handler, req={})
response = format_response(response, settings.DEFAULT_HEADERS)
self.assertEqual(response["body"], b"Hey! We messed up")
self.assertEqual(response["status"], 409)
@@ -33,7 +33,7 @@ async def test_handler(req):
raise CustomHttpException("Hey! We messed up bad", 402)
# Call the handler
- response = await call_handler(test_handler, {})
+ response = await call_handler(test_handler, req={})
response = format_response(response, settings.DEFAULT_HEADERS)
self.assertEqual(response["body"], b"Hey! We messed up bad")
self.assertEqual(response["status"], 402)
diff --git a/test/test_html.py b/test/test_html.py
index 104dd42..2e57381 100644
--- a/test/test_html.py
+++ b/test/test_html.py
@@ -1,42 +1,30 @@
-import asyncio
-from multiprocessing import Process
import unittest
-import requests
-import uvicorn
+
from jinja2 import Environment, PackageLoader, select_autoescape
from ziplineio.app import App
+from ziplineio.request import Request
from ziplineio.html.jinja import jinja
-
-env = Environment(loader=PackageLoader("test.mocks"), autoescape=select_autoescape())
-
-app = App()
+from ziplineio.response import JinjaResponse
-@app.get("/")
-@jinja(env, "home.html")
-def home(req):
- return {"message": "Hello, world!"}
-
-
-def run_server():
- uvicorn.run(app, port=5050)
+env = Environment(loader=PackageLoader("test.mocks"), autoescape=select_autoescape())
class TestHtml(unittest.IsolatedAsyncioTestCase):
- async def asyncSetUp(self):
- """Bring server up."""
- self.proc = Process(target=run_server, args=(), daemon=False)
- self.proc.start()
- await asyncio.sleep(0.3) # time for the server to start
+ def setUp(self):
+ self.app = App()
async def test_render_jinja(self):
- response = requests.get("http://localhost:5050/")
+ @self.app.get("/")
+ @jinja(env, "home.html")
+ def home(req):
+ return {"message": "Hello, world!"}
- self.assertEqual(response.status_code, 200)
- self.assertEqual(response.headers["Content-Type"], "text/html")
- self.assertTrue("Welcome to the home page!
" in response.text)
+ req = Request(method="GET", path="/")
+ response: JinjaResponse = await self.app._get_and_call_handler("GET", "/", req)
- async def asyncTearDown(self):
- self.proc.terminate()
+ self.assertEqual(response.status, 200)
+ self.assertEqual(response.get_headers()["Content-Type"], "text/html")
+ self.assertTrue("Welcome to the home page!
" in response.body)
diff --git a/test/test_middleware.py b/test/test_middleware.py
index 15d7853..b700946 100644
--- a/test/test_middleware.py
+++ b/test/test_middleware.py
@@ -1,4 +1,3 @@
-from operator import call
import unittest
from ziplineio import settings
from ziplineio.request import Request
@@ -83,7 +82,7 @@ async def test_handler_with_middleware(req: Request, ctx: dict):
# Call the route
handler, params = self.app._router.get_handler("GET", "/with-middleware")
- response = await call_handler(handler, req)
+ response = await call_handler(handler, req=req)
response = format_response(response, settings.DEFAULT_HEADERS)
# Assertions
@@ -115,7 +114,7 @@ async def test_handler_with_middleware(req: Request, ctx: dict):
# Call the route
handler, params = self.app._router.get_handler("GET", "/with-middleware")
- response = await call_handler(handler, req)
+ response = await call_handler(handler, req=req)
# Assertions
self.assertEqual(response["message"], "Hi from middleware 2")
diff --git a/test/test_response.py b/test/test_response.py
index b0e5879..9214a11 100644
--- a/test/test_response.py
+++ b/test/test_response.py
@@ -1,40 +1,25 @@
-from multiprocessing import Process
-
import unittest
-
-import uvicorn
-import requests
-import asyncio
-
from ziplineio.app import App
-
-
-app = App()
-app.static("test/mocks/static", path_prefix="/static")
-
-
-@app.get("/")
-async def home(req):
- return {"message": "Hello, world!"}
-
-
-def run_server():
- uvicorn.run(app, port=5050)
+from ziplineio.request import Request
+from ziplineio.response import StaticFileResponse
class TestResponse(unittest.IsolatedAsyncioTestCase):
- async def asyncSetUp(self):
- """Bring server up."""
- self.proc = Process(target=run_server, args=(), daemon=False)
- self.proc.start()
- await asyncio.sleep(0.2) # time for the server to start
+ def setUp(self):
+ self.app = App()
+ self.app.static("test/mocks/static", path_prefix="/static")
- async def test_static_file(self):
- r = requests.get("http://localhost:5050/static/css/test.css")
- self.assertEqual(r.status_code, 200)
- self.assertEqual(r.headers["Content-Type"], "text/css")
- self.assertTrue("background-color: #f0f0f0;" in r.text)
+ @self.app.get("/")
+ async def home(req):
+ return {"message": "Hello, world!"}
- async def asyncTearDown(self):
- self.proc.terminate()
+ async def test_static_file(self):
+ req = Request(method="GET", path="/static/css/test.css")
+ r: StaticFileResponse = await self.app._get_and_call_handler(
+ "GET", "/static/css/test.css", req
+ )
+
+ self.assertEqual(r.status, 200)
+ self.assertEqual(r.get_headers()["Content-Type"], "text/css")
+ self.assertTrue(b"background-color: #f0f0f0;" in r.body)
diff --git a/test/test_validation.py b/test/test_validation.py
new file mode 100644
index 0000000..f9dad2c
--- /dev/null
+++ b/test/test_validation.py
@@ -0,0 +1,198 @@
+import unittest
+
+from pydantic import BaseModel
+
+
+from ziplineio.app import App
+
+from ziplineio.request import Request
+
+from ziplineio.utils import call_handler
+from ziplineio.validation.body import BodyParam, validate_body
+from ziplineio.validation.query import QueryParam, validate_query
+
+
+class TestValidateBody(unittest.IsolatedAsyncioTestCase):
+ def setUp(self):
+ # Initialize the app
+ self.app = App()
+
+ async def test_validate_simple_body_params(self):
+ @self.app.post("/")
+ @validate_body(BodyParam("username", str), BodyParam("age", float))
+ async def create_user_handler(req: Request):
+ username = req.body.get("username")
+ age = req.body.get("age")
+ return {"username": username, "age": age}
+
+ req = Request(method="POST", path="/", body={"username": "Amy", "age": 0.11})
+ response = await call_handler(create_user_handler, req=req)
+
+ self.assertEqual(response["username"], "Amy")
+ self.assertEqual(response["age"], 0.11)
+
+ async def test_validate_pydantic_body_params(self):
+ # # Example Pydantic model
+ class UserModel(BaseModel):
+ username: str
+ age: int
+
+ @self.app.post("/")
+ @validate_body(BodyParam("user", UserModel))
+ async def create_user_handler(req: Request):
+ user = req.body.get("user")
+ return {
+ "user": user.model_dump()
+ } # Access the validated Pydantic model instance
+
+ req = Request(
+ method="POST", path="/", body={"user": {"username": "John", "age": 30}}
+ )
+ response = await call_handler(create_user_handler, req=req)
+
+ self.assertEqual(response["user"], {"username": "John", "age": 30})
+
+ async def test_validate_missing_required_body_params(self):
+ @self.app.post("/")
+ @validate_body(BodyParam("username", str), BodyParam("age", float))
+ async def create_user_handler(req: Request):
+ username = req.body.get("username")
+ age = req.body.get("age")
+ return {"username": username, "age": age}
+
+ req = Request(method="POST", path="/", body={"username": "Amy"})
+ response = await call_handler(create_user_handler, req=req)
+
+ self.assertEqual(
+ response, ({"errors": {"age": "Missing required body parameter"}}, 400)
+ )
+
+ async def test_validate_missing_optional_body_params(self):
+ @self.app.post("/")
+ @validate_body(
+ BodyParam("username", str, required=False), BodyParam("age", float)
+ )
+ async def create_user_handler(req: Request):
+ username = req.body.get("username")
+ age = req.body.get("age")
+ return {"username": username, "age": age}
+
+ req = Request(method="POST", path="/", body={"age": 0.11})
+ response = await call_handler(create_user_handler, req=req)
+
+ self.assertEqual(response, {"username": None, "age": 0.11})
+
+ async def test_validate_invalid_body_params(self):
+ @self.app.post("/")
+ @validate_body(BodyParam("username", str), BodyParam("age", float))
+ async def create_user_handler(req: Request):
+ username = req.body.get("username")
+ age = req.body.get("age")
+ return {"username": username, "age": age}
+
+ req = Request(
+ method="POST", path="/", body={"username": "Amy", "age": "invalid"}
+ )
+ response = await call_handler(create_user_handler, req=req)
+
+ self.assertEqual(
+ response,
+ (
+ {"errors": {"age": "Invalid type for age, expected float"}},
+ 400,
+ ),
+ )
+
+
+class TestValidateQuery(unittest.IsolatedAsyncioTestCase):
+ def setUp(self):
+ # Initialize the app
+ self.app = App()
+
+ async def test_validate_simple_query_params(self):
+ @self.app.get("/")
+ @validate_query(QueryParam("username", str), QueryParam("age", float))
+ async def create_user_handler(req: Request):
+ username = req.query_params.get("username")
+ age = req.query_params.get("age")
+ return {"username": username, "age": age}
+
+ req = Request(
+ method="GET", path="/", query_params={"username": "Amy", "age": 0.11}
+ )
+ response = await call_handler(create_user_handler, req=req)
+
+ self.assertEqual(response["username"], "Amy")
+ self.assertEqual(response["age"], 0.11)
+
+ async def test_validate_pydantic_query_params(self):
+ # # Example Pydantic model
+ class UserModel(BaseModel):
+ username: str
+ age: int
+
+ @self.app.get("/")
+ @validate_query(QueryParam("user", UserModel))
+ async def create_user_handler(req: Request):
+ user = req.query_params.get("user")
+ return {"user": user.model_dump()}
+
+ req = Request(
+ method="GET",
+ path="/",
+ query_params={"user": {"username": "John", "age": 30}},
+ )
+ response = await call_handler(create_user_handler, req=req)
+
+ self.assertEqual(response["user"], {"username": "John", "age": 30})
+
+ async def test_validate_missing_required_query_params(self):
+ @self.app.get("/")
+ @validate_query(QueryParam("username", str), QueryParam("age", float))
+ async def create_user_handler(req: Request):
+ username = req.query_params.get("username")
+ age = req.query_params.get("age")
+ return {"username": username, "age": age}
+
+ req = Request(method="GET", path="/", query_params={"username": "Amy"})
+ response = await call_handler(create_user_handler, req=req)
+
+ self.assertEqual(
+ response, ({"errors": {"age": "Missing required query parameter"}}, 400)
+ )
+
+ async def test_validate_missing_optional_query_params(self):
+ @self.app.get("/")
+ @validate_query(
+ QueryParam("username", str, required=False), QueryParam("age", float)
+ )
+ async def create_user_handler(req: Request):
+ username = req.query_params.get("username")
+ age = req.query_params.get("age")
+ return {"username": username, "age": age}
+
+ req = Request(method="GET", path="/", query_params={"age": 0.11})
+ response = await call_handler(create_user_handler, req=req)
+
+ self.assertEqual(response, {"username": None, "age": 0.11})
+
+ async def test_validate_invalid_query_params(self):
+ @self.app.get("/")
+ @validate_query(QueryParam("username", str), QueryParam("age", float))
+ async def create_user_handler(req: Request):
+ username = req.query_params.get("username")
+ age = req.query_params.get("age")
+ return {"username": username, "age": age}
+
+ req = Request(
+ method="GET", path="/", query_params={"username": "Amy", "age": "invalid"}
+ )
+ response = await call_handler(create_user_handler, req=req)
+
+ self.assertEqual(
+ response,
+ (
+ {"errors": {"age": "Invalid type for age, expected float"}},
+ 400,
+ ),
+ )