tessl/pypi-minos-microservice-networks
The networks core of the Minos Framework providing networking components for reactive microservices
—
routers.mddocs/
Routing System
The routing system provides abstract interfaces and concrete implementations for handling different types of route registration and management. Routers aggregate decorated handlers and provide structured access to routing information across different transport types.
Capabilities
Base Router Interface
Abstract base class that defines the routing contract for all transport types.
from abc import ABC, abstractmethod
from typing import Dict, List, Callable
class Router(ABC):
routes: Dict[Any, Callable]
@abstractmethod
def build_routes(self) -> None: ...
def get_routes(self) -> Dict[Any, Callable]: ...Usage Examples:
from minos.networks import Router
# Routers are typically used internally by the framework
# They aggregate decorated handlers automaticallyBroker Router
Router implementation for message broker handlers, collecting broker-decorated methods.
class BrokerRouter(Router):
def __init__(self): ...
routes: Dict[BrokerEnrouteDecorator, Callable]
def build_routes(self) -> None: ...Usage Examples:
from minos.networks import BrokerRouter, enroute
class MessageService:
@enroute.broker.command("user.create")
async def create_user(self, request):
return Response({"status": "created"})
# Router automatically discovers and manages broker routes
broker_router = BrokerRouter()
broker_router.build_routes()HTTP Router
Router implementation for HTTP handlers, managing REST endpoint registration.
class HttpRouter(Router):
def __init__(self): ...
routes: Dict[HttpEnrouteDecorator, Callable]
def build_routes(self) -> None: ...REST HTTP Router
Specialized HTTP router for RESTful service endpoints.
class RestHttpRouter(HttpRouter):
def __init__(self): ...
routes: Dict[RestEnrouteDecorator, Callable]
def build_routes(self) -> None: ...Usage Examples:
from minos.networks import RestHttpRouter, enroute
class UserAPI:
@enroute.rest.query("/users/{user_id}", method="GET")
async def get_user(self, request):
return Response({"id": "123", "name": "John"})
@enroute.rest.command("/users", method="POST")
async def create_user(self, request):
return Response({"id": "456"}, status=201)
# Router manages REST endpoint registration
rest_router = RestHttpRouter()
rest_router.build_routes()Periodic Router
Router implementation for scheduled/periodic task handlers.
class PeriodicRouter(Router):
def __init__(self): ...
routes: Dict[PeriodicEnrouteDecorator, Callable]
def build_routes(self) -> None: ...Usage Examples:
from minos.networks import PeriodicRouter, enroute
class ScheduledTasks:
@enroute.periodic.event("0 */5 * * * *") # Every 5 minutes
async def cleanup_task(self, request):
# Perform cleanup
return Response({"status": "cleanup_completed"})
@enroute.periodic.event("0 0 * * * *") # Every hour
async def hourly_report(self, request):
return Response({"report": "generated"})
# Router manages periodic task registration
periodic_router = PeriodicRouter()
periodic_router.build_routes()Architecture Integration
Routers work together with the decorator system and ports to provide comprehensive routing:
- Discovery Phase: Routers scan for decorated methods during initialization
- Registration Phase: Routes are built and stored in the routes dictionary
- Execution Phase: Ports use routers to dispatch incoming requests to appropriate handlers
- Type Safety: Each router type handles specific decorator types, ensuring proper routing
The routing system enables the framework's unified decorator approach while maintaining clean separation between different transport mechanisms.
Install with Tessl CLI
npx tessl i tessl/pypi-minos-microservice-networks