not using shim
This commit is contained in:
@@ -3,19 +3,23 @@
|
||||
Hello Action - Python Example Pack
|
||||
|
||||
A minimal Python action that returns "Hello, Python".
|
||||
Demonstrates the basic structure of a Python action in Attune.
|
||||
Demonstrates the basic structure of a self-contained action in Attune.
|
||||
|
||||
Actions receive parameters as JSON on stdin and write results to stdout.
|
||||
"""
|
||||
|
||||
import json
|
||||
import sys
|
||||
|
||||
|
||||
def run(**kwargs):
|
||||
"""Return a simple greeting message."""
|
||||
return {"message": "Hello, Python"}
|
||||
def main():
|
||||
# Read parameters from stdin (JSON format)
|
||||
params = json.loads(sys.stdin.readline())
|
||||
name = params.get("name", "Python")
|
||||
|
||||
result = {"message": f"Hello, {name}"}
|
||||
print(json.dumps(result))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
result = run()
|
||||
print(json.dumps({"result": result, "status": "success"}))
|
||||
sys.exit(0)
|
||||
main()
|
||||
|
||||
@@ -3,20 +3,34 @@
|
||||
HTTP Example Action - Python Example Pack
|
||||
|
||||
Demonstrates using the `requests` library to make an HTTP call to example.com.
|
||||
Receives parameters via stdin JSON (through the Python wrapper).
|
||||
Receives parameters via stdin as JSON.
|
||||
"""
|
||||
|
||||
import requests
|
||||
import json
|
||||
import sys
|
||||
import urllib.request
|
||||
|
||||
|
||||
def run(url="https://example.com", **kwargs):
|
||||
"""Fetch a URL and return status and a snippet of the response body."""
|
||||
response = requests.get(url, timeout=10)
|
||||
def main():
|
||||
# Read parameters from stdin (JSON format)
|
||||
params = json.loads(sys.stdin.readline())
|
||||
url = params.get("url", "https://example.com")
|
||||
|
||||
return {
|
||||
"status_code": response.status_code,
|
||||
"url": response.url,
|
||||
"content_length": len(response.text),
|
||||
"snippet": response.text[:500],
|
||||
"success": response.ok,
|
||||
req = urllib.request.Request(url)
|
||||
with urllib.request.urlopen(req, timeout=10) as response:
|
||||
text = response.read().decode("utf-8")
|
||||
status_code = response.status
|
||||
final_url = response.url
|
||||
|
||||
result = {
|
||||
"status_code": status_code,
|
||||
"url": final_url,
|
||||
"content_length": len(text),
|
||||
"snippet": text[:500],
|
||||
"success": 200 <= status_code < 400,
|
||||
}
|
||||
print(json.dumps(result))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
||||
@@ -5,46 +5,26 @@ Read Counter Action - Python Example Pack
|
||||
Consumes a counter value (typically from the counter sensor trigger payload)
|
||||
and returns a formatted message containing the counter value.
|
||||
|
||||
Parameters are delivered via stdin as JSON from the Python wrapper.
|
||||
Parameters are delivered via stdin as JSON.
|
||||
"""
|
||||
|
||||
import json
|
||||
import sys
|
||||
|
||||
|
||||
def run(counter=0, rule_ref="unknown", **kwargs):
|
||||
"""Return a message containing the counter value.
|
||||
def main():
|
||||
# Read parameters from stdin (JSON format)
|
||||
params = json.loads(sys.stdin.readline())
|
||||
|
||||
Args:
|
||||
counter: The counter value from the trigger payload.
|
||||
rule_ref: The rule reference that produced this counter.
|
||||
**kwargs: Additional parameters (ignored).
|
||||
counter = params.get("counter", 0)
|
||||
rule_ref = params.get("rule_ref", "unknown")
|
||||
|
||||
Returns:
|
||||
dict with a formatted message and the raw counter value.
|
||||
"""
|
||||
return {
|
||||
result = {
|
||||
"message": f"Counter value is {counter} (from rule: {rule_ref})",
|
||||
"counter": counter,
|
||||
"rule_ref": rule_ref,
|
||||
}
|
||||
|
||||
|
||||
def main():
|
||||
"""Entry point when run directly (without the Python wrapper)."""
|
||||
try:
|
||||
content = sys.stdin.read().strip()
|
||||
if content:
|
||||
parts = content.split("---ATTUNE_PARAMS_END---")
|
||||
params = json.loads(parts[0].strip()) if parts[0].strip() else {}
|
||||
else:
|
||||
params = {}
|
||||
except (json.JSONDecodeError, IndexError):
|
||||
params = {}
|
||||
|
||||
result = run(**params)
|
||||
print(json.dumps(result, indent=2))
|
||||
sys.exit(0)
|
||||
print(json.dumps(result))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
Reference in New Issue
Block a user