"""Probe: postgres.metadata.functions — extract PL/pgSQL function and procedure source."""
from core.base_probe import BaseProbe
from core.registry import register
[docs]
@register
class PgFunctionsProbe(BaseProbe):
[docs]
@staticmethod
def probe_name() -> str:
return "postgres.metadata.functions"
[docs]
@staticmethod
def description() -> str:
return "Extract function/procedure definitions from pg_proc + pg_get_functiondef"
[docs]
def execute(self, connection, schema: str, **kwargs) -> list[dict]:
with connection.cursor() as cur:
cur.execute("""
SELECT p.proname, p.prokind, l.lanname,
pg_get_functiondef(p.oid) AS definition
FROM pg_proc p
JOIN pg_namespace n ON n.oid = p.pronamespace
JOIN pg_language l ON l.oid = p.prolang
WHERE n.nspname = %s AND l.lanname IN ('plpgsql', 'sql')
ORDER BY p.proname
""", (schema,))
return [{
"function_name": r[0],
"kind": {"f": "FUNCTION", "p": "PROCEDURE", "a": "AGGREGATE", "w": "WINDOW"}.get(r[1], r[1]),
"language": r[2],
"source_code": r[3] or "",
} for r in cur.fetchall()]