test #7845, #9009
(
self, joins_implicitly, additional_transformation, type_
)
| 202 | "type_", ["table_valued", "table_valued_derived", "column_valued"] |
| 203 | ) |
| 204 | def test_fn_valued( |
| 205 | self, joins_implicitly, additional_transformation, type_ |
| 206 | ): |
| 207 | """test #7845, #9009""" |
| 208 | |
| 209 | my_table = table( |
| 210 | "tbl", |
| 211 | column("id", Integer), |
| 212 | column("data", JSON()), |
| 213 | ) |
| 214 | |
| 215 | sub_dict = my_table.c.data["d"] |
| 216 | |
| 217 | if type_.table_valued or type_.table_valued_derived: |
| 218 | tv = func.json_each(sub_dict) |
| 219 | |
| 220 | tv = tv.table_valued("key", joins_implicitly=joins_implicitly) |
| 221 | |
| 222 | if type_.table_valued_derived: |
| 223 | tv = tv.render_derived(name="tv", with_types=True) |
| 224 | |
| 225 | if additional_transformation.alias: |
| 226 | tv = tv.alias() |
| 227 | |
| 228 | has_key = tv.c.key == "f" |
| 229 | stmt = select(my_table.c.id).where(has_key) |
| 230 | elif type_.column_valued: |
| 231 | tv = func.json_array_elements(sub_dict) |
| 232 | |
| 233 | if additional_transformation.alias: |
| 234 | tv = tv.alias(joins_implicitly=joins_implicitly).column |
| 235 | else: |
| 236 | tv = tv.column_valued("key", joins_implicitly=joins_implicitly) |
| 237 | |
| 238 | stmt = select(my_table.c.id, tv) |
| 239 | else: |
| 240 | type_.fail() |
| 241 | |
| 242 | froms, start = find_unmatching_froms(stmt, my_table) |
| 243 | |
| 244 | if joins_implicitly: |
| 245 | is_(start, None) |
| 246 | is_(froms, None) |
| 247 | elif type_.column_valued: |
| 248 | assert start == my_table |
| 249 | assert froms == {tv.scalar_alias} |
| 250 | |
| 251 | elif type_.table_valued or type_.table_valued_derived: |
| 252 | assert start == my_table |
| 253 | assert froms == {tv} |
| 254 | else: |
| 255 | type_.fail() |
| 256 | |
| 257 | def test_count_non_eq_comparison_operators(self): |
| 258 | query = select(self.a).where(self.a.c.col_a > self.b.c.col_b) |
nothing calls this directly
no test coverage detected