14f96e78f9
Add SQL bridge, support functions, cron jobs, error handling, and PyPI version pinning so Python triggers match PHP capabilities. Co-authored-by: Cursor <cursoragent@cursor.com>
121 lines
2.9 KiB
Python
121 lines
2.9 KiB
Python
"""Database runtime for sandboxed Python triggers (mirrors sys/sys.php via sql_bridge.php)."""
|
|
|
|
import json
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
from typing import Any, Optional
|
|
|
|
|
|
def _project_root() -> str:
|
|
return os.getcwd()
|
|
|
|
|
|
def _php_binary() -> str:
|
|
for candidate in (
|
|
os.environ.get("REACTIVE_PHP_BIN"),
|
|
"/usr/bin/php",
|
|
shutil.which("php"),
|
|
):
|
|
if candidate and os.path.isfile(candidate):
|
|
return candidate
|
|
if candidate and shutil.which(candidate):
|
|
return shutil.which(candidate) # type: ignore[return-value]
|
|
raise RuntimeError("PHP binary not found for sql_bridge")
|
|
|
|
|
|
def _bridge(action: str, payload: dict) -> Any:
|
|
bridge = os.path.join(_project_root(), "sys", "sql_bridge.php")
|
|
proc = subprocess.run(
|
|
[_php_binary(), bridge],
|
|
input=json.dumps({"action": action, **payload}),
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
if proc.returncode != 0:
|
|
err = proc.stderr.strip()
|
|
try:
|
|
err_obj = json.loads(err)
|
|
err = err_obj.get("error", err)
|
|
except json.JSONDecodeError:
|
|
pass
|
|
raise RuntimeError(err or "sql_bridge failed")
|
|
data = json.loads(proc.stdout)
|
|
return data["result"]
|
|
|
|
|
|
def sql(
|
|
query: str,
|
|
db_host: Optional[str] = None,
|
|
db_user: Optional[str] = None,
|
|
db_pass: Optional[str] = None,
|
|
db_name: Optional[str] = None,
|
|
) -> Any:
|
|
return _bridge(
|
|
"sql",
|
|
{
|
|
"query": query,
|
|
"db_host": db_host,
|
|
"db_user": db_user,
|
|
"db_pass": db_pass,
|
|
"db_name": db_name,
|
|
},
|
|
)
|
|
|
|
|
|
def sql_read(
|
|
query: str,
|
|
db_host: Optional[str] = None,
|
|
db_user: Optional[str] = None,
|
|
db_pass: Optional[str] = None,
|
|
db_name: Optional[str] = None,
|
|
) -> Any:
|
|
return _bridge(
|
|
"sql_read",
|
|
{
|
|
"query": query,
|
|
"db_host": db_host,
|
|
"db_user": db_user,
|
|
"db_pass": db_pass,
|
|
"db_name": db_name,
|
|
},
|
|
)
|
|
|
|
|
|
def sql_write(
|
|
query: str,
|
|
db_host: Optional[str] = None,
|
|
db_user: Optional[str] = None,
|
|
db_pass: Optional[str] = None,
|
|
db_name: Optional[str] = None,
|
|
) -> Any:
|
|
return _bridge(
|
|
"sql_write",
|
|
{
|
|
"query": query,
|
|
"db_host": db_host,
|
|
"db_user": db_user,
|
|
"db_pass": db_pass,
|
|
"db_name": db_name,
|
|
},
|
|
)
|
|
|
|
|
|
def sql_read_and_hydrate(query: str) -> Any:
|
|
return _bridge("sql_read_and_hydrate", {"query": query})
|
|
|
|
|
|
def get_tbl_primary_key_col_name(table: str) -> list:
|
|
return _bridge("get_tbl_primary_key_col_name", {"table": table})
|
|
|
|
|
|
def run_process(cmd: str, stdin: str = "") -> tuple[int, str, str]:
|
|
proc = subprocess.run(
|
|
cmd,
|
|
input=stdin,
|
|
shell=True,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
return proc.returncode, proc.stdout, proc.stderr
|