ecko 0.1.0
A modern Python web framework. Flask's simplicity with 2026 expectations.
Requires Python: >=3.11
Keywords: api, async, framework, http, web

# Ecko

A fast, minimal & modern Python web framework.

```python
from ecko import Ecko

app = Ecko()

@app.get("/")
def hello():
    return {"message": "Hello, World!"}
```

```bash
ecko run app:app
```

## Why Ecko?

- **Simple by default** - No boilerplate, no configuration, just write handlers
- **Type-safe without ceremony** - Automatic validation from type hints
- **Sync and async, seamlessly** - Write sync code, it runs in a threadpool automatically
- **Batteries included** - CORS, sessions, OpenAPI docs built-in but optional
- **Fast** - Built on ASGI with msgspec for JSON (5-10x faster than Pydantic)
- **WebSockets** - Out of the box support for WebSockets

## Installation

```bash
pip install ecko
```

Requires Python 3.12+.

## Quick Start

### Create a new project

```bash
ecko new myproject
cd myproject
pip install -e .
ecko run app:app
```

### Or start from scratch

```python
# app.py
from ecko import Ecko

app = Ecko()

@app.get("/")
def home():
    return {"message": "Hello, World!"}

@app.get("/users/{user_id}")
def get_user(user_id: int):
    return {"id": user_id, "name": f"User {user_id}"}

@app.post("/users")
def create_user(data: dict):
    return {"created": data}, 201
```

```bash
ecko run app:app
```

Open http://127.0.0.1:8000

## Features

### Route Parameters

Parameters are extracted automatically from type hints:

```python
@app.get("/users/{user_id}")
def get_user(
    user_id: int,              # From path (matches {user_id})
    include_posts: bool = False # From query string
):
    return {"id": user_id, "include_posts": include_posts}
```

```
GET /users/42?include_posts=true
→ {"id": 42, "include_posts": true}
```

### Type Coercion

Ecko automatically converts strings to the declared type:

```python
@app.get("/search")
def search(
    q: str,                    # Required string
    page: int = 1,             # Optional int with default
    limit: int | None = None,  # Optional, None if missing
    tags: list[str] = [],      # List from ?tags=a&tags=b
):
    return {"query": q, "page": page, "limit": limit, "tags": tags}
```

### Request Body

For POST/PUT/PATCH, dict or msgspec Struct parameters are parsed from JSON:

```python
@app.post("/users")
def create_user(data: dict):
    return {"created": data}, 201
```

For typed validation, use msgspec Structs:

```python
import msgspec

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

@app.post("/users")
def create_user(user: CreateUser):
    return {"name": user.name, "email": user.email}
```

Invalid JSON returns a 422 error automatically.

### Explicit Parameter Markers

For edge cases, use explicit markers:

```python
from ecko import Query, Path, Header, Body

@app.get("/items/{item_id}")
def get_item(
    item_id: int = Path(description="The item ID"),
    q: str = Query("", description="Search query"),
    token: str = Header(alias="x-api-token"),
):
    return {"item_id": item_id, "query": q}
```

### Responses

Return values are automatically converted to JSON responses:

```python
# Dict → 200 JSON
return {"data": "value"}

# Tuple → status code
return {"created": True}, 201

# Tuple with headers
return {"data": "value"}, 200, {"X-Custom": "header"}

# Explicit response
from ecko import Response, JSONResponse, HTMLResponse, RedirectResponse

return HTMLResponse("<h1>Hello</h1>")
return RedirectResponse("/login")
```

### Async and Sync Handlers

Both work seamlessly:

```python
@app.get("/sync")
def sync_handler():
    # Runs in threadpool automatically
    time.sleep(1)
    return {"sync": True}

@app.get("/async")
async def async_handler():
    # Native async
    await asyncio.sleep(1)
    return {"async": True}
```

### Request Context

Access request data from anywhere without passing it around:

```python
from ecko import context

@app.before_request
def load_user():
    token = context.request.headers.get("authorization")
    context.user = get_user_from_token(token)

@app.get("/profile")
def profile():
    return {"name": context.user.name}
```

### Before/After Request Hooks

```python
@app.before_request
def authenticate():
    if not is_valid_token(context.request.headers.get("authorization")):
        return {"error": "Unauthorized"}, 401  # Short-circuit

@app.after_request
def add_headers(response):
    response._headers["X-Request-ID"] = generate_id()
    return response
```

### Exception Handling

Built-in HTTP exceptions:

```python
from ecko import NotFound, BadRequest, Unauthorized

@app.get("/users/{user_id}")
def get_user(user_id: int):
    user = db.get_user(user_id)
    if not user:
        raise NotFound("User not found")
    return user
```

