kewe 1.0.8

Kewe - Python Async Web Framework

 _  __  __        __
| |/ /__\ \      / /__
| ' // _ \ \ /\ / / _ \
| . \  __/\ V  V /  __/
|_|\_\___| \_/\_/ \___|

Python Async Web Framework

Fast, modern, batteries-included — built on pure asyncio

---

Why KeWe?

A single import gives you the full HTTP stack — routing, middleware, security, WebSockets, OpenAPI docs, health checks, background tasks, and more. No boilerplate, no plugin hunting.

from kewe import Kewe, json

app = Kewe("hello")

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

app.run()  # → http://0.0.0.0:8000 + OpenAPI docs at /docs

---

Installation

pip install kewe

Optional feature groups:

Feature	Command
JWT, CSRF, rate limiting, password hashing	pip install kewe[security]
SQLAlchemy + Alembic	pip install kewe[database]
Redis cache backend	pip install kewe[redis]
Pydantic validation	pip install kewe[pydantic]
OpenTelemetry tracing	pip install kewe[otel]
HTTP/2	pip install kewe[http2]
HTTP/3	pip install kewe[http3]
Everything	pip install kewe[all]

---

Core Features

Category	Capabilities
HTTP	HTTP/1.1, HTTP/2, HTTP/3, keep-alive, range requests, SSE
Routing	Regex router, type converters, url_for, blueprints, class-based views
Middleware	Onion-style pipeline — CORS, compression, security headers, request ID
Security	JWT (HS256/RS256), OAuth2, API Key, Basic Auth, CSRF, rate limiting, RBAC
DI	Depends, Query, Header, Body, Form, File, Path, Security
WebSocket	Connection manager, broadcasting, group messaging, JSON transport
OpenAPI	Auto-generated OpenAPI 3.0 + Swagger UI + ReDoc from docstrings
Monitoring	Health checks, Prometheus metrics, anomaly detection, audit logging
Background	Fire-and-forget tasks, scheduled jobs, async processor
Caching	Memory / Redis backends, @cache, @memoize, @cached_property
Testing	TestClient, AsyncTestClient, TestCase, TestWebSocket
CLI	REPL, shell commands, file-watcher reloader
ASGI	Run behind Uvicorn, Hypercorn, or any ASGI server

---

Quick Tour

Routing & Responses

from kewe import Kewe, json, text, html, redirect, Request

app = Kewe("demo")

@app.get("/users")
async def list_users():
    return json([{"id": 1, "name": "Alice"}])

@app.get("/users/{user_id:int}")
async def get_user(user_id: int):
    return json({"id": user_id})

@app.post("/users")
async def create_user(request: Request):
    data = await request.json
    return json(data, status=201)

@app.get("/hello")
async def hello(name: str = "World"):
    return html(f"<h1>Hello, {name}!</h1>")

Blueprints

from kewe import Blueprint

api = Blueprint("api", url_prefix="/api/v1")

@api.get("/status")
async def status():
    return {"ok": True}

app.register_blueprint(api)

Middleware

from kewe.middleware import CORSMiddleware, SecurityHeadersMiddleware

app.add_middleware(CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["GET", "POST"],
    allow_headers=["Content-Type"],
)

# Custom middleware
@app.middleware
async def add_timing(request, call_next):
    response = await call_next(request)
    response.headers["X-Response-Time"] = "0.5ms"
    return response

WebSocket

from kewe import WebSocketConnection

@app.websocket("/ws/chat")
async def chat(ws: WebSocketConnection):
    await ws.accept()
    async for msg in ws:
        await ws.send_json({"echo": msg.json()})

JWT Authentication

from kewe.security.jwt import JWTAuthManager
from kewe.security.auth import login_required

jwt = JWTAuthManager(secret_key="your-256-bit-secret-key")

@app.post("/login")
async def login(request: Request):
    data = await request.json
    token = jwt.create_token({"user_id": data["id"]})
    return {"access_token": token}

@app.get("/me")
@login_required
async def me(request: Request):
    return {"authenticated": True}

Server-Sent Events

from kewe import ServerSentEventResponse

@app.get("/stream")
async def stream():
    async def generate():
        for i in range(5):
            yield f"data: Tick {i}\n\n"
    return ServerSentEventResponse(generate())

Health Checks

from kewe.monitoring.health import setup_health_checks

endpoint = setup_health_checks(app, path="/health")
endpoint.checker.add_check("db", lambda: (True, "OK"))
# → GET /health/live  and  /health/ready

---

Application Factory

from kewe import create_app, KeweConfig

config = KeweConfig()
config.host = "0.0.0.0"
config.port = 8080
config.secret_key = "your-secret-key"

app = create_app("my_app", config=config)

@app.get("/")
async def root():
    return {"status": "running"}

if __name__ == "__main__":
    app.run()  # MOTD logo displayed automatically

---

Startup MOTD

By default, app.run() displays the framework logo:

KeWe
Going Fast @ http://0.0.0.0:8000
     mode: production
   server: kewe, HTTP/1.1
   python: 3.14.3

Pass motd=False to suppress, or use display_motd(full=True) for the full ASCII art banner.

---

Project Layout

kewe/
├── application/     # Core, config, DI, signals, state, MOTD/logo
├── background/      # Tasks, scheduling, cleanup, token rotation
├── base/            # Constants, context vars, types, cancellation
├── cache/           # Manager, memory/redis backends, decorators
├── cli/             # CLI runner, interactive REPL
├── cookies/         # Cookie jar, signed cookies, sessions
├── database/        # SQLAlchemy ORM, query builder, migrations
├── errors/          # Exceptions, handlers, circuit breaker, tracking
├── gateway/         # API gateway adapter
├── handlers/        # Static file serving, directory browser
├── http/            # Client, headers, keepalive, URL parser, HTTP/2-3
├── logging/         # Structured logging, access logs, formatters
├── middleware/       # Onion pipeline, CORS, compression, request ID
├── monitoring/      # Health, metrics, anomaly detection, audit log
├── plugins/         # Auth, database, OpenAPI, Pydantic, builtins
├── request/         # Request, multipart, streaming uploads
├── response/        # JSON, HTML, streaming, SSE, file responses
├── routing/         # Router, blueprints, views, URL converter
├── security/        # JWT, auth, CSRF, rate limiting, RBAC, passwords
├── server/          # HTTP server, ASGI adapter, SSL, reloader
├── telemetry/       # OpenTelemetry tracing
├── touchup/         # Zero-copy, performance tuning, URL schemes
├── utils/           # Testing, i18n, pagination, helpers, locks
├── websocket/       # Protocol, connection, manager, handler
├── worker/          # Process workers, shared context, multiplexer
├── app.py           # Kewe application class
└── factory.py       # create_app() factory

---

Requirements

• Python 3.12+
• orjson — fast JSON (optional; falls back to ujson → json)
• winloop (Windows) / uvloop (Linux) — accelerated event loop

---

License

MIT — Ouopoulos