Skip to content

Latest commit

 

History

History

Folders and files

NameName
Last commit message
Last commit date

parent directory

..
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 

README.md

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