(self)
| 97 | ) |
| 98 | |
| 99 | def test_recursive(self): |
| 100 | parts = table( |
| 101 | "parts", column("part"), column("sub_part"), column("quantity") |
| 102 | ) |
| 103 | |
| 104 | included_parts = ( |
| 105 | select(parts.c.sub_part, parts.c.part, parts.c.quantity) |
| 106 | .where(parts.c.part == "our part") |
| 107 | .cte(recursive=True) |
| 108 | ) |
| 109 | |
| 110 | incl_alias = included_parts.alias() |
| 111 | parts_alias = parts.alias() |
| 112 | included_parts = included_parts.union( |
| 113 | select( |
| 114 | parts_alias.c.sub_part, |
| 115 | parts_alias.c.part, |
| 116 | parts_alias.c.quantity, |
| 117 | ).where(parts_alias.c.part == incl_alias.c.sub_part) |
| 118 | ) |
| 119 | |
| 120 | s = ( |
| 121 | select( |
| 122 | included_parts.c.sub_part, |
| 123 | func.sum(included_parts.c.quantity).label("total_quantity"), |
| 124 | ) |
| 125 | .select_from( |
| 126 | included_parts.join( |
| 127 | parts, included_parts.c.part == parts.c.part |
| 128 | ) |
| 129 | ) |
| 130 | .group_by(included_parts.c.sub_part) |
| 131 | ) |
| 132 | self.assert_compile( |
| 133 | s, |
| 134 | "WITH RECURSIVE anon_1(sub_part, part, quantity) " |
| 135 | "AS (SELECT parts.sub_part AS sub_part, parts.part " |
| 136 | "AS part, parts.quantity AS quantity FROM parts " |
| 137 | "WHERE parts.part = :part_1 UNION " |
| 138 | "SELECT parts_1.sub_part AS sub_part, " |
| 139 | "parts_1.part AS part, parts_1.quantity " |
| 140 | "AS quantity FROM parts AS parts_1, anon_1 AS anon_2 " |
| 141 | "WHERE parts_1.part = anon_2.sub_part) " |
| 142 | "SELECT anon_1.sub_part, " |
| 143 | "sum(anon_1.quantity) AS total_quantity FROM anon_1 " |
| 144 | "JOIN parts ON anon_1.part = parts.part " |
| 145 | "GROUP BY anon_1.sub_part", |
| 146 | ) |
| 147 | |
| 148 | # quick check that the "WITH RECURSIVE" varies per |
| 149 | # dialect |
| 150 | self.assert_compile( |
| 151 | s, |
| 152 | "WITH anon_1(sub_part, part, quantity) " |
| 153 | "AS (SELECT parts.sub_part AS sub_part, parts.part " |
| 154 | "AS part, parts.quantity AS quantity FROM parts " |
| 155 | "WHERE parts.part = :part_1 UNION " |
| 156 | "SELECT parts_1.sub_part AS sub_part, " |
nothing calls this directly
no test coverage detected