Skip to content

Latest commit

 

History

History
235 lines (180 loc) · 5.75 KB

File metadata and controls

235 lines (180 loc) · 5.75 KB

mware

Python middleware decorators that actually make sense

PyPI version Build status Coverage Code style: black

InstallationQuick StartFeaturesDocs


mware brings the elegance of middleware patterns to Python with a DX that rivals the best libraries out there. If you've used Express.js or Koa.js, you'll feel right at home.

from mware import middleware

@middleware
async def measure_time(ctx, next):
    start = time.time()
    result = await next(ctx)
    print(f"Execution took {time.time() - start:.3f}s")
    return result

@measure_time
async def fetch_user(ctx):
    # Your business logic here
    return {"id": ctx.user_id, "name": "Alice"}

# It just works™
result = await fetch_user(ctx={"user_id": 123})

✨ Why mware?

  • 🎯 Intuitive API: Feels natural to Python developers
  • 🔒 Type-Safe: Full type hints with mypy support
  • ⚡ Async-First: Built for modern async/await patterns
  • 🪶 Zero Dependencies: Lightweight core, no bloat
  • 🧩 Composable: Chain middleware easily
  • 🚀 Fast: Minimal overhead, maximum performance

Installation

pip install mware

That's it. No complex setup. It just works.

Quick Start

Basic Middleware

from mware import middleware, Context

@middleware
async def logging_middleware(ctx: Context, next):
    print(f"Before: {ctx.request.path}")
    result = await next(ctx)
    print(f"After: {ctx.response.status}")
    return result

# Chain multiple middleware
@logging_middleware
@auth_middleware
@rate_limit_middleware
async def api_handler(ctx: Context):
    return {"message": "Hello, World!"}

Error Handling

@middleware
async def error_handler(ctx: Context, next):
    try:
        return await next(ctx)
    except ValidationError as e:
        ctx.response.status = 400
        return {"error": str(e)}
    except Exception as e:
        ctx.response.status = 500
        return {"error": "Internal server error"}

@error_handler
async def risky_operation(ctx: Context):
    # This is automatically protected
    return perform_operation(ctx.data)

Context Management

# Context flows through your middleware chain
@middleware 
async def add_user(ctx: Context, next):
    ctx.user = await fetch_user(ctx.auth_token)
    return await next(ctx)

@middleware
async def require_admin(ctx: Context, next):
    if not ctx.user.is_admin:
        raise PermissionError("Admin required")
    return await next(ctx)

@add_user
@require_admin  
async def admin_action(ctx: Context):
    # ctx.user is available here
    return {"admin": ctx.user.name}

Features

🎭 Flexible Patterns

# Before/After
@middleware
async def before_after(ctx, next):
    # Before
    prepare_something()
    
    result = await next(ctx)
    
    # After
    cleanup_something()
    return result

# Short-circuit
@middleware
async def cache(ctx, next):
    cached = get_from_cache(ctx.key)
    if cached:
        return cached
    
    result = await next(ctx)
    save_to_cache(ctx.key, result)
    return result

# Modify response
@middleware
async def add_headers(ctx, next):
    result = await next(ctx)
    ctx.response.headers["X-Powered-By"] = "mware"
    return result

🔧 Custom Middleware

from mware import create_middleware

# Function-based
log_requests = create_middleware(
    before=lambda ctx: print(f"Request: {ctx.request}"),
    after=lambda ctx, result: print(f"Response: {result}")
)

# Class-based  
class RateLimiter:
    def __init__(self, max_requests=100):
        self.max_requests = max_requests
    
    async def __call__(self, ctx, next):
        if self.is_rate_limited(ctx.ip):
            raise TooManyRequestsError()
        return await next(ctx)

rate_limiter = middleware(RateLimiter(max_requests=50))

🧪 Testing

import pytest
from mware import Context, test_middleware

@pytest.mark.asyncio
async def test_auth_middleware():
    # Test utilities make testing middleware a breeze
    ctx = Context(auth_token="valid")
    result = await test_middleware(
        auth_middleware,
        handler=lambda ctx: {"user": ctx.user.name},
        context=ctx
    )
    assert result["user"] == "Alice"

Performance

mware is designed for production use with minimal overhead:

Benchmark: 1M requests
─────────────────────────
No middleware:     1.23s
Single middleware: 1.31s  (+6.5%)
5 middleware:      1.58s  (+28%)
10 middleware:     1.89s  (+53%)

Documentation

For complete documentation, visit mware.readthedocs.io

Contributing

We love contributions! See CONTRIBUTING.md for guidelines.

License

MIT License - see LICENSE for details.


Made with ❤️ by developers who care about DX