Skip to main content

Convenience wrapper for exonware-xwaction - provides 'import xwaction' alias

Project description

xwaction

Decorator-first automation and workflow foundation for real apps. xwaction lets you define one function once, then run it as a native call, API endpoint, background task, workflow step, or bot/command action - with validation, security, monitoring, and OpenAPI-ready metadata built in.

Company: eXonware.com · Author: eXonware Backend Team · Email: connect@exonware.com

Status Python License


📦 Install

pip install exonware-xwaction
pip install exonware-xwaction[lazy]
pip install exonware-xwaction[full]

🚀 Quick start

from exonware.xwaction import XWAction, ActionContext

@XWAction(profile="query")
def my_action(x: int) -> int:
    return x * 2

my_action(3)  # or my_action.execute(ActionContext(actor="user", source="cli"), x=3)

See docs/ for pipelines, validation, and any REF_* files.


🧩 Engine and handler coverage

xwaction is not only a server decorator. The codebase includes:

  • Execution engines: native, fastapi, flask, celery, prefect
  • Cross-cutting handlers: validation, security, monitoring, workflow
  • OpenAPI generation: per action (to_openapi()) and registry-level (export_openapi_spec(...))
  • Authorization contract: pluggable IActionAuthorizer + AuthzDecision
  • Action discovery/loading: extract_actions(...), load_actions(...) for class/instance action wiring
  • Event-driven execution style: actions can be triggered by HTTP events, command/chat events, and async task/workflow execution paths
  • In-context query execution: XWAction.query(...) lets actions run xwquery logic inside action workflows

🤖 Automation + event-driven positioning

  • xwaction is the base layer for action automation across the eXonware stack.
  • You can wire the same action into event sources (API requests, bot commands, internal triggers) without rewriting business logic.
  • Handler phases (BEFORE, AFTER, ERROR, FINALLY) provide an event-style lifecycle around each action execution.
  • Query-in-action support (XWAction.query) makes it a strong orchestration layer for LLM/agent/server scenarios where tool calls need controlled execution context.
  • Validation + security + authorization hooks provide safer boundaries for agentic/server-side automation.

✨ What you get

Area What's in it
Decorator-first DX Define once with @XWAction, execute directly or via engine adapters.
Validation xwschema-backed input/output validation with caching and handler pipeline support.
Security Role checks, pluggable authorizer, auth token hooks, rate-limit and audit flow support.
Workflows Workflow state/checkpoints, rollback paths, and handler-driven orchestration.
In-context queries Run query execution inside actions via XWAction.query(...) (powered by xwquery).
Monitoring Per-action timing/error metrics, thresholds, and alert hooks.
API docs OpenAPI operation metadata and full spec export from the registry.
Multi-host Native function call, FastAPI/Flask endpoints, Celery tasks, Prefect flows.

📘 Better developer examples

1) Action with schema validation + OpenAPI metadata

from exonware.xwaction import XWAction, ActionContext
from exonware.xwschema import XWSchema

@XWAction(
    profile="endpoint",
    operationId="createInvoice",
    summary="Create invoice",
    tags=["billing"],
    in_types={
        "customer_id": XWSchema({"type": "string", "minLength": 3}),
        "total": XWSchema({"type": "number", "minimum": 0}),
    },
    out_types={
        "return": XWSchema({"type": "object"})
    },
)
def create_invoice(customer_id: str, total: float):
    return {"invoice_id": "inv_001", "customer_id": customer_id, "total": total}

ctx = ActionContext(actor="billing-service", source="internal")
result = create_invoice.execute(ctx, customer_id="cust_123", total=99.5)
openapi_operation = create_invoice.xwaction.to_openapi()

2) Expose class actions as FastAPI endpoints

from fastapi import FastAPI
from exonware.xwaction import XWAction, extract_actions
from exonware.xwaction.engines.fastapi import FastAPIActionEngine

class UserActions:
    @XWAction(profile="endpoint", method="GET", summary="Health check")
    def health(self):
        return {"ok": True}

    @XWAction(profile="endpoint", method="POST", summary="Create user")
    def create_user(self, email: str):
        return {"id": "u_1", "email": email}

app = FastAPI()
engine = FastAPIActionEngine()
engine.setup({"app": app})

for action in extract_actions(UserActions):
    route = f"/users/{action.api_name}"
    method = "GET" if action.api_name == "health" else "POST"
    engine.register_action(action, app, path=route, method=method)

3) Security-focused action with custom authorizer

from exonware.xwaction import XWAction, ActionContext, AuthzDecision

class RBACAuthorizer:
    def authorize(self, action, context):
        roles = context.metadata.get("roles", [])
        required = getattr(action, "roles", [])
        allowed = (not required) or any(r in roles for r in required)
        return AuthzDecision(allowed=allowed, reason="role_check", roles=roles)

@XWAction(
    profile="command",
    roles=["admin"],
    security="bearer",
    rate_limit="60/min",
    audit=True,
    handlers=["security", "monitoring"],
)
def rotate_api_keys():
    return {"status": "rotated"}

