Metadata-Version: 2.4
Name: hypern
Version: 1.1.3
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: BSD License
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Programming Language :: Python :: Implementation :: CPython
Classifier: Programming Language :: Rust
Classifier: Topic :: Internet :: WWW/HTTP :: HTTP Servers
Classifier: Topic :: Internet :: WWW/HTTP :: WSGI :: Server
Classifier: Topic :: Internet :: WWW/HTTP :: Dynamic Content
Classifier: Topic :: Software Development :: Libraries :: Application Frameworks
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Framework :: AsyncIO
Classifier: Environment :: Web Environment
Classifier: Typing :: Typed
Requires-Dist: orjson>=3.10.11
Requires-Dist: msgspec>=0.18.0
Requires-Dist: typing-extensions>=4.15.0,<5.0.0
Requires-Dist: watchdog>=6.0.0,<7.0.0
Requires-Dist: click>=8.4.2
License-File: LICENSE
Summary: A Fast Async Python backend with a Rust runtime.
Author-email: Martin Dang <vannghiem848@gmail.com>
Requires-Python: >=3.11, <4.0
Description-Content-Type: text/markdown; charset=UTF-8; variant=GFM


# Hypern

**A High-Performance Python Web Framework with Rust Runtime**

Hypern is a flexible, open-source web framework that combines the ease of Python with the raw performance of [Rust](https://github.com/rust-lang/rust). Built on top of production-ready Rust libraries, Hypern empowers you to rapidly develop high-performance web applications, RESTful APIs, and real-time systems.

With Hypern, you get seamless async/await support, built-in WebSocket and SSE capabilities, powerful task scheduling, and comprehensive middleware support—all while writing familiar Python code with the performance characteristics of Rust.

### 🏁 Get started

### ⚙️ To Develop Locally

- Setup a virtual environment:
```
python3 -m venv venv
source venv/bin/activate
```
- Install required packages

```
pip install pre-commit poetry maturin
```
- Install development dependencies
```
poetry install --with dev --with test
```
- Install pre-commit git hooks
```
pre-commit install
```
- Build & install Rust package
```
maturin develop
```

## 🚀 Quick Start

### Basic Example

```python
# main.py
from hypern import Hypern

app = Hypern()

@app.get("/")
async def home(req, res, ctx):
    res.json({"message": "Hello, World!"})

@app.get("/users/:id")
async def get_user(req, res, ctx):
    user_id = req.param("id")
    res.json({"id": user_id, "name": "John Doe"})

@app.post("/users")
async def create_user(req, res, ctx):
    body = req.json()
    res.status(201).json({"created": body})

if __name__ == "__main__":
    app.listen(port=5000, host="0.0.0.0")
```

```bash
$ python3 main.py
```

Your server will be available at `http://localhost:5000`

### OpenAPI/Swagger Documentation

Enable built-in API documentation:

```python
app = Hypern()

# Enable OpenAPI
app.setup_openapi(
    title="My API",
    version="1.0.0",
    description="My awesome API"
)

# Your routes here...
```

Access documentation at:
- **Swagger UI**: `http://localhost:5000/docs`
- **ReDoc**: `http://localhost:5000/redoc`
- **OpenAPI Spec**: `http://localhost:5000/openapi.json` 

## ⚙️ Server Configuration

### Production Server

The `listen()` method is a simplified wrapper for starting the server:

```python
app.listen(
    port=5000,                   # Port number
    host="0.0.0.0",             # Host address to bind to
    callback=None,               # Optional callback after server starts
)
```

For more control, use the `start()` method:

```python
app.start(
    host="0.0.0.0",             # Host address to bind to
    port=5000,                   # Port number
    num_processes=1,             # Number of worker processes (multi-core)
    workers_threads=1,           # Number of worker threads per process
    max_blocking_threads=16,     # Max blocking threads for sync operations
    max_connections=10000,       # Max concurrent connections
)
```

### Development Server

For development with auto-reload on file changes:

```python
app.run_dev(
    port=3000,                   # Port number
    host="0.0.0.0",             # Host address
    reload=True,                 # Enable auto-reload (default: True)
    reload_dirs=[".", "./src"],  # Directories to watch
    reload_delay=0.5,            # Debounce delay in seconds
)
```

**Example - Multi-process production server:**

```python
# Utilize all CPU cores with multiple processes
app.start(
    port=8000, 
    num_processes=4,          # 4 processes
    workers_threads=2,        # 2 threads per process
    max_blocking_threads=32
)
```


## 💡 Features

### ⚡ High Performance
- **Rust-powered core** with Python flexibility
- **Multi-process architecture** for optimal CPU utilization across cores
- **Async/await support** for non-blocking I/O operations
- **Zero-copy** data handling where possible
- **Optimized routing** with efficient path matching

### 🌐 Web Capabilities
- **RESTful API** routing with decorators (`@app.get`, `@app.post`, etc.)
- **WebSocket support** with rooms and broadcasting
- **Server-Sent Events (SSE)** for real-time streaming
- **File uploads** with multipart form data handling
- **Static file serving** with caching headers
- **Streaming responses** for large data transfers

### 🔌 Integration & Extensions
- **Dependency Injection** (DI) with singleton and factory patterns
- **Middleware support** with one async `next()` pipeline
  - CORS middleware
  - Rate limiting middleware
  - Compression middleware
  - Security headers middleware
  - Request ID tracking
  - Timeout middleware
  - Logging middleware
  - Basic authentication middleware
- **Background task execution** with TaskExecutor
- **Task scheduling** with cron expressions and intervals
- **Router mounting** for modular application structure

### 🛠 Development Experience
- **Type hints** and comprehensive IDE support
- **Built-in Swagger/OpenAPI** documentation (Swagger UI + ReDoc)
- **Hot reload** during development
- **Comprehensive error handling** with custom exception handlers
- **Request validation** decorators for query params, body, and path params
- **Detailed logging** with configurable levels

### 🔒 Security & Authentication
- **JWT Authentication** with token validation
- **API Key authentication** for service-to-service communication
- **Role-Based Access Control (RBAC)** with decorators
- **Permission-based authorization** (`@requires_role`, `@requires_permission`)
- **CORS configuration** for cross-origin requests
- **Rate limiting** to prevent abuse
- **Request validation** to prevent malformed input
- **Security headers** middleware (HSTS, CSP, etc.)

---

## 📚 Comprehensive Examples

### Middleware

```python
from hypern import Hypern
from hypern.middleware import CorsMiddleware, RateLimitMiddleware, LogMiddleware

app = Hypern()

# Add CORS support
cors = CorsMiddleware(
    allow_origins=["*"],
    allow_methods=["GET", "POST", "PUT", "DELETE"],
    allow_headers=["Content-Type", "Authorization"]
)
app.use(cors)

# Add rate limiting (100 requests per minute)
rate_limit = RateLimitMiddleware(max_requests=100, window_seconds=60)
app.use(rate_limit)

# Add request logging
app.use(LogMiddleware())

# Custom middleware can run before and after the downstream handler
async def add_server_header(req, res, ctx, next):
    print(f"→ {req.method} {req.path}")
    await next()
    res.header("X-Powered-By", "Hypern")

app.use(add_server_header)
```

### WebSocket Support

```python
from hypern import Hypern
from hypern.websocket import WebSocket, WebSocketRoom, WebSocketDisconnect

app = Hypern()
chat_room = WebSocketRoom()

@app.ws("/chat")
async def chat_handler(ws: WebSocket):
    await ws.accept()
    chat_room.join(ws)
    
    try:
        while True:
            message = await ws.receive_text()
            # Broadcast to all connected clients
            chat_room.broadcast(f"User: {message}")
    except WebSocketDisconnect:
        chat_room.leave(ws)
```

### Server-Sent Events (SSE)

```python
from hypern import Hypern, SSEEvent

app = Hypern()

@app.get("/events")
async def stream_events(req, res, ctx):
    events = [
        SSEEvent("Connected!", event="greeting"),
        SSEEvent('{"price": 100}', event="update"),
        SSEEvent("System online", event="status")
    ]
    res.sse(events)

# For continuous streaming
@app.get("/live")
async def live_stream(req, res, ctx):
    stream = app.sse()
    stream.send_event("message", "Hello!")
    stream.send_data("Plain data")
    # Return collected events
```

### Background Tasks

```python
from hypern import Hypern, background, submit_task, get_task

app = Hypern()

# Using decorator
@background()
def send_email(to: str, subject: str, body: str):
    # This runs in background
    print(f"Sending email to {to}")
    # ... email sending logic
    return {"status": "sent"}

@app.post("/notify")
async def notify_user(req, res, ctx):
    data = req.json()
    # Submit background task
    send_email(data["email"], "Welcome!", "Thanks for joining!")
    res.json({"status": "queued"})

# Programmatic submission with task tracking
@app.post("/process")
async def process_data(req, res, ctx):
    def heavy_computation(data):
        # CPU-intensive work
        return {"result": "processed"}
    
    task_id = submit_task(heavy_computation, args=(req.json(),))
    res.json({"task_id": task_id})

@app.get("/task/:id")
async def check_task(req, res, ctx):
    task_id = req.param("id")
    result = get_task(task_id)
    
    if result and result.is_success():
        res.json({"status": "completed", "result": result.result})
    else:
        res.json({"status": "pending"})
```

### Task Scheduling

```python
from hypern import Hypern
from hypern.scheduler import periodic, RetryPolicy

app = Hypern()

# Scheduled task - runs every 30 seconds
@periodic(seconds=30)
def health_check():
    print("Health check running...")

# Cron-style scheduling - every day at 3 AM
@app.scheduler.cron("0 3 * * *")
def nightly_cleanup():
    print("Running nightly cleanup...")

# With retry policy
@app.scheduler.task(retry=RetryPolicy(max_retries=3, backoff=2.0))
def flaky_task():
    # This will retry up to 3 times with exponential backoff
    pass
```

### Authentication & Authorization

```python
from hypern import Hypern
from hypern.auth import JWTAuth, requires_role, requires_permission

app = Hypern()

# JWT Authentication
jwt = JWTAuth(secret="your-secret-key", algorithm="HS256")

@app.post("/login")
async def login(req, res, ctx):
    body = req.json()
    # Validate credentials...
    
    token = jwt.encode(
        {"sub": "user-1", "user_id": 123, "roles": ["admin"]},
        expiry_seconds=3600  # 1 hour
    )
    res.json({"token": token})

@app.get("/protected")
@jwt.required
async def protected_route(req, res, ctx):
    # User payload is available in ctx
    user = ctx.get("auth_user")
    res.json({"user": user})

# Manual token verification
@app.get("/verify")
async def verify_token(req, res, ctx):
    token = req.header("Authorization").replace("Bearer ", "")
    try:
        payload = jwt.decode(token)
        res.json({"valid": True, "payload": payload})
    except Exception as e:
        res.status(401).json({"error": str(e)})

# Role-based access control
@app.get("/admin")
@jwt.required
@requires_role("admin")
async def admin_only(req, res, ctx):
    res.json({"message": "Admin area"})

@app.delete("/users/:id")
@jwt.required
@requires_permission("users:delete")
async def delete_user(req, res, ctx):
    res.json({"deleted": True})
```

### Request Binding and Validation

```python
import msgspec

from hypern import Body, Header, Hypern, Json, Path, Query

app = Hypern()

class CreateUser(msgspec.Struct):
    name: str
    email: str
    age: int

class SearchQuery(msgspec.Struct):
    q: str
    limit: int = 10

@app.post("/users/:user_id")
async def create_user(
    res,
    user_id: int = Path(),
    payload: CreateUser = Json(),
    request_id: str | None = Header("X-Request-ID"),
):
    res.status(201).json({"id": user_id, "name": payload.name, "request_id": request_id})

@app.get("/search")
async def search(res, query: SearchQuery = Query()):
    res.json({"query": query.q, "limit": query.limit})

@app.post("/events")
async def receive_event(res, raw_body: bytes = Body()):
    res.json({"size": len(raw_body)})
```

`Json()`, `Query()`, `Path()`, and `Header()` coerce values using their type annotations. Invalid JSON, missing required values, and failed coercions flow through Hypern's normal validation error handling. `Body()` supplies raw `bytes` without parsing JSON.

### Router Mounting

```python
from hypern import Hypern, Router

app = Hypern()

# Create API v1 router
api_v1 = Router(prefix="/api/v1")

@api_v1.get("/users")
async def get_users_v1(req, res, ctx):
    res.json({"version": "v1", "users": []})

# Create API v2 router
api_v2 = Router(prefix="/api/v2")

@api_v2.get("/users")
async def get_users_v2(req, res, ctx):
    res.json({"version": "v2", "users": []})

# Mount routers
app.use(api_v1)
app.use(api_v2)

# Routes are now available at:
# - /api/v1/users
# - /api/v2/users
```

### File Uploads

```python
from hypern import Hypern

app = Hypern()

@app.post("/upload")
async def upload_file(req, res, ctx):
    form_data = await req.form()
    
    file = form_data.get_file("document")
    if file:
        # Access file properties
        filename = file.filename
        content_type = file.content_type
        content = file.content  # bytes
        
        # Save file
        with open(f"uploads/{filename}", "wb") as f:
            f.write(content)
        
        res.json({
            "filename": filename,
            "size": len(content),
            "type": content_type
        })
    else:
        res.status(400).json({"error": "No file uploaded"})
```

### Dependency Injection

```python
from hypern import Hypern, Inject

app = Hypern()

# Register an application-owned singleton (shared instance)
class CatalogService:
    def list_items(self):
        return [{"id": 1, "name": "John"}]

app.provide(CatalogService, CatalogService())

# Register a provider once per request
class RequestLogger:
    def log(self, message):
        print(f"LOG: {message}")

app.provide(RequestLogger, RequestLogger, scope="request")

# Use dependencies in routes
@app.get("/data")
async def get_data(
    res,
    catalog_service: CatalogService = Inject(),
    logger: RequestLogger = Inject(),
):
    logger.log("Fetching data")
    data = catalog_service.list_items()
    res.json(data)
```

`app.provide(key, provider, scope=...)` accepts existing values, classes, synchronous factories, and asynchronous factories. The scopes are `singleton` (the default, shared for the app lifetime), `request` (cached for one request), and `transient` (created for every resolution). Async factories must use `request` or `transient` scope because a singleton created on one worker event loop is unsafe to share with other loops. `Inject()` uses the annotation as its key; use `Inject("name")` for an explicit key. An unsupported scope fails immediately with `InjectionConfigurationError`. Calling the public `app.provide()` after application registration has frozen fails immediately with `RuntimeError`; calling `ProviderRegistry.provide()` after its registry is frozen raises `InjectionConfigurationError`. Missing provider dependencies, dependency cycles, and async singleton graphs fail when Hypern freezes the provider graph; invalid handler marker declarations fail when Hypern compiles the handler plan.

Each handler parameter declares its own source: `Inject()`, `Json()`, `Query()`, `Header(name)`, `Path(name)`, or `Body()`. This means binding does not depend on decorator order. The older DI and validation decorators are not part of this API.

### Error Handling

```python
from hypern import Hypern, HTTPException, NotFound, exception_handler

app = Hypern()

# Custom exception handler
@app.errorhandler(NotFound)
def handle_not_found(req, res, error):
    res.status(404).json({
        "error": "Resource not found",
        "path": req.path
    })

@app.errorhandler(Exception)
def handle_generic_error(req, res, error):
    res.status(500).json({
        "error": "Internal server error",
        "message": str(error)
    })

# Raise HTTP exceptions
@app.get("/users/:id")
async def get_user(req, res, ctx):
    user_id = req.param("id")
    
    # Simulate user not found
    if user_id == "999":
        raise NotFound("User not found")
    
    res.json({"id": user_id})
```

### Static Files

```python
from hypern import Hypern

app = Hypern()

# Serve static files from 'public' directory at '/static' URL
app.static("/static", "public")

# Serve frontend build from 'dist' directory at root
app.static("/", "dist")

# Now files are accessible:
# - /static/css/style.css → public/css/style.css
# - /static/js/app.js → public/js/app.js
# - / → dist/index.html
```

---

## 📖 Documentation

For more detailed documentation, visit:
- [Getting Started Guide](docs/getting-started.md)
- [Routing](docs/routing.md)
- [Middleware](docs/middleware.md)
- [Authentication](docs/auth.md)
- [WebSocket](docs/websocket.md)
- [Task Scheduling](docs/scheduling.md)
- [Request Validation](docs/validation.md)
- [Dependency Injection](docs/dependency-injection.md)

## 🤝 Contributing

Contributions are welcome! Please feel free to submit a Pull Request.

## 📄 License

This project is licensed under the terms specified in the [LICENSE](LICENSE) file.

## 🔗 Links

- [Documentation](https://hypern.dev/docs)
- [GitHub Repository](https://github.com/yourusername/hypern)
- [Issue Tracker](https://github.com/yourusername/hypern/issues)

---

**Built with ❤️ using Python and Rust**