Custom exception handlers:

```python
class RateLimitExceeded(Exception):
    pass

@app.exception_handler(RateLimitExceeded)
def handle_rate_limit(exc):
    return {"error": "Too many requests"}, 429

@app.exception_handler(ValueError)
def handle_value_error(exc):
    return {"error": str(exc)}, 400
```

### Middleware

```python
async def timing_middleware(request, call_next):
    start = time.time()
    response = await call_next()
    response._headers["X-Response-Time"] = f"{time.time() - start:.3f}s"
    return response

app.use(timing_middleware)
```

### CORS

```python
from ecko.middleware import cors

# Allow all origins (development)
app.use(cors())

# Specific origins (production)
app.use(cors(
    origins=["https://myapp.com"],
    methods=["GET", "POST", "PUT", "DELETE"],
    allow_credentials=True,
))
```

### Sessions

Signed cookie-based sessions:

```python
from ecko import context
from ecko.middleware import sessions

app.use(sessions(secret="your-secret-key-min-32-chars!!"))

@app.post("/login")
def login(data: dict):
    context.session["user_id"] = data["user_id"]
    return {"logged_in": True}

@app.get("/profile")
def profile():
    user_id = context.session.get("user_id")
    if not user_id:
        return {"error": "Not logged in"}, 401
    return {"user_id": user_id}

@app.post("/logout")
def logout():
    context.session.clear()
    return {"logged_out": True}
```

### WebSockets

Full WebSocket support with a clean async API:

```python
from ecko import Ecko, WebSocket

app = Ecko()

@app.websocket("/ws")
async def websocket_handler(ws: WebSocket):
    await ws.accept()
    async for message in ws:
        await ws.send_text(f"Echo: {message}")
```

#### WebSocket Methods

```python
@app.websocket("/chat/{room_id}")
async def chat(ws: WebSocket):
    # Access path parameters
    room_id = ws.path_params["room_id"]

    # Accept the connection
    await ws.accept()

    # Receive messages
    text = await ws.receive_text()
    data = await ws.receive_json()
    binary = await ws.receive_bytes()

    # Send messages
    await ws.send_text("Hello!")
    await ws.send_json({"status": "ok"})
    await ws.send_bytes(b"\x00\x01")

    # Close the connection
    await ws.close(code=1000, reason="Goodbye")
```

#### Handling Disconnections

```python
from ecko import WebSocket, WebSocketDisconnect

@app.websocket("/ws")
async def handler(ws: WebSocket):
    await ws.accept()
    try:
        while True:
            message = await ws.receive_text()
            await ws.send_text(f"Got: {message}")
    except WebSocketDisconnect:
        print("Client disconnected")
```

Or use the async iterator which handles disconnections automatically:

```python
@app.websocket("/ws")
async def handler(ws: WebSocket):
    await ws.accept()
    async for message in ws:
        await ws.send_text(f"Got: {message}")
    # Loop exits cleanly on disconnect
```

### Lifecycle Events

```python
@app.on_startup
async def startup():
    await database.connect()
    print("App started!")

@app.on_shutdown
async def shutdown():
    await database.disconnect()
    print("App stopped!")
```

### OpenAPI Documentation

Auto-generated API docs with Swagger UI:

```python
from ecko import Ecko, setup_openapi

app = Ecko()

setup_openapi(
    app,
    title="My API",
    version="1.0.0",
    description="My awesome API",
)

@app.get("/users/{user_id}")
def get_user(user_id: int):
    """Get a user by their ID."""
    return {"id": user_id}
```

- `GET /docs` → Swagger UI
- `GET /openapi.json` → OpenAPI 3.1 schema

Docstrings become descriptions. Type hints become schemas.

## CLI

```bash
# Run with auto-reload (development)
ecko run app:app

# Run in production mode
ecko run app:app --prod --workers 4

# Custom host/port
ecko run app:app --host 0.0.0.0 --port 3000

# Create new project
ecko new myproject

# List all routes
ecko routes app:app
```

## Full Example

