Test subquery operations
(self)
| 377 | session.close() |
| 378 | |
| 379 | def test_subqueries(self): |
| 380 | """Test subquery operations""" |
| 381 | session = self.Session() |
| 382 | try: |
| 383 | # Clean up from previous tests |
| 384 | session.query(Post).delete() |
| 385 | session.query(User).delete() |
| 386 | session.commit() |
| 387 | |
| 388 | # Create test data |
| 389 | for i in range(5): |
| 390 | user = User(username=f"sub{i}", email=f"sub{i}@example.com", age=20+i*5) |
| 391 | for j in range(i): |
| 392 | post = Post(title=f"Post {i}-{j}", author=user) |
| 393 | session.add(post) |
| 394 | session.add(user) |
| 395 | session.commit() |
| 396 | |
| 397 | # Subquery for users with posts |
| 398 | subq = session.query(Post.author_id).distinct().subquery() |
| 399 | users_with_posts = session.query(User).filter( |
| 400 | User.id.in_(subq) |
| 401 | ).all() |
| 402 | assert len(users_with_posts) == 4 # user0 has no posts |
| 403 | |
| 404 | # Correlated subquery |
| 405 | users_with_many_posts = session.query(User).filter( |
| 406 | session.query(func.count(Post.id)).filter( |
| 407 | Post.author_id == User.id |
| 408 | ).scalar_subquery() >= 2 |
| 409 | ).all() |
| 410 | assert len(users_with_many_posts) == 3 |
| 411 | |
| 412 | finally: |
| 413 | session.close() |
| 414 | |
| 415 | def run_all_tests(self): |
| 416 | """Run all tests""" |