Extract a single CREATE TABLE / VIEW / FUNCTION DDL node.
(
self,
node,
source: bytes,
file_path_str: str,
nodes: list[NodeInfo],
edges: list[EdgeInfo],
)
| 2097 | self._walk_sql_tree(child, source, file_path_str, nodes, edges) |
| 2098 | |
| 2099 | def _extract_sql_ddl( |
| 2100 | self, |
| 2101 | node, |
| 2102 | source: bytes, |
| 2103 | file_path_str: str, |
| 2104 | nodes: list[NodeInfo], |
| 2105 | edges: list[EdgeInfo], |
| 2106 | ) -> None: |
| 2107 | """Extract a single CREATE TABLE / VIEW / FUNCTION DDL node.""" |
| 2108 | node_type = node.type |
| 2109 | line_start = node.start_point[0] + 1 |
| 2110 | line_end = node.end_point[0] + 1 |
| 2111 | |
| 2112 | # Locate the identifier / object_reference child that holds the name. |
| 2113 | name: Optional[str] = None |
| 2114 | for child in node.children: |
| 2115 | if child.type in ("identifier", "object_reference", "dotted_name"): |
| 2116 | raw = source[child.start_byte: child.end_byte].decode("utf-8", errors="replace") |
| 2117 | # Strip schema prefix (schema.name → name) |
| 2118 | name = raw.strip("`\"").split(".")[-1] |
| 2119 | break |
| 2120 | # Some grammars nest: relation > object_reference > identifier |
| 2121 | if child.type == "relation": |
| 2122 | for gc in child.children: |
| 2123 | if gc.type in ("object_reference", "identifier"): |
| 2124 | raw = source[gc.start_byte: gc.end_byte].decode( |
| 2125 | "utf-8", errors="replace", |
| 2126 | ) |
| 2127 | name = raw.strip("`\"").split(".")[-1] |
| 2128 | break |
| 2129 | if name: |
| 2130 | break |
| 2131 | |
| 2132 | if not name: |
| 2133 | return |
| 2134 | |
| 2135 | if node_type == "create_table": |
| 2136 | kind = "Class" |
| 2137 | sql_kind = "table" |
| 2138 | elif node_type == "create_view": |
| 2139 | kind = "Class" |
| 2140 | sql_kind = "view" |
| 2141 | else: # create_function |
| 2142 | kind = "Function" |
| 2143 | sql_kind = "function" |
| 2144 | |
| 2145 | qualified = f"{file_path_str}::{name}" |
| 2146 | nodes.append(NodeInfo( |
| 2147 | kind=kind, |
| 2148 | name=name, |
| 2149 | file_path=file_path_str, |
| 2150 | line_start=line_start, |
| 2151 | line_end=line_end, |
| 2152 | language="sql", |
| 2153 | extra={"sql_kind": sql_kind}, |
| 2154 | )) |
| 2155 | edges.append(EdgeInfo( |
| 2156 | kind="CONTAINS", |
no test coverage detected