```python
"""A complete Ecko application."""

import msgspec
from ecko import Ecko, Query, Header, context, NotFound, setup_openapi
from ecko.middleware import cors, sessions


class CreateTodo(msgspec.Struct):
    title: str
    completed: bool = False


app = Ecko()

# Documentation
setup_openapi(app, title="Todo API", version="1.0.0")

# Middleware
app.use(cors())
app.use(sessions(secret="change-me-in-production-use-env-var"))

# In-memory storage (use a real database in production)
todos: dict[int, dict] = {}
next_id = 1


@app.on_startup
def startup():
    print("Todo API ready!")


@app.before_request
def request_logging():
    print(f"{context.request.method} {context.request.path}")


@app.get("/")
def home():
    """API information."""
    return {
        "name": "Todo API",
        "docs": "/docs",
    }


@app.get("/todos")
def list_todos(completed: bool | None = None, limit: int = Query(10)):
    """List all todos with optional filtering."""
    result = list(todos.values())
    if completed is not None:
        result = [t for t in result if t["completed"] == completed]
    return {"todos": result[:limit]}


@app.get("/todos/{todo_id}")
def get_todo(todo_id: int):
    """Get a specific todo by ID."""
    if todo_id not in todos:
        raise NotFound(f"Todo {todo_id} not found")
    return todos[todo_id]


@app.post("/todos")
def create_todo(todo: CreateTodo):
    """Create a new todo."""
    global next_id
    new_todo = {
        "id": next_id,
        "title": todo.title,
        "completed": todo.completed,
    }
    todos[next_id] = new_todo
    next_id += 1
    return new_todo, 201


@app.put("/todos/{todo_id}")
def update_todo(todo_id: int, data: dict):
    """Update a todo."""
    if todo_id not in todos:
        raise NotFound(f"Todo {todo_id} not found")
    todos[todo_id].update(data)
    return todos[todo_id]


@app.delete("/todos/{todo_id}")
def delete_todo(todo_id: int):
    """Delete a todo."""
    if todo_id not in todos:
        raise NotFound(f"Todo {todo_id} not found")
    del todos[todo_id]
    return {"deleted": True}


if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="127.0.0.1", port=8000)
```

## API Reference

### Core

| Export                | Description                  |
|-----------------------|------------------------------|
| `Ecko`                | Main application class       |
| `Request`             | Request object               |
| `WebSocket`           | WebSocket connection         |
| `WebSocketDisconnect` | Exception for disconnections |
| `context`             | Request-scoped context       |
| `setup_openapi`       | Enable OpenAPI docs          |

### Parameter Markers

| Export   | Description              |
|----------|--------------------------|
| `Query`  | Explicit query parameter |
| `Path`   | Explicit path parameter  |
| `Header` | Extract from headers     |
| `Body`   | Explicit body parameter  |

### Responses

| Export             | Description   |
|--------------------|---------------|
| `Response`         | Base response |
| `JSONResponse`     | JSON response |
| `HTMLResponse`     | HTML response |
| `RedirectResponse` | HTTP redirect |

### Exceptions

| Export                | Description         |
|-----------------------|---------------------|
| `HTTPException`       | Base HTTP exception |
| `BadRequest`          | 400 error           |
| `Unauthorized`        | 401 error           |
| `Forbidden`           | 403 error           |
| `NotFound`            | 404 error           |
| `MethodNotAllowed`    | 405 error           |
| `ValidationError`     | 422 error           |
| `InternalServerError` | 500 error           |

### Middleware

```python
from ecko.middleware import cors, sessions
```

| Export     | Description                |
|------------|----------------------------|
| `cors`     | CORS middleware factory    |
| `sessions` | Session middleware factory |

## Comparison

| Feature              | Ecko     | Flask      | FastAPI     |
|----------------------|----------|------------|-------------|
| Async support        | Native   | Extension  | Native      |
| WebSockets           | Built-in | Extension  | Built-in    |
| Type validation      | Auto     | Manual     | Pydantic    |
| OpenAPI docs         | Built-in | Extension  | Built-in    |
| Dependency injection | Context  | `g` object | `Depends()` |
| Learning curve       | Low      | Low        | Medium      |
| Performance          | Fast     | Moderate   | Fast        |

## TODO

- [ ] Default HTML page for new projects
- [ ] More examples
- [ ] Styled exception pages with comprehensive stack traces
- [ ] More middleware
- [ ] Native OAuth support
- [ ] Robust configuration pattern
- [ ] Project organization and structure
- [ ] Enriched CLI tooling for easier project management and scaffolding
- [ ] More tests
- [ ] More docs
- [ ] More examples
- [ ] Built-in authentication framework 
- [ ] Built-in event bus
- [ ] Built-in task queue
- [ ] Built-in caching
- [ ] Built-in file storage interface compatible with block storage services
- [ ] Pre-built integrations with popular services
- [ ] Pre-configured CI/CD pipeline scripts with Github Actions 
- [ ] Pre-configured Docker images
- [ ] Pre-configured logging and monitoring scaffolds for AWS and GCP
- [ ] start.ecko.sh page to get started quickly

## License

MIT

## Links

- Author: https://github.com/sn
- Website: https://ecko.sh
- GitHub: https://github.com/ecko-sh/ecko