patx/micropie
replace main with development branch. MICEOPIE IS NOW ASGI BASED, RUN METHOD GONE
Commit 2519a30 · patx · 2025-01-28T12:13:04-05:00
Comments
No comments yet.
Diff
diff --git a/MicroPie.py b/MicroPie.py
index 048adcd..e422ebc 100644
--- a/MicroPie.py
+++ b/MicroPie.py
@@ -1,5 +1,5 @@
"""
-MicroPie: A simple Python ultra-micro web framework with WSGI
+MicroPie: A simple Python ultra-micro web framework with ASGI
support. https://patx.github.io/micropie
Copyright Harrison Erd
@@ -31,7 +31,6 @@ OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE,
EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
"""
-from wsgiref.simple_server import make_server
import time
import uuid
import inspect
@@ -59,266 +58,146 @@ class Server:
self.body_params: Dict[str, List[str]] = {}
self.path_params: List[str] = []
self.session: Dict[str, Any] = {}
- self.environ: Optional[Dict[str, Any]] = None
- self.start_response: Optional[Any] = None
+ self.files: Dict[str, Any] = {}
- def run(self, host: str = "127.0.0.1", port: int = 8080) -> None:
- print(f"Serving on http://{host}:{port}")
- with make_server(host, port, self.wsgi_app) as httpd:
- try:
- httpd.serve_forever()
- except KeyboardInterrupt:
- print("\nShutting down server...")
-
- def get_session(self, request_handler: Any) -> Dict[str, Any]:
- cookie = request_handler.headers.get("Cookie")
- session_id = None
-
- if cookie:
- cookies = {
- item.split("=")[0].strip(): item.split("=")[1].strip()
- for item in cookie.split(";")
- }
- session_id = cookies.get("session_id")
+ async def __call__(self, scope, receive, send):
+ await self.asgi_app(scope, receive, send)
- if not session_id or session_id not in self.sessions:
- session_id = str(uuid.uuid4())
- self.sessions[session_id] = {"last_access": time.time()}
- request_handler.send_response(200)
- request_handler.send_header(
- "Set-Cookie", f"session_id={session_id}; Path=/; HttpOnly; SameSite=Strict"
- )
- request_handler.end_headers()
+ async def asgi_app(self, scope: Dict[str, Any], receive: Any, send: Any) -> None:
+ """ASGI application entrypoint for both HTTP and WebSockets."""
- session = self.sessions.get(session_id)
- if session:
- session["last_access"] = time.time()
- else:
- session = {"last_access": time.time()}
- self.sessions[session_id] = session
+ if scope["type"] == "http":
+ self.scope = scope
+ method = scope["method"]
+ path = scope["path"].lstrip("/")
+ path_parts = path.split("/") if path else []
+ func_name = path_parts[0] if path_parts else "index"
+ self.path_params = path_parts[1:] if len(path_parts) > 1 else []
- return session
+ handler_function = getattr(self, func_name, None)
+ if not handler_function:
+ self.path_params = path_parts
+ handler_function = getattr(self, "index", None)
- def cleanup_sessions(self) -> None:
- now = time.time()
- self.sessions = {
- sid: data
- for sid, data in self.sessions.items()
- if data.get("last_access", now) + self.SESSION_TIMEOUT > now
- }
+ raw_query = scope.get("query_string", b"")
+ self.query_params = parse_qs(raw_query.decode("utf-8", "ignore"))
- def redirect(self, location: str) -> Tuple[int, str]:
- return (
- 302,
- (
- "<html><head>"
- f"<meta http-equiv='refresh' content='0;url={location}'>"
- "</head></html>"
- ),
- )
-
- def render_template(self, name: str, **kwargs: Any) -> str:
- if not JINJA_INSTALLED:
- raise ImportError("Jinja2 is not installed.")
- return self.env.get_template(name).render(kwargs)
+ headers_dict = {
+ k.decode("latin-1").lower(): v.decode("latin-1")
+ for k, v in scope.get("headers", [])
+ }
+ cookies = self._parse_cookies(headers_dict.get("cookie", ""))
- def serve_static(self, filepath: str) -> Union[Tuple[int, str], Tuple[int, bytes, List[Tuple[str, str]]]]:
- safe_root = os.path.abspath("static")
- requested_file = os.path.abspath(os.path.join("static", filepath))
- if not requested_file.startswith(safe_root):
- return 403, "403 Forbidden"
- if not os.path.isfile(requested_file):
- return 404, "404 Not Found"
- content_type, _ = mimetypes.guess_type(requested_file)
- if not content_type:
- content_type = "application/octet-stream"
- with open(requested_file, "rb") as f:
- content = f.read()
- return 200, content, [("Content-Type", content_type)]
+ session_id = cookies.get("session_id")
+ if session_id and session_id in self.sessions:
+ self.session = self.sessions[session_id]
+ self.session["last_access"] = time.time()
+ else:
+ session_id = str(uuid.uuid4())
+ self.session = {"last_access": time.time()}
+ self.sessions[session_id] = self.session
+
+ self.body_params = {}
+ self.files = {}
+ if method in ("POST", "PUT", "PATCH"):
+ body_data = bytearray()
+ while True:
+ msg = await receive()
+ if msg["type"] == "http.request":
+ body_data += msg.get("body", b"")
+ if not msg.get("more_body"):
+ break
+ content_type = headers_dict.get("content-type", "")
+ if "multipart/form-data" in content_type:
+ self.parse_multipart(bytes(body_data), content_type)
+ else:
+ body_str = body_data.decode("utf-8", "ignore")
+ self.body_params = parse_qs(body_str)
- def validate_request(self, method: str) -> bool:
- try:
- if method == "GET":
- for key, value in self.query_params.items():
- if (
- not isinstance(key, str)
- or not all(isinstance(v, str) for v in value)
- ):
- print(f"Invalid query parameter: {key} -> {value}")
- return False
-
- if method == "POST":
- for key, value in self.body_params.items():
- if (
- not isinstance(key, str)
- or not all(isinstance(v, str) for v in value)
- ):
- print(f"Invalid body parameter: {key} -> {value}")
- return False
-
- return True
- except Exception as e:
- print(f"Error during request validation: {e}")
- return False
-
- def wsgi_app(self, environ: Dict[str, Any], start_response: Any) -> List[bytes]:
- self.environ = environ
- self.start_response = start_response
-
- path = environ["PATH_INFO"].strip("/")
- method = environ["REQUEST_METHOD"]
-
- path_parts = path.split("/") if path else []
- func_name = path_parts[0] if path_parts else "index"
- self.path_params = path_parts[1:] if len(path_parts) > 1 else []
-
- handler_function = getattr(self, func_name, None)
-
- if not handler_function:
- self.path_params = path_parts
- handler_function = getattr(self, "index", None)
-
- self.query_params = parse_qs(environ.get("QUERY_STRING", ""))
-
-
- class MockRequestHandler:
- def __init__(self, environ: Dict[str, Any]) -> None:
- self.environ = environ
- self.headers = {
- key[5:].replace("_", "-").lower(): value
- for key, value in environ.items()
- if key.startswith("HTTP_")
- }
- self.cookies = self._parse_cookies()
- self._headers_to_send: List[Tuple[str, str]] = []
-
- def _parse_cookies(self) -> Dict[str, str]:
- cookies = {}
- if "HTTP_COOKIE" in self.environ:
- cookie_header = self.environ["HTTP_COOKIE"]
- for cookie in cookie_header.split(";"):
- if "=" in cookie:
- k, v = cookie.strip().split("=", 1)
- cookies[k] = v
- return cookies
-
- def send_response(self, code: int) -> None:
- pass
-
- def send_header(self, key: str, value: str) -> None:
- self._headers_to_send.append((key, value))
-
- def end_headers(self) -> None:
- pass
-
- request_handler = MockRequestHandler(environ)
-
- session_id = request_handler.cookies.get("session_id")
- if session_id and session_id in self.sessions:
- self.session = self.sessions[session_id]
- self.session["last_access"] = time.time()
- else:
- session_id = str(uuid.uuid4())
- self.session = {"last_access": time.time()}
- self.sessions[session_id] = self.session
- request_handler.send_header(
- "Set-Cookie", f"session_id={session_id}; Path=/; HttpOnly; SameSite=Strict;"
- )
+ sig = inspect.signature(handler_function)
+ func_args = []
+ for param in sig.parameters.values():
+ if self.path_params:
+ func_args.append(self.path_params.pop(0))
+ elif param.name in self.query_params:
+ func_args.append(self.query_params[param.name][0])
+ elif param.name in self.body_params:
+ func_args.append(self.body_params[param.name][0])
+ elif param.name in self.files:
+ func_args.append(self.files[param.name])
+ elif param.name in self.session:
+ func_args.append(self.session[param.name])
+ elif param.default is not param.empty:
+ func_args.append(param.default)
+ else:
+ await self._send_response(
+ send,
+ status_code=400,
+ body=f"400 Bad Request: Missing required parameter '{param.name}'",
+ )
+ return
- self.request = method
- self.body_params = {}
- self.files = {}
+ if handler_function == getattr(self, "index", None) and not func_args and path:
+ await self._send_response(send, status_code=404, body="404 Not Found")
+ return
- if method == "POST":
try:
- content_type = environ.get("CONTENT_TYPE", "")
- content_length = int(environ.get("CONTENT_LENGTH", 0) or 0)
- body = environ["wsgi.input"].read(content_length)
-
- if "multipart/form-data" in content_type:
- self.parse_multipart(body, content_type)
+ if inspect.iscoroutinefunction(handler_function):
+ result = await handler_function(*func_args)
else:
- body_str = body.decode("utf-8", "ignore")
- self.body_params = parse_qs(body_str)
+ result = handler_function(*func_args)
except Exception as e:
- start_response("400 Bad Request", [("Content-Type", "text/html")])
- return [f"400 Bad Request: {str(e)}".encode("utf-8")]
-
- sig = inspect.signature(handler_function)
- func_args = []
-
- for param in sig.parameters.values():
- if self.path_params:
- func_args.append(self.path_params.pop(0))
- elif param.name in self.query_params:
- func_args.append(self.query_params[param.name][0])
- elif param.name in self.body_params:
- func_args.append(self.body_params[param.name][0])
- elif param.name in self.files:
- func_args.append(self.files[param.name])
- elif param.name in self.session:
- func_args.append(self.session[param.name])
- elif param.default is not param.empty:
- func_args.append(param.default)
- else:
- msg = f"400 Bad Request: Missing required parameter '{param.name}'"
- start_response("400 Bad Request", [("Content-Type", "text/html")])
- return [msg.encode("utf-8")]
-
- if handler_function == getattr(self, "index", None) and not func_args and path:
- start_response("404 Not Found", [("Content-Type", "text/html")])
- return [b"404 Not Found"]
+ print(f"Error processing request: {e}")
+ await self._send_response(
+ send, status_code=500, body="500 Internal Server Error"
+ )
+ return
- try:
- response = handler_function(*func_args)
status_code = 200
- response_body = response
- extra_headers = []
-
- if isinstance(response, tuple):
- if len(response) == 2:
- status_code, response_body = response
- elif len(response) == 3:
- status_code, response_body, extra_headers = response
+ response_body = result
+ extra_headers: List[Tuple[str, str]] = []
+
+ if isinstance(result, tuple):
+ if len(result) == 2:
+ status_code, response_body = result
+ elif len(result) == 3:
+ status_code, response_body, extra_headers = result
else:
- start_response("500 Internal Server Error", [("Content-Type", "text/html")])
- return [b"500 Internal Server Error: Invalid response tuple"]
-
- status_map = {
- 206: "206 Partial Content",
- 302: "302 Found",
- 404: "404 Not Found",
- 500: "500 Internal Server Error",
- }
- status_str = status_map.get(status_code, f"{status_code} OK")
- headers = request_handler._headers_to_send
- headers.extend(extra_headers)
- if not any(h[0].lower() == "content-type" for h in headers):
- headers.append(("Content-Type", "text/html; charset=utf-8"))
-
- start_response(status_str, headers)
-
- if hasattr(response_body, "__iter__") and not isinstance(response_body, (bytes, str)):
- def byte_stream(gen: Any) -> Any:
- for chunk in gen:
- if isinstance(chunk, str):
- yield chunk.encode("utf-8")
- else:
- yield chunk
- return byte_stream(response_body)
-
- if isinstance(response_body, str):
- response_body = response_body.encode("utf-8")
-
- return [response_body]
-
- except Exception as e:
- print(f"Error processing request: {e}")
- try:
- start_response("500 Internal Server Error", [("Content-Type", "text/html")])
- except:
- pass
- return [b"500 Internal Server Error"]
+ await self._send_response(
+ send, status_code=500,
+ body="500 Internal Server Error: Invalid response tuple"
+ )
+ return
+
+ session_cookie_header = (
+ "Set-Cookie",
+ f"session_id={session_id}; Path=/; HttpOnly; SameSite=Strict"
+ )
+ has_session_cookie = any(
+ h[0].lower() == "set-cookie" and "session_id=" in h[1]
+ for h in extra_headers
+ )
+ if not has_session_cookie:
+ extra_headers.append(session_cookie_header)
+
+ await self._send_response(
+ send,
+ status_code=status_code,
+ body=response_body,
+ extra_headers=extra_headers
+ )
+ else:
+ pass
+
+ def _parse_cookies(self, cookie_header: str) -> Dict[str, str]:
+ cookies: Dict[str, str] = {}
+ if not cookie_header:
+ return cookies
+ for cookie in cookie_header.split(";"):
+ if "=" in cookie:
+ k, v = cookie.strip().split("=", 1)
+ cookies[k] = v
+ return cookies
def parse_multipart(self, body: bytes, content_type: str) -> None:
boundary = None
@@ -333,30 +212,28 @@ class Server:
raise ValueError("Boundary not found in Content-Type header.")
boundary_bytes = boundary.encode("utf-8")
- delimiter = b'--' + boundary_bytes
- end_delimiter = b'--' + boundary_bytes + b'--'
-
+ delimiter = b"--" + boundary_bytes
sections = body.split(delimiter)
for section in sections:
- if not section or section == b'--' or section == b'--\r\n':
+ if not section or section in (b"--", b"--\r\n"):
continue
- if section.startswith(b'\r\n'):
+ if section.startswith(b"\r\n"):
section = section[2:]
- if section.endswith(b'\r\n'):
+ if section.endswith(b"\r\n"):
section = section[:-2]
- if section == b'--':
+ if section == b"--":
continue
try:
- headers, content = section.split(b'\r\n\r\n', 1)
+ headers, content = section.split(b"\r\n\r\n", 1)
except ValueError:
continue
- headers = headers.decode("utf-8", "ignore").split("\r\n")
+ headers_list = headers.decode("utf-8", "ignore").split("\r\n")
header_dict = {}
- for header in headers:
- if ':' in header:
- key, value = header.split(':', 1)
+ for header_line in headers_list:
+ if ":" in header_line:
+ key, value = header_line.split(":", 1)
header_dict[key.strip().lower()] = value.strip()
disposition = header_dict.get("content-disposition", "")
@@ -364,19 +241,18 @@ class Server:
disposition_dict = {}
for disp_part in disposition_parts:
if "=" in disp_part:
- key, value = disp_part.strip().split("=", 1)
- disposition_dict[key] = value.strip('"')
+ k, v = disp_part.strip().split("=", 1)
+ disposition_dict[k] = v.strip('"')
name = disposition_dict.get("name")
filename = disposition_dict.get("filename")
if filename:
file_content_type = header_dict.get("content-type", "application/octet-stream")
- file_data = content
self.files[name] = {
- 'filename': filename,
- 'content_type': file_content_type,
- 'data': file_data
+ "filename": filename,
+ "content_type": file_content_type,
+ "data": content
}
elif name:
value = content.decode("utf-8", "ignore")
@@ -384,3 +260,130 @@ class Server:
self.body_params[name].append(value)
else:
self.body_params[name] = [value]
+
+ async def _send_response(
+ self,
+ send,
+ status_code: int,
+ body,
+ extra_headers=None
+ ):
+ if extra_headers is None:
+ extra_headers = []
+
+ # Common HTTP status text
+ status_map = {
+ 200: "200 OK",
+ 206: "206 Partial Content",
+ 302: "302 Found",
+ 403: "403 Forbidden",
+ 404: "404 Not Found",
+ 500: "500 Internal Server Error",
+ }
+ # Fallback if not in map
+ status_text = status_map.get(status_code, f"{status_code} OK")
+
+ # Ensure there's a Content-Type unless already provided
+ has_content_type = any(h[0].lower() == "content-type" for h in extra_headers)
+ if not has_content_type:
+ extra_headers.append(("Content-Type", "text/html; charset=utf-8"))
+
+ # Send the initial response start
+ await send({
+ "type": "http.response.start",
+ "status": status_code,
+ "headers": [
+ (k.encode("latin-1"), v.encode("latin-1")) for k, v in extra_headers
+ ],
+ })
+
+ # 1) Check if body is an async generator (has __aiter__)
+ if hasattr(body, "__aiter__"):
+ async for chunk in body:
+ if isinstance(chunk, str):
+ chunk = chunk.encode("utf-8")
+ await send({
+ "type": "http.response.body",
+ "body": chunk,
+ "more_body": True
+ })
+ # Send a final empty chunk to mark the end
+ await send({
+ "type": "http.response.body",
+ "body": b"",
+ "more_body": False
+ })
+ return
+
+ # 2) Check if body is a *sync* generator (has __iter__) and
+ # is not a plain string/bytes
+ if hasattr(body, "__iter__") and not isinstance(body, (bytes, str)):
+ for chunk in body:
+ if isinstance(chunk, str):
+ chunk = chunk.encode("utf-8")
+ await send({
+ "type": "http.response.body",
+ "body": chunk,
+ "more_body": True
+ })
+ # Send a final empty chunk
+ await send({
+ "type": "http.response.body",
+ "body": b"",
+ "more_body": False
+ })
+ return
+
+ if isinstance(body, str):
+ response_body = body.encode("utf-8")
+ elif isinstance(body, bytes):
+ response_body = body
+ else:
+ # Convert anything else to string then to bytes
+ response_body = str(body).encode("utf-8")
+
+ await send({
+ "type": "http.response.body",
+ "body": response_body,
+ "more_body": False
+ })
+
+ def cleanup_sessions(self) -> None:
+ now = time.time()
+ self.sessions = {
+ sid: data
+ for sid, data in self.sessions.items()
+ if data.get("last_access", now) + self.SESSION_TIMEOUT > now
+ }
+
+ def redirect(self, location: str) -> Tuple[int, str]:
+ return (
+ 302,
+ (
+ "<html><head>"
+ f"<meta http-equiv='refresh' content='0;url={location}'>"
+ "</head></html>"
+ ),
+ )
+
+ def render_template(self, name: str, **kwargs: Any) -> str:
+ if not JINJA_INSTALLED:
+ raise ImportError("Jinja2 is not installed.")
+ return self.env.get_template(name).render(kwargs)
+
+ def serve_static(
+ self, filepath: str
+ ) -> Union[Tuple[int, str], Tuple[int, bytes, List[Tuple[str, str]]]]:
+ safe_root = os.path.abspath("static")
+ requested_file = os.path.abspath(os.path.join("static", filepath))
+ if not requested_file.startswith(safe_root):
+ return 403, "403 Forbidden"
+ if not os.path.isfile(requested_file):
+ return 404, "404 Not Found"
+ content_type, _ = mimetypes.guess_type(requested_file)
+ if not content_type:
+ content_type = "application/octet-stream"
+ with open(requested_file, "rb") as f:
+ content = f.read()
+ return 200, content, [("Content-Type", content_type)]
+
diff --git a/README.md b/README.md
index cfee18e..552cdcf 100644
--- a/README.md
+++ b/README.md
@@ -2,92 +2,81 @@
## **Introduction**
-**MicroPie** is a lightweight Python web framework that makes building web applications simple and efficient. It includes features such as routing, session management, WSGI support, and Jinja2 template rendering.
+**MicroPie** is a lightweight, modern Python web framework that supports both synchronous and asynchronous web applications. Designed with flexibility and simplicity in mind, MicroPie enables you to handle high-concurrency HTTP applications with ease while allowing easy and natural integration with external tools like Socket.IO for real-time communication.
### **Key Features**
-*"Fast, efficient, and deliciously simple."*
-
-- 🚀 **Easy Setup:** Minimal configuration required. Our setup is so simple, you’ll have time for dessert.
-- 🔄 **Routing:** Maps URLs to functions automatically. So easy, even your grandma could do it (probably).
-- 🔐 **Sessions:** Simple session management using cookies.
-- 🎨 **Templates:** Jinja2 for dynamic HTML pages.
-- ⚡ **Fast & Lightweight:** No unnecessary dependencies. Life’s too short for bloated frameworks.
-- 🖥️ **WSGI support:** Deploy with WSGI servers like gunicorn making web development easy as... pie!
+- 🚀 **Async & Sync Support:** Define routes as asynchronous or synchronous functions to suit your application needs.
+- 🔄 **Routing:** Automatic mapping of URLs to functions with support for dynamic and query parameters.
+- 🔒 **Sessions:** Simple session management using cookies.
+- 🎨 **Templates:** Jinja2, if installed, for rendering dynamic HTML pages.
+- ✨ **ASGI-Powered:** Built for modern web servers like Uvicorn and Daphne, enabling high concurrency.
+- 🛠️ **Lightweight Design:** Minimal dependencies for faster development and deployment.
## **Installing MicroPie**
-### **Normal Installation**
-To install MicroPie [from the PyPI](https://pypi.org/project/MicroPie/) run the following command:
+
+### **Installation**
+Install MicroPie via pip:
```bash
pip install micropie
```
-This will install MicroPie along with `jinja2` as a dependency, enabling the built-in `render_template` method. This is the recommended way to install this framework.
+This will install MicroPie along with `jinja2` for template rendering. Jinja2 is optional but recommended for using the `render_template` method.
-### **Minimal Installation**
-If you prefer an ultra-minimalistic setup, you can run MicroPie without installing Jinja. Simply download the standalone script:
+### **Minimal Setup**
+For an ultra-minimalistic approach, download the standalone script:
[MicroPie.py](https://raw.githubusercontent.com/patx/micropie/refs/heads/main/MicroPie.py)
-Place the script in your project directory, and you're good to go. Please note you will not be able to use the `render_template` method. It will raise an `ImportError`, unless you have Jinja installed via `pip install jinja2`.
+Place it in your project directory, and your good to go. Note that Jinja2 must be installed separately to use templates, but this *is* optional:
+```bash
+pip install jinja2
+```
+
+### **Install an ASGI Web Server**
+In order to test and deploy your apps you will need a ASGI web server like uvicorn or Daphne. Install uvicorn with:
+```bash
+pip install uvicorn
+```
## **Getting Started**
-Create a basic MicroPie app in `app.py`:
+### **Create Your First ASGI App**
+Save the following as `app.py`:
```python
from MicroPie import Server
class MyApp(Server):
- def index(self, name="Guest"):
- return f"Hello, {name}!"
+ async def index(self):
+ return "Welcome to MicroPie ASGI."
-MyApp().run()
+app = MyApp()
```
-
-Run the server:
-
+Run the server with:
```bash
-python app.py
+uvicorn app:app
```
-
-Visit your app at [http://127.0.0.1:8080](http://127.0.0.1:8080).
+Access your app at [http://127.0.0.1:8000](http://127.0.0.1:8000).
## **Core Features**
-### **1. Routing**
-Define methods to handle URLs:
+### **1. Flexible Routing**
+MicroPie automatically maps URLs to methods within your `Server` class. Routes can be defined as either synchronous or asynchronous functions, offering unparalleled flexibility.
+
+#### **Basic Routing**
```python
class MyApp(Server):
def hello(self):
return "Hello, world!"
-```
-
-**Access:**
-- Basic route: [http://127.0.0.1:8080/hello](http://127.0.0.1:8080/hello)
-
-### **2. Handling GET Requests**
-
-MicroPie allows passing data using query strings (`?key=value`) and URL path segments.
-
-#### **Query Parameters**
-You can pass query parameters via the URL, which will be automatically mapped to method arguments:
-
-```python
-class MyApp(Server):
- def greet(self, name="Guest"):
- return f"Hello, {name}!"
+ async def async_hello(self):
+ return "Hello from an async route!"
```
-
**Access:**
-- Using query parameters: [http://127.0.0.1:8080/greet?name=Alice](http://127.0.0.1:8080/greet?name=Alice)
- - This will return: `Hello, Alice!`
-- Using URL path segments: [http://127.0.0.1:8080/greet](http://127.0.0.1:8080/greet)
- - This will return: `Hello, Guest!`
-
-#### **Path Parameters (Dynamic Routing)**
-
-You can also pass parameters directly in the URL path instead of query strings:
+- Sync route: [http://127.0.0.1:8000/hello](http://127.0.0.1:8000/hello)
+- Async route: [http://127.0.0.1:8000/async_hello](http://127.0.0.1:8000/async_hello)
+### **2. Query and Path Parameters**
+Pass data through query strings or URL path segments, automatically mapped to method arguments.
```python
class MyApp(Server):
def greet(self, name="Guest"):
@@ -95,242 +84,115 @@ class MyApp(Server):
```
**Access:**
-- Using path parameters: [http://127.0.0.1:8080/greet/Alice](http://127.0.0.1:8080/greet/Alice)
- - This will return: `Hello, Alice!`
-- Another example: [http://127.0.0.1:8080/greet/John](http://127.0.0.1:8080/greet/John)
- - This will return: `Hello, John!`
-
-#### **Using Both Query and Path Parameters Together**
-
-```python
-class MyApp(Server):
- def profile(self, user_id):
- age = self.query_params.get('age', ['Unknown'])[0]
- return f"User ID: {user_id}, Age: {age}"
-```
-
-**Access:**
-- [http://127.0.0.1:8080/profile/123?age=25](http://127.0.0.1:8080/profile/123?age=25)
- - Returns: `User ID: 123, Age: 25`
-- [http://127.0.0.1:8080/profile/456](http://127.0.0.1:8080/profile/456)
- - Returns: `User ID: 456, Age: Unknown`
-
-### **3. Handling POST Requests**
-
-MicroPie supports handling form data submitted via HTTP POST requests. Form data is automatically mapped to method arguments.
-
-#### **Handling Form Submission with Default Values**
-
-```python
-class MyApp(Server):
- def submit(self, username="Anonymous"):
- return f"Form submitted by: {username}"
-```
-
-#### **Accessing Raw POST Data**
-
-```python
-class MyApp(Server):
- def submit(self):
- username = self.body_params.get('username', ['Anonymous'])[0]
- return f"Submitted by: {username}"
-```
-
-#### **Handling Multiple POST Parameters**
-
-```python
-class MyApp(Server):
- def register(self):
- username = self.body_params.get('username', ['Guest'])[0]
- email = self.body_params.get('email', ['No Email'])[0]
- return f"Registered {username} with email {email}"
-```
-
-### 4. **WSGI Support**
-MicroPie includes built-in WSGI support via the wsgi_app() method, allowing you to deploy your applications with WSGI-compatible servers like Gunicorn.
-
-#### **Example**
-Create a file named app.py:
-```python
-from MicroPie import Server
-
-class MyApp(Server):
- def index(self):
- return "Hello, WSGI World!"
-
-app = MyApp()
-wsgi_application = app.wsgi_app
-```
-
-Run `app.py` with:
-```bash
-gunicorn app:wsgi_application
-```
+- [http://127.0.0.1:8000/greet?name=Alice](http://127.0.0.1:8000/greet?name=Alice) returns `Hello, Alice!`
+- [http://127.0.0.1:8000/greet/Alice](http://127.0.0.1:8000/greet/Alice) returns `Hello, Alice!`
-#### Why Use WSGI?
-WSGI (Web Server Gateway Interface) is the standard Python interface between web servers and web applications. Deploying with a WSGI server like Gunicorn provides benefits such as:
-- Better Performance: Multi-threaded and multi-process capabilities.
-- Scalability: Easily handle multiple requests concurrently.
-- Production Readiness: Designed for high-load environments.
+### **3. Real-Time Communication with Socket.IO**
+Because of its designed simplicity, MicroPie does not handle WebSockets out of the box. While the underlying ASGI interface can theoretically handle WebSocket connections, MicroPie’s routing and request-handling logic is designed primarily for HTTP. While MicroPie does not natively support WebSockets, you can easily integrate dedicated Websockets libraries like **Socket.IO** alongside Uvicorn to handle real-time, bidirectional communication. Check out [examples/socketio](https://github.com/patx/micropie/tree/development/examples/socketio) to see this in action.
-### **5. Handling Sessions**
-MicroPie has built in session handling:
-```python
-class MyApp(Server):
- def index(self):
- # Initialize or increment visit count in session
- if 'visits' not in self.session:
- self.session['visits'] = 1
- else:
- self.session['visits'] += 1
-
- return f"Welcome! You have visited this page {self.session['visits']} times."
-
-MyApp().run()
-```
-
-### **6. Jinja2 Built In**
-MicroPie has Jinja template engine built in. You can use it with the `render_template` method. You can also implement any other template engine you would like.
+### **4. Jinja2 Template Rendering**
+Dynamic HTML generation is supported via Jinja2.
#### **`app.py`**
-Save the following as `app.py`:
```python
class MyApp(Server):
def index(self):
- # Pass data to the template for rendering
return self.render_template("index.html", title="Welcome", message="Hello from MicroPie!")
-
-MyApp().run(port=8080)
```
-#### **HTML**
-In order to use the `render_template` method you must put your HTML template files in a directory at the same level as `app.py` titled `templates`. Save the following as `templates/index.html`:
+#### **`templates/index.html`**
```html
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
- <meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>{{ title }}</title>
</head>
<body>
<h1>{{ message }}</h1>
- <p>This page is rendered using Jinja2 templates.</p>
</body>
</html>
```
-### **7. Serving Static Files**
-MicroPie can serve static files (such as CSS, JavaScript, and images) from a static directory using the built in `serve_static` method. To do this you must define a route you would like to serve your static files from. For example:
+### **5. Static File Serving**
+Serve static files such as CSS, JS, and images from a `static` directory.
+
```python
-class Root(Server):
+class MyApp(Server):
def static(self, filename):
return self.serve_static(filename)
```
+To serve static files, place your files in the `static` directory and access them via `/static/<filename>`. You can define any route method handler you would like to serve static files, but for security reasons the built-in `serve_static` method will only serve files from the `static` directory.
-#### **Setup**
-- Create a directory named `static` in the same location as your MicroPie application. For saftey the `serve_static` method will only work if `filename` is in the `static` directory.
-- Place your static files (e.g., style.css, script.js, logo.png) inside the static directory.
+### **6. Streaming Responses**
+Support for streaming responses makes it easy to send data in chunks.
-#### **Accessing Static Files**
-Static files can be accessed via the `/static/` URL path. For example, if you have a file named `style.css` in the `static` directory, you can access it using:
-```
-http://127.0.0.1:8080/static/style.css
-```
-
-### **8. Streaming Responses and WebSockets**
-
-#### **Streaming Response Support**
-MicroPie provides support for streaming responses, allowing you to send data to the client in chunks instead of all at once. This is particularly useful for scenarios where data is generated or processed over time, such as live feeds, large file downloads, or incremental data generation.
-
-#### How It Works
-Streaming is supported through the `wsgi_app` method, making it compatible with WSGI servers like **Gunicorn** or the built in `run` method. When a route handler returns a generator or an iterable (excluding strings), MicroPie automatically streams the response to the client.
-
-With the following saved as `app.py`:
```python
-import time
-from MicroPie import Server
-
-class Root(Server):
-
- def index(self):
- def generator():
+class MyApp(Server):
+ async def stream(self):
+ async def generator():
for i in range(1, 6):
yield f"Chunk {i}\n"
- time.sleep(1) # Simulate slow processing or data generation
return generator()
-
-app = Root()
-wsgi_app = app.wsgi_app
```
-Run your application with `gunicorn app:wsgi_app`. For best performance with streaming, consider tuning Gunicorn settings such as worker types (e.g., `--worker-class gevent`) to handle long-lived connections efficiently.
-
-#### **WebSockets Integration**
-MicroPie applications can seamlessly integrate **WebSockets** by running a separate WebSocket server using Python’s `websockets` library. This enables real-time, bidirectional communication between clients and the server, independent of the HTTP server being used. To get started with WebSockets in MicroPie, ensure you have the `websockets` package installed `pip install websockets`.
-#### How It Works
+### **7. Sessions and Cookies**
+Built-in session handling simplifies state management:
-- **The HTTP server (MicroPie)** handles regular web requests and serves the frontend.
-- **A separate WebSocket server** runs concurrently to handle real-time communication.
-- Clients connect to the WebSocket server via the frontend and exchange messages asynchronously.
-- **Threading Considerations:** Since the WebSocket server runs in a separate thread, developers should handle shared resources carefully to avoid concurrency issues.
-- **Port Management:** The WebSocket server must run on a different port than the HTTP server to avoid conflicts.
-- **Client Compatibility:** Ensure that clients support WebSockets when implementing features relying on real-time communication.
-
-**For a full example showing `websockets` and MicroPie check out the [chatroom example](https://github.com/patx/micropie/tree/main/examples/chatroom).**
-
-## **API Reference**
-
-### Class: Server
-
-#### run(host='127.0.0.1', port=8080)
-Starts the WSGI server with the specified host and port.
-
-#### get_session(request_handler)
-Retrieves or creates a session for the current request. Sessions are managed via cookies.
+```python
+class MyApp(Server):
+ def index(self):
+ if "visits" not in self.session:
+ self.session["visits"] = 1
+ else:
+ self.session["visits"] += 1
+ return f"You have visited {self.session['visits']} times."
+```
-#### cleanup_sessions()
-Removes expired sessions that have surpassed the timeout period.
+### **8. Deployment**
+MicroPie ASGI apps can be deployed using any ASGI server. For example, using Uvicorn:
+```bash
+uvicorn app:MyApp --workers 4 --port 8000
+```
-#### redirect(location)
-Returns a 302 redirect response to the specified URL.
-#### render_template(name, **args)
-Renders a Jinja2 template with provided context variables.
+## **Learn by Examples**
+Check out the [examples folder](https://github.com/patx/micropie/tree/development/examples) for more advanced usage, including:
+- Template rendering
+- Custom HTTP request handling
+- File uploads
+- Serving static content
+- Session usage
+- Websockets with Socket.io
+- Async Streaming
+- Form handling
-#### validate_request(method)
-Validates incoming requests for both GET and POST methods based on query and body parameters.
-#### wsgi_app(environ, start_response)
-WSGI-compliant method for parsing requests and returning responses. Ideal for production deployment using WSGI servers.
+## **Why ASGI?**
+ASGI is the future of Python web development, offering:
+- **Concurrency**: Handle thousands of simultaneous connections efficiently.
+- **WebSockets**: Use tools like Socket.IO for real-time communication.
+- **Scalability**: Ideal for modern, high-traffic applications.
-#### serve_static(filename)
-Serve static files from the `static` directory.
+MicroPie ASGI allows you to take full advantage of these benefits while maintaining simplicity and ease of use your used to with your WSGI apps.
-## **Examples**
-Check out the [examples folder](https://github.com/patx/micropie/tree/main/examples) for more advanced usage, including template rendering, custom HTTP request handling, file uploads, session usage, websockets, streaming and form handling.
## **Feature Comparison**
-| Feature | MicroPie | Flask | CherryPy | Bottle | Django | FastAPI |
-|---------------------|-----------|------------|-----------|------------|-------------------|------------|
-| **Ease of Use** | Very Easy | Easy | Easy | Easy | Moderate | Moderate |
-| **Routing** | Automatic | Manual | Manual | Manual | Automatic | Automatic |
-| **Template Engine** | Jinja2 | Jinja2 | None | SimpleTpl | Django Templating | Jinja2 |
-| **Session Handling**| Built-in | Extension | Built-in | Plugin | Built-in | Extension |
-| **Request Handling**| Simple | Flexible | Advanced | Simple | Advanced | Advanced |
-| **Performance** | High [^1] | High | Moderate | High | Moderate | Very High |
-| **WSGI Support** | Yes | Yes | Yes | Yes | Yes | No (ASGI) |
-| **Async Support** | No | No (Quart) | No | No | Limited | Yes |
-| **Deployment** | Simple | Moderate | Moderate | Simple | Complex | Moderate |
+| Feature | MicroPie | Flask | CherryPy | Bottle | Django | FastAPI |
+|---------------------|---------------|--------------|------------|--------------|--------------|-----------------|
+| **Ease of Use** | Very Easy | Easy | Easy | Easy | Moderate | Moderate |
+| **Routing** | Automatic | Manual | Manual | Manual | Automatic | Automatic |
+| **Template Engine** | Jinja2 (Opt.) | Jinja2 | None | SimpleTpl | Django Templating | Jinja2 |
+| **Session Handling**| Simple | Extension | Built-in | Plugin | Built-in | Extension |
+| **Async Support** | Yes | No (Quart) | No | No | Limited | Yes |
+| **Performance** | Very High | High | Moderate | High | Moderate | Extremely High |
+| **Built-in Server** | No | No | Yes | Yes | Yes | No |
-[^1]: *Note that while MicroPie is high-performing for lightweight applications, it may not scale well for complex, high-traffic web applications due to the lack of advanced features such as asynchronous request handling and database connection pooling, which are found in frameworks like Django and Flask. To achieve similar performance with MicroPie use `gunicorn` with `gevent`.*
## **Suggestions or Feedback?**
We welcome suggestions, bug reports, and pull requests!
- File issues or feature requests [here](https://github.com/patx/micropie/issues).
-### **Why not ASGI? And Future-Proofing**
-If your looking for async and websockets support with ASGI out of the box, we are working on that! Check out the [development branch](https://github.com/patx/micropie/tree/development). Please note in order to run your apps you will need an external ASGI server like `uvicorn`.
diff --git a/docs/index.html b/docs/index.html
index 7655944..d3ebd21 100644
--- a/docs/index.html
+++ b/docs/index.html
@@ -5,7 +5,7 @@
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<link rel="shortcut icon" href="https://cherrypy.dev/images/favicon.gif" type="image/x-icon">
<link href="https://fonts.googleapis.com/css2?family=Poppins:wght@300;400;600&display=swap" rel="stylesheet">
- <title>MicroPie - An ultra-micro Python web framework</title>
+ <title>MicroPie - An ultra-micro Python ASGI web framework</title>
<style>
body {
font-family: 'Poppins', sans-serif;
@@ -107,7 +107,7 @@
<div class="logo">
<img src="logo.png" alt="MicroPie logo">
</div>
- <p><strong>MicroPie is an ultra-lightweight Python web framework</strong> that gets out of your way, letting you build fast and dynamic web apps with ease.
+ <p><strong>MicroPie is an ultra-small ASGI Python web framework</strong> that gets out of your way, letting you build fast and dynamic web apps with ease.
Inspired by <a href="https://cherrypy.dev/">CherryPy</a> and licensed under the BSD three-clause license.</p>
<h2>MicroPie is Fun</h2>
@@ -116,10 +116,10 @@
<span class="c2">class</span> MyApp(<span class="c9">Server</span>):
- <span class="c2">def</span> index(<span class="c9">self</span>):
+ <span class="c2">async def</span> index(<span class="c9">self</span>):
return <span class="c9">'Hello World!'</span>
-MyApp().run()
+app = MyApp() <small><em># Run with `uvicorn app:app`</em></small>
</code></pre>
<h2> And Easy to Install</h2>
diff --git a/examples/file_uploads/app.py b/examples/file_uploads/app.py
index afd4cd6..ba17aaf 100644
--- a/examples/file_uploads/app.py
+++ b/examples/file_uploads/app.py
@@ -1,6 +1,7 @@
from MicroPie import Server
import os, uuid
+
class Root(Server):
def upload_file(self, file):
@@ -55,4 +56,6 @@ class Root(Server):
"</html>"
)
-Root().run()
+
+
+app = Root()
diff --git a/examples/headers/app.py b/examples/headers/app.py
index ccc3811..34ccae6 100644
--- a/examples/headers/app.py
+++ b/examples/headers/app.py
@@ -1,5 +1,6 @@
from MicroPie import Server
+
class Root(Server):
def index(self):
headers = [
@@ -12,8 +13,6 @@ class Root(Server):
]
return 200, "hello world", headers
-app = Root()
-wsgi_app = app.wsgi_app
-if __name__ == "__main__":
- app.run()
+
+app = Root()
diff --git a/examples/hello_world/app.py b/examples/hello_world/app.py
new file mode 100644
index 0000000..1ded149
--- /dev/null
+++ b/examples/hello_world/app.py
@@ -0,0 +1,11 @@
+from MicroPie import Server
+
+
+class Root(Server):
+
+ def index(self, name=None):
+ if name:
+ return f'Hello {name}'
+ return 'Hello ASGI World!'
+
+app = Root() # Run with `uvicorn app:app`
diff --git a/examples/pastebin/app.py b/examples/pastebin/app.py
index ba6e8a2..97ca3af 100644
--- a/examples/pastebin/app.py
+++ b/examples/pastebin/app.py
@@ -1,6 +1,6 @@
# -*- coding: utf-8 -*-
"""
- A simple no frills pastebin using MicroPie, pickleDB, and highlight.js.
+A simple no frills pastebin using MicroPie, pickleDB, and highlight.js.
"""
from uuid import uuid4
@@ -9,14 +9,13 @@ from MicroPie import Server
from pickledb import PickleDB
from markupsafe import escape
-
db = PickleDB('pastes.db')
-
class Root(Server):
- def index(self):
- if self.request == 'POST':
+ async def index(self):
+ # Check the HTTP method from the ASGI scope
+ if self.scope["method"] == "POST":
paste_content = self.body_params.get('paste_content', [''])[0]
pid = str(uuid4())
db.set(pid, escape(paste_content))
@@ -24,21 +23,16 @@ class Root(Server):
return self.redirect(f'/paste/{pid}')
return self.render_template('index.html')
- def paste(self, paste_id, delete=None):
+ async def paste(self, paste_id, delete=None):
if delete == 'delete':
db.remove(paste_id)
db.save()
return self.redirect('/')
- return self.render_template('paste.html', paste_id=paste_id,
- paste_content=db.get(paste_id))
+ return self.render_template(
+ 'paste.html',
+ paste_id=paste_id,
+ paste_content=db.get(paste_id)
+ )
-# Create a instance of our MicroPie App
app = Root()
-
-# Run with `gunicorn app:wsgi_app`
-wsgi_app = app.wsgi_app
-
-# Run with `python3 app.py`
-if __name__ == '__main__':
- app.run()
diff --git a/examples/requests/app.py b/examples/requests/app.py
index 5ca8baf..3fda164 100644
--- a/examples/requests/app.py
+++ b/examples/requests/app.py
@@ -28,7 +28,7 @@ def options_handler():
]
# Define the custom server application
-class MyApp(Server):
+class Root(Server):
# Handle root URL requests and delegate based on HTTP method
def index(self):
@@ -47,8 +47,8 @@ class MyApp(Server):
}
# Check if the request method is supported and call the handler
- if self.request in method_map:
- response = method_map[self.request]()
+ if self.scope['method'] in method_map:
+ response = method_map[self.scope['method']]()
# Ensure response is formatted correctly for WSGI
if isinstance(response, tuple):
@@ -61,8 +61,6 @@ class MyApp(Server):
# Return 405 if the request method is not supported
return 405, b"405 Method Not Allowed", [("Content-Type", "text/html")]
-# Run the web server
-if __name__ == '__main__':
- app = MyApp()
- app.run()
+
+app = Root()
diff --git a/examples/socketio/chatroom.py b/examples/socketio/chatroom.py
new file mode 100644
index 0000000..da59e21
--- /dev/null
+++ b/examples/socketio/chatroom.py
@@ -0,0 +1,60 @@
+import socketio
+from MicroPie import Server
+
+# Create a Socket.IO server with CORS support
+sio = socketio.AsyncServer(async_mode="asgi", cors_allowed_origins="*") # Allow all origins
+
+# Create the MicroPie server
+class MyApp(Server):
+ def index(self):
+ return """
+ <html>
+ <head>
+ <script src="https://cdn.socket.io/4.5.4/socket.io.min.js"></script>
+ <script>
+ var socket = io("http://localhost:8000");
+ socket.on("connect", function() {
+ console.log("Connected to Socket.IO server");
+ });
+ socket.on("message", function(data) {
+ document.getElementById("output").innerHTML += data + "<br>";
+ });
+ function sendMessage() {
+ var message = document.getElementById("message").value;
+ socket.send(message);
+ document.getElementById("message").value = ""; // Clear input after sending
+ }
+ window.onbeforeunload = function() {
+ socket.disconnect();
+ };
+ </script>
+ </head>
+ <body>
+ <h1>Socket.IO Chat</h1>
+ <input type="text" id="message" placeholder="Type a message">
+ <button onclick="sendMessage()">Send</button>
+ <div id="output"></div>
+ </body>
+ </html>
+ """
+
+# Socket.IO event handlers
[email protected]
+async def connect(sid, environ):
+ print(f"Client connected: {sid}")
+
[email protected]
+async def disconnect(sid):
+ print(f"Client disconnected: {sid}")
+
[email protected]
+async def message(sid, data):
+ print(f"Received message from {sid}: {data}")
+ # Broadcast the message to all connected clients
+ await sio.emit("message", f"User: {data}", room=None)
+
+
+
+# Attach Socket.IO to the ASGI app
+app = MyApp()
+asgi_app = socketio.ASGIApp(sio, app)
diff --git a/examples/socketio/templates/index_stream.html b/examples/socketio/templates/index_stream.html
new file mode 100644
index 0000000..b710858
--- /dev/null
+++ b/examples/socketio/templates/index_stream.html
@@ -0,0 +1,17 @@
+<!DOCTYPE html>
+<html lang="en">
+<head>
+ <meta charset="UTF-8">
+ <meta name="viewport" content="width=device-width, initial-scale=1.0">
+ <title>Webcam Streaming</title>
+</head>
+<body>
+ <h1>Webcam Streaming App</h1>
+ <form method="post" action="/submit">
+ <input type="text" name="username" placeholder="Enter username" required>
+ <button type="submit" name="action" value="Start Streaming">Start Streaming</button>
+ <button type="submit" name="action" value="Watch Stream">Watch Stream</button>
+ </form>
+</body>
+</html>
+
diff --git a/examples/socketio/templates/stream.html b/examples/socketio/templates/stream.html
new file mode 100644
index 0000000..b4f4e37
--- /dev/null
+++ b/examples/socketio/templates/stream.html
@@ -0,0 +1,57 @@
+<!DOCTYPE html>
+<html lang="en">
+<head>
+ <meta charset="UTF-8" />
+ <meta name="viewport" content="width=device-width, initial-scale=1.0"/>
+ <script src="https://cdn.socket.io/4.5.4/socket.io.min.js"></script>
+ <title>Streaming: {{ username }}</title>
+</head>
+<body>
+ <h1>Streaming as {{ username }}</h1>
+ <video id="webcam" autoplay playsinline></video>
+
+<script>
+ const username = "{{ username }}";
+ const socket = io();
+
+ // Join the room as a streamer
+ socket.emit("join_room", { username });
+
+ socket.on("connect", () => {
+ console.log("Connected as streamer for", username);
+ startWebcam();
+ });
+
+ socket.on("disconnect", () => {
+ console.log("Disconnected");
+ });
+
+ async function startWebcam() {
+ try {
+ const stream = await navigator.mediaDevices.getUserMedia({ video: true });
+ const videoElement = document.getElementById("webcam");
+ videoElement.srcObject = stream;
+
+ const canvas = document.createElement("canvas");
+ const context = canvas.getContext("2d");
+ const track = stream.getVideoTracks()[0];
+ const settings = track.getSettings();
+ canvas.width = settings.width || 640;
+ canvas.height = settings.height || 480;
+
+ setInterval(() => {
+ context.drawImage(videoElement, 0, 0, canvas.width, canvas.height);
+ const frameDataUrl = canvas.toDataURL("image/webp");
+
+ // Send frame to server
+ socket.emit("stream_frame", { username, frame: frameDataUrl });
+ }, 100);
+ } catch (err) {
+ console.error("Error accessing webcam:", err);
+ }
+ }
+</script>
+
+</body>
+</html>
+
diff --git a/examples/socketio/templates/watch.html b/examples/socketio/templates/watch.html
new file mode 100644
index 0000000..e4f432d
--- /dev/null
+++ b/examples/socketio/templates/watch.html
@@ -0,0 +1,37 @@
+<!DOCTYPE html>
+<html lang="en">
+<head>
+ <meta charset="UTF-8" />
+ <meta name="viewport" content="width=device-width, initial-scale=1.0"/>
+ <script src="https://cdn.socket.io/4.5.4/socket.io.min.js"></script>
+ <title>Watching: {{ username }}</title>
+</head>
+<body>
+ <h1>Watching Stream of {{ username }}</h1>
+ <img id="videoFeed" style="width: 100%; max-width: 800px;" />
+
+
+<script>
+ const username = "{{ username }}";
+ const socket = io();
+
+ // Join the room as a watcher
+ socket.emit("join_room", { username });
+
+ socket.on("connect", () => {
+ console.log("Connected as watcher for", username);
+ });
+
+ socket.on("video_frame", (data) => {
+ if (data.username === username) {
+ document.getElementById("videoFeed").src = data.frame;
+ }
+ });
+
+ socket.on("disconnect", () => {
+ console.log("Disconnected");
+ });
+</script>
+
+</body>
+</html>
diff --git a/examples/socketio/webcam.py b/examples/socketio/webcam.py
new file mode 100644
index 0000000..a094793
--- /dev/null
+++ b/examples/socketio/webcam.py
@@ -0,0 +1,64 @@
+import socketio
+from MicroPie import Server
+
+# Create the Socket.IO server
+sio = socketio.AsyncServer(async_mode="asgi")
+
+# Track active users and their watchers/streamers
+active_users = set()
+
+# MicroPie Server with integrated Socket.IO
+class MyApp(Server):
+ async def index(self):
+ return self.render_template("index_stream.html")
+
+ async def submit(self, username: str, action: str):
+ if username:
+ active_users.add(username)
+ route = f"/stream/{username}" if action == "Start Streaming" else f"/watch/{username}"
+ return self.redirect(route)
+ return self.redirect("/")
+
+ async def stream(self, username: str):
+ return self.render_template("stream.html", username=username) if username in active_users else self.redirect("/")
+
+ async def watch(self, username: str):
+ return self.render_template("watch.html", username=username) if username in active_users else self.redirect("/")
+
+# Socket.IO event handlers
[email protected]
+async def connect(sid, environ):
+ print(f"Client connected: {sid}")
+
[email protected]
+async def disconnect(sid):
+ print(f"Client disconnected: {sid}")
+
[email protected]("stream_frame")
+async def handle_stream_frame(sid, data):
+ """Broadcast the streamed frame to all watchers."""
+ username = data.get("username")
+ frame = data.get("frame")
+ if username in active_users:
+ # Emit the frame to watchers
+ await sio.emit("video_frame", {"username": username, "frame": frame}, room=username)
+
[email protected]("join_room")
+async def join_room(sid, data):
+ """Add a client to a room (either as a streamer or watcher)."""
+ username = data.get("username")
+ if username in active_users:
+ await sio.enter_room(sid, username) # Await the method
+ print(f"{sid} joined room for {username}")
+
[email protected]("leave_room")
+async def leave_room(sid, data):
+ """Remove a client from a room."""
+ username = data.get("username")
+ if username in active_users:
+ sio.leave_room(sid, username)
+ print(f"{sid} left room for {username}")
+
+# Attach the Socket.IO server to the ASGI app
+app = MyApp()
+asgi_app = socketio.ASGIApp(sio, app)
diff --git a/examples/static_content/app.py b/examples/static_content/app.py
new file mode 100644
index 0000000..a83fde6
--- /dev/null
+++ b/examples/static_content/app.py
@@ -0,0 +1,9 @@
+from MicroPie import Server
+
+class Root(Server):
+
+ def static(self, filename):
+ return self.serve_static(filename)
+
+
+app = Root()
diff --git a/examples/static_content/static/logo.png b/examples/static_content/static/logo.png
new file mode 100644
index 0000000..20eb33f
Binary files /dev/null and b/examples/static_content/static/logo.png differ
diff --git a/examples/streaming/text.py b/examples/streaming/text.py
index 0fd7b66..096aad8 100644
--- a/examples/streaming/text.py
+++ b/examples/streaming/text.py
@@ -15,5 +15,6 @@ class Root(Server):
time.sleep(1) # simulate slow processing or data generation
return generator()
+
+
app = Root()
-app.run()
diff --git a/examples/streaming/video.py b/examples/streaming/video.py
new file mode 100644
index 0000000..fe4601d
--- /dev/null
+++ b/examples/streaming/video.py
@@ -0,0 +1,74 @@
+import os
+from MicroPie import Server
+
+VIDEO_PATH = "video.mp4"
+
+class Root(Server):
+ def index(self):
+ return '''
+ <html>
+ <body>
+ <center>
+ <video width="640" height="360" controls>
+ <source src="/stream" type="video/mp4">
+ Your browser does not support the video tag. Use Chrome for best results.
+ </video>
+ </center>
+ </body>
+ </html>
+ '''
+
+ async def stream(self):
+ headers = {
+ k.decode('latin-1').lower(): v.decode('latin-1')
+ for k, v in self.scope.get('headers', [])
+ }
+ range_header = headers.get('range')
+ file_size = os.path.getsize(VIDEO_PATH)
+
+ # Decide on start/end
+ start, end = 0, file_size - 1
+ status_code = 200
+ extra_headers = [
+ ("Accept-Ranges", "bytes"),
+ ("Content-Type", "video/mp4"),
+ ]
+
+ if range_header:
+ # e.g. "bytes=1234-" or "bytes=1234-5678"
+ try:
+ byte_range = range_header.replace("bytes=", "")
+ start_str, end_str = byte_range.split("-")
+ start = int(start_str) if start_str else 0
+ end = int(end_str) if end_str else file_size - 1
+ if start >= file_size or end >= file_size:
+ start, end = 0, file_size - 1
+
+ content_length = end - start + 1
+ extra_headers += [
+ ("Content-Range", f"bytes {start}-{end}/{file_size}"),
+ ("Content-Length", str(content_length)),
+ ]
+ status_code = 206
+ except ValueError:
+ # Malformed range; fallback
+ pass
+ else:
+ # Full content
+ extra_headers.append(("Content-Length", str(file_size)))
+
+ # Make an async generator that yields file chunks
+ async def file_chunk_generator(start_pos, end_pos, chunk_size=1024 * 1024):
+ with open(VIDEO_PATH, "rb") as f:
+ f.seek(start_pos)
+ remaining = (end_pos + 1) - start_pos
+ while remaining > 0:
+ data = f.read(min(chunk_size, remaining))
+ if not data:
+ break
+ yield data
+ remaining -= len(data)
+
+ return (status_code, file_chunk_generator(start, end), extra_headers)
+
+app = Root()
diff --git a/examples/twutr/twutr.py b/examples/twutr/twutr.py
index 83a71da..c8ba804 100644
--- a/examples/twutr/twutr.py
+++ b/examples/twutr/twutr.py
@@ -260,7 +260,7 @@ class Twutr(Server):
if not self.session.get('logged_in'):
return self.redirect('/login')
- if self.request == 'POST':
+ if self.scope['method'] == 'POST':
message = self.body_params.get('message', [''])[0]
# Convert @link syntax and escape everything else
@@ -284,7 +284,7 @@ class Twutr(Server):
if self.session.get('logged_in'):
return self.redirect('/')
- if self.request == 'POST':
+ if self.scope['method'] == 'POST':
username = escape(self.body_params.get('username', [''])[0].strip())
password = escape(self.body_params.get('password', [''])[0].strip())
@@ -306,7 +306,7 @@ class Twutr(Server):
if self.session.get('logged_in'):
return self.redirect('/')
- if self.request == 'POST':
+ if self.scope['method'] == 'POST':
username = escape(self.body_params.get('username', [''])[0].strip())
password = escape(self.body_params.get('password', [''])[0].strip())
@@ -333,9 +333,6 @@ class Twutr(Server):
self.session.clear()
return self.redirect('/public')
-app = Twutr()
-wsgi_app = app.wsgi_app
-if __name__ == "__main__":
- app.run()
+app = Twutr()
diff --git a/setup.py b/setup.py
index c861616..67cd881 100644
--- a/setup.py
+++ b/setup.py
@@ -5,14 +5,14 @@ MicroPie is Fun
::
- from MicroPie import Server
+ from MicroPie import
class MyApp(Server):
def index(self):
return 'Hello world!'
- MyApp().run()
+ app = MyApp() # Run with `uvicorn app:app`
Links
@@ -25,7 +25,7 @@ Links
from distutils.core import setup
setup(name="MicroPie",
- version="0.7",
+ version="0.8",
description="A ultra micro web framework w/ Jinja2.",
long_description=__doc__,
author="Harrison Erd",
diff --git a/tests.py b/tests.py
index c988a36..04c0537 100644
--- a/tests.py
+++ b/tests.py
@@ -1,230 +1,113 @@
import unittest
-import uuid
-import time
-from io import BytesIO
-from unittest.mock import patch
+from unittest.mock import AsyncMock, MagicMock, patch
from MicroPie import Server
+import os
+import uuid
-
-class TestMicroPie(unittest.TestCase):
+class TestServer(unittest.TestCase):
def setUp(self):
- """
- Create a fresh instance of the Server for each test and define some
- sample endpoints on the fly.
- """
self.server = Server()
- # Define a simple default endpoint
- def index():
- return "Hello from index!"
-
- # Define an endpoint that greets the user by name
- def greet(name="World"):
- return f"Hello, {name}!"
-
- # Define an endpoint that triggers a redirect
- def go_away():
- return self.server.redirect("/gone")
-
- # Define an endpoint for testing template rendering (requires a
- # templates/greet.html file in your project for a real test).
- def greet_template(name="World"):
- return self.server.render_template("greet.html", name=name)
-
- # Attach the functions to the server instance
- self.server.index = index
- self.server.greet = greet
- self.server.go_away = go_away
- self.server.greet_template = greet_template
-
- def _start_response(self, status, headers):
- """
- Helper method to capture the status and headers from wsgi_app calls.
- """
- self._wsgi_status = status
- self._wsgi_headers = headers
-
- def test_index_get(self):
- """
- Ensure that accessing '/' via GET calls the 'index' function and
- returns the correct body.
- """
- environ = {
- "REQUEST_METHOD": "GET",
- "PATH_INFO": "/", # Access root
- "QUERY_STRING": "",
- "wsgi.input": BytesIO(b""),
- "CONTENT_LENGTH": "0",
- }
- response = self.server.wsgi_app(environ, self._start_response)
- body = b"".join(response).decode()
-
- self.assertEqual(self._wsgi_status, "200 OK")
- self.assertIn("Hello from index!", body)
-
- def test_custom_endpoint_get(self):
- """
- Ensure that a custom endpoint (greet) can accept query parameters
- through GET and return the correct response.
- """
- environ = {
- "REQUEST_METHOD": "GET",
- "PATH_INFO": "/greet",
- "QUERY_STRING": "name=Alice",
- "wsgi.input": BytesIO(b""),
- "CONTENT_LENGTH": "0",
- }
- response = self.server.wsgi_app(environ, self._start_response)
- body = b"".join(response).decode()
-
- self.assertEqual(self._wsgi_status, "200 OK")
- self.assertIn("Hello, Alice!", body)
-
- def test_custom_endpoint_path_param(self):
- """
- Test that path parameters are used if present. For example, accessing
- '/greet/Bob' should call greet("Bob").
- """
- environ = {
- "REQUEST_METHOD": "GET",
- "PATH_INFO": "/greet/Bob",
- "QUERY_STRING": "",
- "wsgi.input": BytesIO(b""),
- "CONTENT_LENGTH": "0",
- }
- response = self.server.wsgi_app(environ, self._start_response)
- body = b"".join(response).decode()
-
- self.assertEqual(self._wsgi_status, "200 OK")
- self.assertIn("Hello, Bob!", body)
-
- def test_endpoint_not_found(self):
- """
- Access a path that does not map to a defined method. Expect 404.
- """
- environ = {
- "REQUEST_METHOD": "GET",
- "PATH_INFO": "/does_not_exist",
- "QUERY_STRING": "",
- "wsgi.input": BytesIO(b""),
- "CONTENT_LENGTH": "0",
- }
- response = self.server.wsgi_app(environ, self._start_response)
- body = b"".join(response).decode()
-
- self.assertEqual(self._wsgi_status, "404 Not Found")
- self.assertIn("404 Not Found", body)
-
- def test_post_request(self):
- """
- Test handling of a POST request with form data in the body.
- """
- post_body = "name=Charlie"
- environ = {
- "REQUEST_METHOD": "POST",
- "PATH_INFO": "/greet",
- "QUERY_STRING": "",
- "wsgi.input": BytesIO(post_body.encode("utf-8")),
- "CONTENT_LENGTH": str(len(post_body)),
- }
- response = self.server.wsgi_app(environ, self._start_response)
- body = b"".join(response).decode()
+ def test_parse_cookies(self):
+ cookie_header = "session_id=abc123; theme=dark"
+ cookies = self.server._parse_cookies(cookie_header)
+ self.assertEqual(cookies, {"session_id": "abc123", "theme": "dark"})
- self.assertEqual(self._wsgi_status, "200 OK")
- self.assertIn("Hello, Charlie!", body)
+ def test_parse_cookies_empty(self):
+ cookies = self.server._parse_cookies("")
+ self.assertEqual(cookies, {})
def test_redirect(self):
- """
- Test that an endpoint can return a 302 redirect.
- """
- environ = {
- "REQUEST_METHOD": "GET",
- "PATH_INFO": "/go_away", # Calls the go_away function
- "QUERY_STRING": "",
- "wsgi.input": BytesIO(b""),
- "CONTENT_LENGTH": "0",
- }
- response = self.server.wsgi_app(environ, self._start_response)
- body = b"".join(response).decode()
-
- # MicroPie returns (302, <html>...) so we expect status to be "302 Found"
- self.assertEqual(self._wsgi_status, "302 Found")
- self.assertIn("url=/gone", body) # Basic check that the redirect body is correct
-
- def test_session_creation(self):
- """
- Test that a session is created if no session cookie is present.
- """
- environ = {
- "REQUEST_METHOD": "GET",
- "PATH_INFO": "/",
- "QUERY_STRING": "",
- "wsgi.input": BytesIO(b""),
- "CONTENT_LENGTH": "0",
- # No HTTP_COOKIE -> expect new session
+ location = "/new-path"
+ status, body = self.server.redirect(location)
+ self.assertEqual(status, 302)
+ self.assertIn(location, body)
+
+ @patch("os.path.isfile", return_value=True)
+ @patch("builtins.open", new_callable=MagicMock)
+ def test_serve_static_file(self, mock_open, mock_isfile):
+ mock_open.return_value.__enter__.return_value.read.return_value = b"file content"
+ response = self.server.serve_static("test.txt")
+ self.assertEqual(response[0], 200)
+ self.assertEqual(response[1], b"file content")
+ self.assertEqual(response[2][0][0], "Content-Type")
+
+ @patch("os.path.isfile", return_value=False)
+ def test_serve_static_file_not_found(self, mock_isfile):
+ response = self.server.serve_static("missing.txt")
+ self.assertEqual(response, (404, "404 Not Found"))
+
+ def test_cleanup_sessions(self):
+ self.server.sessions = {
+ "session1": {"last_access": time.time() - 1000},
+ "session2": {"last_access": time.time() - 10000},
}
- response = self.server.wsgi_app(environ, self._start_response)
- _ = b"".join(response).decode()
-
- # Find the Set-Cookie header among the WSGI headers
- set_cookie_headers = [h for h in self._wsgi_headers if h[0] == 'Set-Cookie']
- self.assertTrue(set_cookie_headers, "Expected a Set-Cookie header for new session.")
-
- # The session should be stored in self.server.sessions
- cookie_value = set_cookie_headers[0][1]
- session_id = cookie_value.split("=")[1].split(";")[0]
- self.assertIn(session_id, self.server.sessions)
-
- def test_session_usage(self):
- """
- Test that an existing session is reused if a valid session_id is provided.
- """
- # Create a session manually
- session_id = str(uuid.uuid4())
- self.server.sessions[session_id] = {"last_access": time.time()}
-
- environ = {
- "REQUEST_METHOD": "GET",
- "PATH_INFO": "/",
- "QUERY_STRING": "",
- "wsgi.input": BytesIO(b""),
- "CONTENT_LENGTH": "0",
- "HTTP_COOKIE": f"session_id={session_id}",
+ self.server.SESSION_TIMEOUT = 3600
+ self.server.cleanup_sessions()
+ self.assertEqual(len(self.server.sessions), 1)
+ self.assertIn("session1", self.server.sessions)
+
+ @patch("uuid.uuid4", return_value="test-session-id")
+ @patch("time.time", return_value=1000)
+ async def test_asgi_app_creates_session(self, mock_time, mock_uuid):
+ mock_send = AsyncMock()
+ mock_receive = AsyncMock(return_value={"type": "http.request", "body": b""})
+
+ scope = {
+ "type": "http",
+ "method": "GET",
+ "path": "/",
+ "headers": [],
}
- response = self.server.wsgi_app(environ, self._start_response)
- _ = b"".join(response).decode()
-
- # We should not receive a new Set-Cookie; we should reuse existing one.
- set_cookie_headers = [h for h in self._wsgi_headers if h[0] == 'Set-Cookie']
- self.assertFalse(set_cookie_headers, "Did not expect a new Set-Cookie header.")
-
- # Ensure we didn't lose the session
- self.assertIn(session_id, self.server.sessions, "Existing session should still be present.")
-
- @unittest.skip("Requires a valid 'greet.html' in the 'templates' folder to work.")
- def test_template_rendering(self):
- """
- Optional test for verifying a Jinja2 template render. This test will
- require a 'templates/greet.html' file that references a variable 'name'.
-
- Example 'greet.html' content:
-
- <h1>Hello {{ name }}!</h1>
- """
- environ = {
- "REQUEST_METHOD": "GET",
- "PATH_INFO": "/greet_template",
- "QUERY_STRING": "name=Tester",
- "wsgi.input": BytesIO(b""),
- "CONTENT_LENGTH": "0",
+
+ async def mock_index():
+ return "Hello, world!"
+
+ self.server.index = mock_index
+
+ await self.server.asgi_app(scope, mock_receive, mock_send)
+
+ self.assertIn("test-session-id", self.server.sessions)
+ self.assertEqual(self.server.sessions["test-session-id"].get("last_access"), 1000)
+
+ @patch("time.time", return_value=1000)
+ async def test_asgi_app_handles_request(self, mock_time):
+ mock_send = AsyncMock()
+ mock_receive = AsyncMock(return_value={"type": "http.request", "body": b""})
+
+ scope = {
+ "type": "http",
+ "method": "GET",
+ "path": "/",
+ "headers": [(b"cookie", b"session_id=test-session-id")],
}
- response = self.server.wsgi_app(environ, self._start_response)
- body = b"".join(response).decode()
- self.assertEqual(self._wsgi_status, "200 OK")
- self.assertIn("<h1>Hello Tester!</h1>", body)
+ self.server.sessions["test-session-id"] = {"last_access": 500}
+
+ async def mock_index():
+ return "Hello, test!"
+
+ self.server.index = mock_index
+
+ await self.server.asgi_app(scope, mock_receive, mock_send)
+
+ self.assertEqual(self.server.sessions["test-session-id"].get("last_access"), 1000)
+
+ @patch("jinja2.Environment.get_template")
+ def test_render_template(self, mock_get_template):
+ mock_template = MagicMock()
+ mock_template.render.return_value = "Rendered content"
+ mock_get_template.return_value = mock_template
+
+ result = self.server.render_template("test.html", var="value")
+ self.assertEqual(result, "Rendered content")
+ mock_template.render.assert_called_with({"var": "value"})
+ def test_render_template_no_jinja(self):
+ self.server.env = None
+ with self.assertRaises(ImportError):
+ self.server.render_template("test.html")
-if __name__ == '__main__':
+if __name__ == "__main__":
unittest.main()