"""Probe: oracle.metadata.source_code — extract PL/SQL source from ALL_SOURCE."""
from core.base_probe import BaseProbe
from core.registry import register
[docs]
@register
class SourceCodeProbe(BaseProbe):
[docs]
@staticmethod
def probe_name() -> str:
return "oracle.metadata.source_code"
[docs]
@staticmethod
def description() -> str:
return "Extract PL/SQL source code (procedures, functions, packages, triggers) from ALL_SOURCE"
[docs]
def execute(self, connection, schema: str, **kwargs) -> list[dict]:
with connection.cursor() as cur:
cur.execute(
"""SELECT name, type, line, text
FROM all_source
WHERE owner = :schema
AND type IN ('PROCEDURE', 'FUNCTION', 'PACKAGE BODY', 'PACKAGE', 'TRIGGER')
ORDER BY name, type, line""",
schema=schema.upper(),
)
# Assemble lines into full source per object
objects: dict[tuple[str, str], list[str]] = {}
for name, obj_type, line, text in cur.fetchall():
key = (name, obj_type)
objects.setdefault(key, []).append(text or "")
return [
{
"object_name": name,
"object_type": obj_type,
"line_count": len(lines),
"source_code": "".join(lines),
}
for (name, obj_type), lines in sorted(objects.items())
]