(self)
| 37 | __dialect__ = "default_enhanced" |
| 38 | |
| 39 | def test_nonrecursive(self): |
| 40 | orders = table( |
| 41 | "orders", |
| 42 | column("region"), |
| 43 | column("amount"), |
| 44 | column("product"), |
| 45 | column("quantity"), |
| 46 | ) |
| 47 | |
| 48 | regional_sales = ( |
| 49 | select( |
| 50 | orders.c.region, |
| 51 | func.sum(orders.c.amount).label("total_sales"), |
| 52 | ) |
| 53 | .group_by(orders.c.region) |
| 54 | .cte("regional_sales") |
| 55 | ) |
| 56 | |
| 57 | top_regions = ( |
| 58 | select(regional_sales.c.region) |
| 59 | .where( |
| 60 | regional_sales.c.total_sales |
| 61 | > select( |
| 62 | func.sum(regional_sales.c.total_sales) // 10 |
| 63 | ).scalar_subquery() |
| 64 | ) |
| 65 | .cte("top_regions") |
| 66 | ) |
| 67 | |
| 68 | s = ( |
| 69 | select( |
| 70 | orders.c.region, |
| 71 | orders.c.product, |
| 72 | func.sum(orders.c.quantity).label("product_units"), |
| 73 | func.sum(orders.c.amount).label("product_sales"), |
| 74 | ) |
| 75 | .where(orders.c.region.in_(select(top_regions.c.region))) |
| 76 | .group_by(orders.c.region, orders.c.product) |
| 77 | ) |
| 78 | |
| 79 | # needs to render regional_sales first as top_regions |
| 80 | # refers to it |
| 81 | self.assert_compile( |
| 82 | s, |
| 83 | "WITH regional_sales AS (SELECT orders.region AS region, " |
| 84 | "sum(orders.amount) AS total_sales FROM orders " |
| 85 | "GROUP BY orders.region), " |
| 86 | "top_regions AS (SELECT " |
| 87 | "regional_sales.region AS region FROM regional_sales " |
| 88 | "WHERE regional_sales.total_sales > " |
| 89 | "(SELECT FLOOR(sum(regional_sales.total_sales) / :sum_1) AS " |
| 90 | "anon_1 FROM regional_sales)) " |
| 91 | "SELECT orders.region, orders.product, " |
| 92 | "sum(orders.quantity) AS product_units, " |
| 93 | "sum(orders.amount) AS product_sales " |
| 94 | "FROM orders WHERE orders.region " |
| 95 | "IN (SELECT top_regions.region FROM top_regions) " |
| 96 | "GROUP BY orders.region, orders.product", |
nothing calls this directly
no test coverage detected