patx/micropie
add pydantic example
Commit 133b6af · patx · 2026-05-20T17:56:01-04:00
Comments
No comments yet.
Diff
diff --git a/examples/pydantic/app.py b/examples/pydantic/app.py
new file mode 100644
index 0000000..3e498fa
--- /dev/null
+++ b/examples/pydantic/app.py
@@ -0,0 +1,78 @@
+"""
+Run with:
+ pip install -r requirements.txt
+ uvicorn app:app --reload
+
+Try:
+ curl -X POST http://127.0.0.1:8000/todos \
+ -H "Content-Type: application/json" \
+ -d '{"title": "Write docs", "priority": 2, "tags": ["docs"]}'
+"""
+
+from typing import Dict, List, Optional
+
+from micropie import App
+from pydantic import BaseModel, ConfigDict, Field, ValidationError
+
+
+class TodoCreate(BaseModel):
+ model_config = ConfigDict(extra="forbid")
+
+ title: str = Field(min_length=1, max_length=120)
+ priority: int = Field(default=3, ge=1, le=5)
+ tags: List[str] = Field(default_factory=list)
+
+
+class Todo(TodoCreate):
+ id: int
+ completed: bool = False
+
+
+class Root(App):
+ def __init__(self):
+ super().__init__()
+ self._todos: Dict[int, Todo] = {}
+ self._next_id = 1
+
+ async def index(self):
+ return {
+ "message": "MicroPie + pydantic example",
+ "routes": {
+ "GET /todos": "List todos",
+ "GET /todos/<id>": "Fetch one todo",
+ "POST /todos": "Create a todo from a JSON body",
+ },
+ "example": {
+ "title": "Write a MicroPie example",
+ "priority": 2,
+ "tags": ["docs", "pydantic"],
+ },
+ }
+
+ async def todos(self, todo_id: Optional[str] = None):
+ if self.request.method == "GET":
+ if todo_id is None:
+ return {"todos": [todo.model_dump() for todo in self._todos.values()]}
+
+ try:
+ todo = self._todos[int(todo_id)]
+ except (KeyError, ValueError):
+ return 404, {"error": "Todo not found"}
+
+ return todo.model_dump()
+
+ if self.request.method == "POST":
+ try:
+ todo_in = TodoCreate.model_validate(self.request.json())
+ except ValidationError as exc:
+ return 422, {"errors": exc.errors()}
+
+ todo = Todo(id=self._next_id, **todo_in.model_dump())
+ self._todos[todo.id] = todo
+ self._next_id += 1
+ return 201, todo.model_dump()
+
+ return 405, {"error": "Method not allowed"}
+
+
+app = Root() # Run with `uvicorn app:app --reload`
diff --git a/examples/pydantic/requirements.txt b/examples/pydantic/requirements.txt
new file mode 100644
index 0000000..ace99de
--- /dev/null
+++ b/examples/pydantic/requirements.txt
@@ -0,0 +1,2 @@
+micropie[all]
+pydantic>=2
diff --git a/examples/url_shortener/middlewares/rate_limit.py b/examples/url_shortener/middlewares/rate_limit.py
index 7eecd0e..1751c60 100644
--- a/examples/url_shortener/middlewares/rate_limit.py
+++ b/examples/url_shortener/middlewares/rate_limit.py
@@ -117,6 +117,7 @@ class MongoRateLimitMiddleware(HttpMiddleware):
allowed_hosts: Set[str] | None = None,
trust_proxy_headers: bool = True,
require_cf_ray: bool = True,
+ limit_methods: Iterable[str] | None = None,
# Optional: include METHOD+PATH in key (lets you set different limits by route)
bucket_by_route: bool = False,
# If we can't reliably identify the client, return 403 (recommended)
@@ -129,6 +130,9 @@ class MongoRateLimitMiddleware(HttpMiddleware):
self.allowed_hosts = set(h.lower() for h in (allowed_hosts or set()))
self.trust_proxy_headers = trust_proxy_headers
self.require_cf_ray = require_cf_ray
+ self.limit_methods = (
+ {method.upper() for method in limit_methods} if limit_methods else None
+ )
self.bucket_by_route = bucket_by_route
self.fail_closed = fail_closed
@@ -203,6 +207,14 @@ class MongoRateLimitMiddleware(HttpMiddleware):
# ---------------------------------------------------------
async def before_request(self, request):
+ method = (
+ getattr(request, "method", None)
+ or (request.scope or {}).get("method")
+ or "GET"
+ ).upper()
+ if self.limit_methods is not None and method not in self.limit_methods:
+ return None
+
client_ip = self._client_ip(request)
# Avoid collapsing unknowns into a shared key (DoS vector)