rotate_api_keys.xwaction.set_authorizer(RBACAuthorizer())
ctx = ActionContext(actor="ops-user", source="cli", metadata={"roles": ["admin"]})
result = rotate_api_keys.execute(ctx)

4) Command/chat style action metadata

from exonware.xwaction import XWAction, extract_actions, ActionContext

class BotCommands:
    @XWAction(profile="command", cmd_shortcut="price")
    def get_price(self, symbol: str):
        return {"symbol": symbol, "price": 123.45}

bot = BotCommands()
commands = {
    a.cmd_shortcut: a
    for a in extract_actions(BotCommands)
    if getattr(a, "cmd_shortcut", None)
}

ctx = ActionContext(actor="telegram-user", source="chat")
result = commands["price"].execute(ctx, bot, symbol="XW")

5) Export one OpenAPI spec for all registered actions

from exonware.xwaction import ActionRegistry

spec = ActionRegistry.export_openapi_spec(
    title="My Action API",
    version="1.0.0",
    description="Generated from registered @XWAction definitions",
)

6) Query execution inside action context

from exonware.xwaction import XWAction, ActionContext

@XWAction(profile="query", handlers=["validation", "security"])
def find_high_value_users(data: dict):
    return XWAction.query(
        "SELECT * FROM users WHERE spend > 1000",
        data,
        format="sql",
    )

ctx = ActionContext(actor="agent-service", source="automation", metadata={"roles": ["analyst"]})
result = find_high_value_users.execute(ctx, data={"users": [...]})

🌐 Ecosystem functional contributions

xwaction is the orchestration layer; sibling XW libraries provide the typed contracts it executes against. You can use xwaction standalone for decorator-based execution and engine adapters. Additional XW integrations are optional and are primarily useful when building enterprise and mission-critical automation infrastructure under your own operational control.

Supporting XW lib What it provides to xwaction Functional requirement it satisfies
XWSchema Schema-backed input/output validation for actions and OpenAPI metadata generation. Contract-safe action invocation and predictable payload validation.
XWQuery Query execution surface callable from action context (XWAction.query). In-action data retrieval/transform workflows without custom query adapters.
XWSystem Shared runtime primitives (logging, errors, async/utilities, base object/contracts). Engine-neutral action behavior and consistent operational semantics.
XWAuth Authorization/token context integration for protected actions. Enforceable security boundaries for command/endpoint automation paths.
XWEntity Entity-bound action workflows over domain objects. Domain-centric automation instead of function-only orchestration.
XWAPI Endpoint hosting pattern and middleware ecosystem for action-exposed APIs. Production API deployment for actions with standardized middleware/error handling.

Competitive edge: action definitions stay portable across native calls, APIs, workers, and workflow engines because validation, query, and security contracts are provided by shared XW layers.


📖 Docs and tests


📜 License and links

Apache-2.0 - see LICENSE. Homepage: https://exonware.com · Repository: https://github.com/exonware/xwaction

⏱️ Async Support

  • xwaction includes asynchronous execution paths in production code.
  • Source validation: 15 async def definitions and 26 await usages under src/.
  • Use async APIs for I/O-heavy or concurrent workloads to improve throughput and responsiveness.

Version: 0.9.0.16 | Updated: 25-Apr-2026

Built with ❤️ by eXonware.com - Revolutionizing Python Development Since 2025

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

xwaction-0.9.0.16.tar.gz (105.7 kB view details)

Uploaded Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

xwaction-0.9.0.16-py3-none-any.whl (13.3 kB view details)

Uploaded Python 3

File details

Details for the file xwaction-0.9.0.16.tar.gz.

File metadata

  • Download URL: xwaction-0.9.0.16.tar.gz
  • Upload date:
  • Size: 105.7 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/6.2.0 CPython/3.12.13

File hashes

Hashes for xwaction-0.9.0.16.tar.gz
Algorithm Hash digest
SHA256 2ec32a5b2017cc65a2ee8f2144f06364a6f6e2d57a72c11cca628dd68fef2d05
MD5 33b66203ab0a455cc8407f4522b7453c
BLAKE2b-256 0f061fe2531e218ffafee7683c125aa59ade5d3a7874e50ca14cf787c2287c2f

See more details on using hashes here.

File details

Details for the file xwaction-0.9.0.16-py3-none-any.whl.

File metadata

  • Download URL: xwaction-0.9.0.16-py3-none-any.whl
  • Upload date:
  • Size: 13.3 kB
  • Tags: Python 3
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/6.2.0 CPython/3.12.13

File hashes

Hashes for xwaction-0.9.0.16-py3-none-any.whl
Algorithm Hash digest
SHA256 20acb86f6cf2db420c06d8e4f5c08ab9323938c468db9cc5935a3c265f152eb5
MD5 ad94c7a5382da28c2089df0ecce24790
BLAKE2b-256 1357c0f2c785fa39475b7d9faa8637dd34a70fac3ba104956db524b503157652

See more details on using hashes here.

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Depot Continuous Integration Fastly CDN Google Download Analytics Pingdom Monitoring Sentry Error logging StatusPage Status page