Plugin API Reference
This section documents the built-in plugins provided by Genro Routes.
For additional reference, see:
Plugin Development Guide - Create custom plugins
Plugin Configuration Guide - Runtime configuration
Auto-Generated Plugin API
AuthPlugin - Authorization plugin with tag-based access control.
This plugin provides tag-based authorization for router entries. It evaluates authorization rules defined on endpoints against user tags.
Usage:
from genro_routes import Router, RoutingClass, route
class MyAPI(RoutingClass):
def __init__(self):
self.api = Router(self, name="api")
@route("api", auth_rule="admin&internal")
def admin_action(self):
return "admin only"
@route("api", auth_rule="public")
def public_action(self):
return "public"
# Query with user's tags (comma-separated list of tags the user has)
obj = MyAPI()
obj.api.node("admin_action", auth_tags="admin,internal") # user has both tags
obj.api.nodes(auth_tags="admin") # user has admin tag
obj.api.nodes(auth_tags="admin,public") # user has admin AND public tags
- Rule syntax (on entry auth_rule):
|: OR (user must have at least one)&: AND (user must have all)!: NOT (user must not have)(): grouping
- NOTE: Comma is NOT allowed in auth_rule. Use
|for OR,&for AND. Comma in auth_tags means the user has multiple tags (always AND).
Example: auth_rule=”admin|manager” means “user must have admin OR manager” Example: auth_rule=”admin&!guest” means “user must have admin AND NOT guest”
- class genro_routes.plugins.auth.AuthPlugin(router, **config)[source]
Bases:
BasePluginAuthorization plugin with tag-based access control.
Evaluates boolean rule expressions against user tags to control access to router entries. Rules are defined per-entry via
@route(auth_rule=...)and checked at runtime against tags provided viaauth_tagsparameter.- Rule syntax (on entry):
|: OR (user must have at least one tag)&: AND (user must have all tags)!: NOT (user must not have tag)(): grouping for complex expressions
- User tags (on query):
Comma-separated string of tags the user possesses. The comma means the user has ALL those tags (implicit AND).
- plugin_code
“auth” - used for registration and config prefix.
- plugin_description
Human-readable description.
Example
Entry definition:
@route("api", auth_rule="admin|manager") # OR def sensitive_action(self): ... @route("api", auth_rule="admin&!guest") # AND + NOT def admin_only(self): ...
Query with user tags:
# User has admin tag -> can access admin|manager entries router.node("sensitive_action", auth_tags="admin") # User has both admin and hr tags router.nodes(auth_tags="admin,hr")
- Parameters:
router (
Any)config (
Any)
- plugin_code: str = 'auth'
- plugin_description: str = 'Authorization plugin with tag-based access control'
- plugin_default_param: str | None = 'rule'
- configure(*, rule='', enabled=True, _target='_all_', flags=None)[source]
Define authorization rule for this entry/router.
- Parameters:
rule (
str) – Boolean rule expression (e.g., “admin&internal”, “!guest”). Use|for OR,&for AND. Comma is not allowed.enabled (
bool) – Whether the plugin is enabled (default True)_target (
str) – Internal - target bucket nameflags (
str|None) – Internal - flag string
- Raises:
ValueError – If rule contains comma (use
|for OR instead).- Return type:
None
- deny_reason(entry, **filters)[source]
Filter entries based on authorization rule.
- Parameters:
entry (
MethodEntry|RouterInterface) – MethodEntry or Router being checked.**filters (
Any) – May containtagswith user’s tags.
- Returns:
Access allowed (no reason to deny). “not_authenticated”: Entry requires tags but none provided. “not_authorized”: Tags provided but don’t match rule.
- Return type:
””
- name
EnvPlugin - Environment capability-based access control plugin.
This plugin provides capability-based filtering for router entries. It evaluates capability requirements defined on endpoints against system capabilities.
Capabilities can come from two sources:
Request capabilities: Passed explicitly via
env_capabilitiesparameterInstance capabilities: Declared on RoutingClass instances via a
CapabilitiesSetsubclass
When traversing a router hierarchy, capabilities are accumulated from all RoutingClass instances along the path. This allows child services to inherit capabilities from their parents while adding their own.
Usage:
from genro_routes import Router, RoutingClass, route
from genro_routes.plugins.env import CapabilitiesSet, capability
class MyCapabilities(CapabilitiesSet):
@capability
def redis(self) -> bool:
return True # Check if redis is available
@capability
def pyjwt(self) -> bool:
return "pyjwt" in sys.modules
class MyAPI(RoutingClass):
def __init__(self):
self.api = Router(self, name="api").plug("env")
self.capabilities = MyCapabilities()
@route("api", env_requires="pyjwt&redis")
def create_jwt(self):
return "jwt created"
@route("api", env_requires="paypal|stripe")
def process_payment(self):
return "payment processed"
# Query with additional request capabilities
obj = MyAPI()
obj.api.node("create_jwt", env_capabilities="pyjwt") # OK: pyjwt from request + redis from instance
obj.api.node("create_jwt") # not_available: only redis from instance, missing pyjwt
# Dynamic capabilities via CapabilitiesSet
class PaymentCapabilities(CapabilitiesSet):
def __init__(self, service):
self._service = service
@capability
def stripe(self) -> bool:
return self._service._stripe_configured
@capability
def paypal(self) -> bool:
return self._service._paypal_configured
class PaymentService(RoutingClass):
def __init__(self):
self.api = Router(self, name="api").plug("env")
self._stripe_configured = True
self._paypal_configured = False
self.capabilities = PaymentCapabilities(self)
- Rule syntax (on entry env_requires):
|: OR (system must have at least one)&: AND (system must have all)!: NOT (system must not have)(): grouping
- NOTE: Comma is NOT allowed in env_requires. Use
|for OR,&for AND. Comma in env_capabilities means the system has multiple capabilities.
Example: env_requires=”pyjwt&redis” means “system must have pyjwt AND redis” Example: env_requires=”paypal|stripe” means “system must have paypal OR stripe”
- class genro_routes.plugins.env.EnvPlugin(router, **config)[source]
Bases:
BasePluginEnvironment capability-based access control plugin.
Controls access to router entries based on system capabilities. Capabilities represent runtime features (installed modules, configured services, etc.) that may or may not be available.
- Capability sources (combined with OR):
Instance capabilities: Declared on RoutingClass via
CapabilitiesSetRequest capabilities: Passed via
env_capabilitiesparameterAccumulated capabilities: Inherited from parent RoutingClass instances
- Rule syntax (on entry via
env_requires): |: OR (system must have at least one capability)&: AND (system must have all capabilities)!: NOT (system must not have capability)(): grouping for complex expressions
- plugin_code
“env” - used for registration and config prefix.
- plugin_description
Human-readable description.
Example
Entry definition:
@route("api", env_requires="redis&pyjwt") # requires both def create_session(self): ... @route("api", env_requires="stripe|paypal") # requires one def process_payment(self): ...
Dynamic capabilities via CapabilitiesSet:
class ServerCaps(CapabilitiesSet): @capability def redis(self) -> bool: return self._redis_client is not None class MyService(RoutingClass): def __init__(self): self.api = Router(self, name="api").plug("env") self.capabilities = ServerCaps()
Query with additional request capabilities:
# Adds pyjwt to instance capabilities for this request router.node("create_session", env_capabilities="pyjwt")
- Parameters:
router (
Any)config (
Any)
- plugin_code: str = 'env'
- plugin_description: str = 'Environment capability-based access control plugin'
- plugin_default_param: str | None = 'requires'
- configure(*, requires='', enabled=True, _target='_all_', flags=None)[source]
Define capability requirements for this entry/router.
- Parameters:
requires (
str) – Boolean rule expression (e.g., “pyjwt&redis”, “paypal|stripe”). Use|for OR,&for AND. Comma is not allowed.enabled (
bool) – Whether the plugin is enabled (default True)_target (
str) – Internal - target bucket nameflags (
str|None) – Internal - flag string
- Raises:
ValueError – If requires contains comma (use
|for OR instead).- Return type:
None
- deny_reason(entry, **filters)[source]
Filter entries based on capability requirements.
Capabilities are accumulated from: 1. Router capabilities (
router_capabilitiesif pre-computed, else from router) 2. Request capabilities (capabilitiesparameter)The combined set is checked against the entry’s
env_requires.- Parameters:
entry (
MethodEntry) – MethodEntry being checked.**filters (
Any) – May containrouter_capabilities(pre-computed) and/orcapabilities(from request).
- Returns:
Access allowed (no reason to deny). “not_available”: Entry requires capabilities but none available,
or capabilities don’t match rule.
- Return type:
””
- name
- class genro_routes.plugins.env.CapabilitiesSet[source]
Bases:
objectBase class for dynamic capability sets.
Subclasses define capabilities as methods decorated with
@capability. The class behaves like a set: supportsin,len, and iteration.Capabilities are evaluated dynamically on each access, allowing for runtime conditions (e.g., time of day, module availability, configuration).
Usage:
from genro_routes.plugins.env import CapabilitiesSet, capability class ServerCapabilities(CapabilitiesSet): @capability def jwt(self) -> bool: return "pyjwt" in sys.modules @capability def send_mail(self) -> bool: hour = datetime.now().hour return 8 <= hour <= 20 caps = ServerCapabilities() "jwt" in caps # True if pyjwt is installed list(caps) # ["jwt", "send_mail"] (only active ones) len(caps) # number of active capabilities
Integration with RoutingClass:
class MyService(RoutingClass): def __init__(self): self.api = Router(self, name="api").plug("env") self.capabilities = ServerCapabilities()
- genro_routes.plugins.env.capability(func)[source]
Mark a method as a capability checker.
The decorated method should return a bool indicating whether the capability is currently active.
Usage:
class ServerCapabilities(CapabilitiesSet): @capability def jwt(self) -> bool: return "jwt" in sys.modules @capability def redis(self) -> bool: return self._redis_client is not None
Logging plugin for Genro Routes.
Wraps handler calls with configurable logging messages including timing.
Configuration
- Accepted keys (router-level or per-handler):
enabled: Gate the plugin entirely (default True)before: Log “start” message (default True)after: Log “end” message with timing (default True)log: Use logger.info() when available (default True)print: Always use print() (default False)
Example:
from genro_routes import Router, RoutingClass, route
class MyService(RoutingClass):
def __init__(self):
self.api = Router(self, name="api").plug("logging")
@route("api")
def hello(self):
return "Hello!"
# Or configure per-handler:
@route("api", logging_after=False)
def fast_handler(self):
return "fast"
- class genro_routes.plugins.logging.LoggingPlugin(router, *, logger=None, **cfg)[source]
Bases:
BasePluginLogging plugin with configurable start/end messages and timing.
Wraps handler invocations with optional “start” and “end” log messages, including execution timing in milliseconds.
- Configuration options:
enabled: Enable/disable the plugin entirely (default True)before: Log “{handler} start” before execution (default True)after: Log “{handler} end (X ms)” after execution (default True)log: Use logger.info() when handlers available (default True)print: Always use print() instead of logger (default False)
- Output sinks:
By default uses Python’s logging module (
logging.getLogger("genro_routes")). Falls back to print() if no handlers are configured on the logger. Setprint=Trueto always use print().
- plugin_code
“logging” - used for registration and config prefix.
- plugin_description
Human-readable description.
Example
Basic usage:
class MyService(RoutingClass): def __init__(self): self.api = Router(self, name="api").plug("logging") @route("api") def hello(self): return "Hello!"
Per-handler configuration:
@route("api", logging_after=False) # disable end message def fast_handler(self): return "fast"
Runtime configuration:
svc.api.logging.configure(before=False) # disable globally svc.api.logging.configure(_target="slow_handler", after=True)
- Parameters:
logger (
Logger|None)
- plugin_code: str = 'logging'
- plugin_description: str = 'Logs handler calls with timing'
- configure(enabled=True, before=True, after=True, log=True, print=False)[source]
Configure logging plugin options.
- Parameters:
enabled (
bool) – Enable/disable the plugin entirely.before (
bool) – Log “{handler} start” before execution.after (
bool) – Log “{handler} end (X ms)” after execution.log (
bool) – Use logger.info() when handlers available.print (
bool) – Always use print() instead of logger.
OpenAPI plugin for Genro Routes.
Provides explicit control over OpenAPI schema generation for handlers. Use this plugin to override automatically guessed HTTP methods or add OpenAPI-specific metadata like tags, summary, description, and security.
Configuration
- Accepted keys (router-level or per-handler):
enabled: Gate the plugin entirely (default True)method: HTTP method override (e.g., “get”, “post”, “delete”)tags: OpenAPI tags (string or list of strings)summary: Summary override for the operationdescription: Description override for the operationdeprecated: Mark the operation as deprecated (default False)security_scheme: Security scheme name (default “BearerAuth”)security: Explicit security override (list, or [] for public)
- Cross-plugin integration:
When pydantic plugin is active, uses pre-computed
response_schemafrom metadata. Falls back to directget_type_hintsextraction otherwise.When auth plugin is active,
securityis auto-derived fromauth_rule.When env plugin is active,
x-requiresis auto-derived fromenv_requires.
Example:
from genro_routes import Router, RoutingClass, route
class MyService(RoutingClass):
def __init__(self):
self.api = Router(self, name="api").plug("openapi").plug("auth")
@route("api", openapi_method="delete")
def delete_item(self, item_id: int) -> dict:
return {"deleted": item_id}
@route("api", openapi_tags=["users"], openapi_deprecated=True)
def old_list(self) -> list:
return []
@route("api", auth_rule="admin")
def admin_only(self) -> dict:
return {}
@route("api", openapi_security=[])
def force_public(self) -> dict:
return {}
- class genro_routes.plugins.openapi.OpenAPIPlugin(router, **config)[source]
Bases:
BasePluginOpenAPI plugin for explicit schema control.
Provides explicit control over OpenAPI schema generation. By default, HTTP methods are guessed from function signatures (GET for scalar params, POST for complex types). Use this plugin to override the guessed method or add OpenAPI-specific metadata.
- Configuration options:
enabled: Enable/disable the plugin (default True)method: HTTP method override (“get”, “post”, “put”, “delete”, “patch”)tags: OpenAPI tags for grouping (string or list of strings)summary: Summary text override for the operationdescription: Description override for the operationdeprecated: Mark the operation as deprecated (default False)security_scheme: Security scheme name (default “BearerAuth”)security: Explicit per-operation security override (list or [])
- Cross-plugin integration:
Auth plugin:
auth_ruleauto-generatessecurityfield.Env plugin:
env_requiresauto-generatesx-requiresextension.
- Method guessing rules (when not overridden):
GET: All parameters are scalar types (str, int, float, bool, Enum)
POST: Any parameter is complex (dict, list, TypedDict, Pydantic model)
- plugin_code
“openapi” - used for registration and config prefix.
- plugin_description
Human-readable description.
Example
Override HTTP method:
@route("api", openapi_method="delete") def remove_item(self, item_id: int) -> dict: return {"deleted": item_id}
Add tags and mark deprecated:
@route("api", openapi_tags=["users", "admin"], openapi_deprecated=True) def old_list_users(self) -> list: return []
Retrieve OpenAPI info:
node = router.node("remove_item", openapi=True) print(node.openapi) # {"delete": {"operationId": "remove_item", ...}}
- Parameters:
router (
Any)config (
Any)
- plugin_code: str = 'openapi'
- plugin_description: str = 'Provides explicit control over OpenAPI schema generation'
- configure(enabled=True, method=None, tags=None, summary=None, description=None, deprecated=False, security_scheme='BearerAuth', security=None)[source]
Configure OpenAPI plugin options.
- Parameters:
enabled (
bool) – Enable/disable the plugin entirely.method (
str|None) – HTTP method override (get, post, put, delete, patch).tags (
str|list[str] |None) – OpenAPI tags for grouping operations.summary (
str|None) – Summary text override for the operation.description (
str|None) – Description override for the operation.deprecated (
bool) – Mark the operation as deprecated.security_scheme (
str) – Name of the security scheme for auth-based security derivation (default “BearerAuth”).security (
list|None) – Explicit security override for the operation. Use [] to force public, or [{“OAuth2”: [“read”]}] for custom.
- entry_metadata(router, entry)[source]
Provide OpenAPI-specific metadata for a handler.
- Return type:
dict[str,Any]- Returns:
Dict containing openapi configuration for this handler.
- Parameters:
router (
Any)entry (
MethodEntry)
- name
- class genro_routes.plugins.openapi.OpenAPITranslator[source]
Bases:
objectTranslator for converting router nodes to OpenAPI format.
This class provides static methods to translate the output of
router.nodes()to OpenAPI-compatible formats.- Modes for nodes():
openapi: Flat format with all paths merged into a single paths dict.h_openapi: Hierarchical format preserving the router tree structure.
- For single entry info:
entry_to_openapi(entry): Convert a MethodEntry to OpenAPI path item. Used byrouter.node(path, openapi=True)to populate theopenapiattribute on the returned RouterNode.
Example:
node = router.node("my_handler", openapi=True) print(node.openapi) # {"get": {"operationId": "my_handler", ...}}
- static translate_openapi(nodes_data, lazy=False, path_prefix='')[source]
Translate nodes() output to flat OpenAPI format.
- Parameters:
nodes_data (
dict[str,Any]) – Output from nodes() in standard format.lazy (
bool) – If True, child routers are returned as router references.path_prefix (
str) – Prefix for generated paths (used internally for recursion).
- Return type:
dict[str,Any]- Returns:
Dict with “paths” containing OpenAPI path items (flat structure), “$defs” containing all type definitions (if any nested types), and “routers” for children in lazy mode.
- static translate_h_openapi(nodes_data, lazy=False, path_prefix='')[source]
Translate nodes() output to hierarchical OpenAPI format.
Unlike translate_openapi which flattens all paths, this preserves the router hierarchy while converting entries to OpenAPI format.
- Parameters:
nodes_data (
dict[str,Any]) – Output from nodes() in standard format.lazy (
bool) – If True, child routers are returned as router references.path_prefix (
str) – Prefix for generated paths (used when called with basepath).
- Return type:
dict[str,Any]- Returns:
Dict with “paths” containing local OpenAPI path items, “$defs” containing all type definitions (if any nested types), and “routers” containing nested h_openapi structures for children.
- static entry_info_to_openapi(name, entry_info)[source]
Convert entry info dict to OpenAPI path item format.
HTTP method determination priority: 1. Explicit override via openapi plugin config (openapi_method in metadata) 2. Guessed from function signature (guess_http_method)
- Return type:
tuple[dict[str,Any],dict[str,Any]]- Returns:
Tuple of (path_item, defs) where defs contains any $defs extracted from schemas (for nested types like TypedDict).
- Parameters:
name (
str)entry_info (
dict[str,Any])
- static create_pydantic_model_for_func(func)[source]
Create a pydantic model from function type hints.
This is used when the pydantic plugin is not active but we still want to extract parameter schema for OpenAPI.
- Parameters:
func (
Callable) – The callable to analyze.- Return type:
Any|None- Returns:
A pydantic model class, or None if no type hints available.
- static schema_to_parameters(schema)[source]
Convert pydantic JSON schema to OpenAPI query parameters.
- Parameters:
schema (
dict[str,Any]) – Pydantic model JSON schema.- Return type:
list[dict[str,Any]]- Returns:
List of OpenAPI parameter objects for query string.
- static python_type_to_openapi_schema(python_type)[source]
Convert Python type to OpenAPI schema dict using pydantic.
Uses pydantic’s TypeAdapter to generate JSON schema for any type.
- Parameters:
python_type (
Any)- Return type:
dict[str,Any]
- SCALAR_TYPES: set[type] = {<class 'NoneType'>, <class 'bool'>, <class 'float'>, <class 'int'>, <class 'str'>}
- static guess_http_method(func)[source]
Guess HTTP method from function signature.
Rules: - GET if all parameters are scalar types (str, int, float, bool, Enum) - POST if any parameter is complex (dict, list, TypedDict, models) - POST on error (safer fallback)
Scalar types can be serialized as query string parameters. Complex types require a request body.
Examples
def health() -> dict: # GET - no params def get_user(id: int) -> dict: # GET - scalar param def search(q: str, limit: int): # GET - all scalars def create(data: dict): # POST - complex param def update(user: UserModel): # POST - model param
- Parameters:
func (
Callable) – The callable to analyze.- Return type:
str- Returns:
“get” or “post” based on signature analysis.
- static entry_to_openapi(entry)[source]
Convert a MethodEntry to OpenAPI format.
This is a convenience method for use with router.node(openapi=True).
- Parameters:
entry (
MethodEntry) – The MethodEntry to convert.- Returns:
{…}} or {“post”: {…}}).
- Return type:
Dict with the OpenAPI path item (e.g., {“get”
Pydantic validation and response schema plugin for Genro Routes.
Validates handler inputs using Pydantic type hints and generates JSON Schema from return type annotations for bridge consumption (MCP, OpenAPI).
At registration time (on_decore):
- Inspects parameter type hints and builds a Pydantic model for input validation.
- Inspects return type annotation and generates a JSON Schema via TypeAdapter,
stored in
entry.metadata["pydantic"]["response_schema"].
At call time (wrap_handler), validates annotated args/kwargs before calling
the real handler.
Example:
from typing import TypedDict
from genro_routes import Router, RoutingClass, route
class UserResponse(TypedDict):
id: int
name: str
class MyService(RoutingClass):
def __init__(self):
self.api = Router(self, name="api").plug("pydantic")
@route("api")
def get_user(self, user_id: int) -> UserResponse:
return {"id": user_id, "name": "alice"}
svc = MyService()
svc.api.node("get_user")(user_id=123) # OK, validated
svc.api.node("get_user")(user_id="not_an_int") # ValidationError
# Response schema available in metadata
entry = svc.api._entries["get_user"]
entry.metadata["pydantic"]["response_schema"]
# {"type": "object", "properties": {"id": ..., "name": ...}, ...}
Configuration:
# Disable validation for a specific handler
@route("api", pydantic_disabled=True)
def unvalidated_handler(self):
pass
- class genro_routes.plugins.pydantic.PydanticPlugin(router, **config)[source]
Bases:
BasePluginValidate handler inputs and generate response schemas with Pydantic.
At registration time (
on_decore), builds a Pydantic model from parameter type hints for input validation, and generates a JSON Schema from the return type annotation viaTypeAdapterfor bridge consumption.- Behavior:
Only annotated parameters are validated
Unannotated parameters pass through unchanged
ValidationError is raised on invalid input
Return type annotations produce
response_schemain metadataCan be disabled per-handler via
pydantic_disabled=True
- Configuration options:
disabled: Skip validation for this handler/router (default False)
- plugin_code
“pydantic” - used for registration and config prefix.
- plugin_description
Human-readable description.
Example
Basic usage:
class MyService(RoutingClass): def __init__(self): self.api = Router(self, name="api").plug("pydantic") @route("api") def get_user(self, user_id: int) -> dict[str, int]: return {"id": user_id} svc = MyService() svc.api.node("get_user")(user_id=123) # OK svc.api.node("get_user")(user_id="not_an_int") # ValidationError # Response schema in metadata svc.api._entries["get_user"].metadata["pydantic"]["response_schema"]
Disable validation:
@route("api", pydantic_disabled=True) def unvalidated(self, data): return data # no validation
- Parameters:
config (
Any)
- plugin_code: str = 'pydantic'
- plugin_description: str = 'Validates inputs and generates response schemas using Pydantic'
- configure(disabled=False)[source]
Configure pydantic plugin options.
- Parameters:
disabled (
bool) – If True, skip validation for this handler/router.
- on_decore(route, func, entry)[source]
Build Pydantic model from handler type hints and generate response schema.
- Parameters:
route (
Router)func (
Callable)entry (
MethodEntry)
- Return type:
None
- wrap_handler(route, entry, call_next)[source]
Validate annotated parameters with the cached Pydantic model before calling.
- Parameters:
route (
Router)entry (
MethodEntry)call_next (
Callable)
- get_model(entry)[source]
Return the Pydantic model for this handler if not disabled.
- Parameters:
entry (
MethodEntry) – The MethodEntry to get the model for.- Return type:
tuple[str,Any] |None- Returns:
Tuple of (“pydantic_model”, model_class) if available, else None.
- name