feat: add join to crud functions (#21)

tests/test_crud.py

@@ -10,6 +10,9 @@ from fastapi_toolsets.crud.factory import AsyncCrud
 from fastapi_toolsets.exceptions import NotFoundError
 
 from .conftest import (
+    Post,
+    PostCreate,
+    PostCrud,
     Role,
     RoleCreate,
     RoleCrud,
@@ -481,3 +484,271 @@ class TestCrudPaginate:
 
         names = [r.name for r in result["data"]]
         assert names == ["alpha", "bravo", "charlie"]
+
+
+class TestCrudJoins:
+    """Tests for CRUD operations with joins."""
+
+    @pytest.mark.anyio
+    async def test_get_with_join(self, db_session: AsyncSession):
+        """Get with inner join filters correctly."""
+        # Create user with posts
+        user = await UserCrud.create(
+            db_session,
+            UserCreate(username="author", email="author@test.com"),
+        )
+        await PostCrud.create(
+            db_session,
+            PostCreate(title="Post 1", author_id=user.id, is_published=True),
+        )
+
+        # Get user with join on published posts
+        fetched = await UserCrud.get(
+            db_session,
+            filters=[User.id == user.id, Post.is_published == True],  # noqa: E712
+            joins=[(Post, Post.author_id == User.id)],
+        )
+        assert fetched.id == user.id
+
+    @pytest.mark.anyio
+    async def test_first_with_join(self, db_session: AsyncSession):
+        """First with join returns matching record."""
+        user = await UserCrud.create(
+            db_session,
+            UserCreate(username="writer", email="writer@test.com"),
+        )
+        await PostCrud.create(
+            db_session,
+            PostCreate(title="Draft", author_id=user.id, is_published=False),
+        )
+
+        # Find user with unpublished posts
+        result = await UserCrud.first(
+            db_session,
+            filters=[Post.is_published == False],  # noqa: E712
+            joins=[(Post, Post.author_id == User.id)],
+        )
+        assert result is not None
+        assert result.id == user.id
+
+    @pytest.mark.anyio
+    async def test_first_with_outer_join(self, db_session: AsyncSession):
+        """First with outer join includes records without related data."""
+        # User without posts
+        user = await UserCrud.create(
+            db_session,
+            UserCreate(username="no_posts", email="no_posts@test.com"),
+        )
+
+        # With outer join, user should be found even without posts
+        result = await UserCrud.first(
+            db_session,
+            filters=[User.id == user.id],
+            joins=[(Post, Post.author_id == User.id)],
+            outer_join=True,
+        )
+        assert result is not None
+        assert result.id == user.id
+
+    @pytest.mark.anyio
+    async def test_get_multi_with_inner_join(self, db_session: AsyncSession):
+        """Get multiple with inner join only returns matching records."""
+        # User with published post
+        user1 = await UserCrud.create(
+            db_session,
+            UserCreate(username="publisher", email="pub@test.com"),
+        )
+        await PostCrud.create(
+            db_session,
+            PostCreate(title="Published", author_id=user1.id, is_published=True),
+        )
+
+        # User without posts
+        await UserCrud.create(
+            db_session,
+            UserCreate(username="lurker", email="lurk@test.com"),
+        )
+
+        # Inner join should only return user with published post
+        users = await UserCrud.get_multi(
+            db_session,
+            joins=[(Post, Post.author_id == User.id)],
+            filters=[Post.is_published == True],  # noqa: E712
+        )
+        assert len(users) == 1
+        assert users[0].username == "publisher"
+
+    @pytest.mark.anyio
+    async def test_get_multi_with_outer_join(self, db_session: AsyncSession):
+        """Get multiple with outer join includes all records."""
+        # User with post
+        user1 = await UserCrud.create(
+            db_session,
+            UserCreate(username="has_post", email="has@test.com"),
+        )
+        await PostCrud.create(
+            db_session,
+            PostCreate(title="My Post", author_id=user1.id),
+        )
+
+        # User without posts
+        await UserCrud.create(
+            db_session,
+            UserCreate(username="no_post", email="no@test.com"),
+        )
+
+        # Outer join should return both users
+        users = await UserCrud.get_multi(
+            db_session,
+            joins=[(Post, Post.author_id == User.id)],
+            outer_join=True,
+        )
+        assert len(users) == 2
+
+    @pytest.mark.anyio
+    async def test_count_with_join(self, db_session: AsyncSession):
+        """Count with join counts correctly."""
+        # Create users with different post statuses
+        user1 = await UserCrud.create(
+            db_session,
+            UserCreate(username="active_author", email="active@test.com"),
+        )
+        await PostCrud.create(
+            db_session,
+            PostCreate(title="Published 1", author_id=user1.id, is_published=True),
+        )
+
+        user2 = await UserCrud.create(
+            db_session,
+            UserCreate(username="draft_author", email="draft@test.com"),
+        )
+        await PostCrud.create(
+            db_session,
+            PostCreate(title="Draft 1", author_id=user2.id, is_published=False),
+        )
+
+        # Count users with published posts
+        count = await UserCrud.count(
+            db_session,
+            filters=[Post.is_published == True],  # noqa: E712
+            joins=[(Post, Post.author_id == User.id)],
+        )
+        assert count == 1
+
+    @pytest.mark.anyio
+    async def test_exists_with_join(self, db_session: AsyncSession):
+        """Exists with join checks correctly."""
+        user = await UserCrud.create(
+            db_session,
+            UserCreate(username="poster", email="poster@test.com"),
+        )
+        await PostCrud.create(
+            db_session,
+            PostCreate(title="Exists Post", author_id=user.id, is_published=True),
+        )
+
+        # Check if user with published post exists
+        exists = await UserCrud.exists(
+            db_session,
+            filters=[Post.is_published == True],  # noqa: E712
+            joins=[(Post, Post.author_id == User.id)],
+        )
+        assert exists is True
+
+        # Check if user with specific title exists
+        exists = await UserCrud.exists(
+            db_session,
+            filters=[Post.title == "Nonexistent"],
+            joins=[(Post, Post.author_id == User.id)],
+        )
+        assert exists is False
+
+    @pytest.mark.anyio
+    async def test_paginate_with_join(self, db_session: AsyncSession):
+        """Paginate with join works correctly."""
+        # Create users with posts
+        for i in range(5):
+            user = await UserCrud.create(
+                db_session,
+                UserCreate(username=f"author{i}", email=f"author{i}@test.com"),
+            )
+            await PostCrud.create(
+                db_session,
+                PostCreate(
+                    title=f"Post {i}",
+                    author_id=user.id,
+                    is_published=i % 2 == 0,
+                ),
+            )
+
+        # Paginate users with published posts
+        result = await UserCrud.paginate(
+            db_session,
+            joins=[(Post, Post.author_id == User.id)],
+            filters=[Post.is_published == True],  # noqa: E712
+            page=1,
+            items_per_page=10,
+        )
+
+        assert result["pagination"]["total_count"] == 3
+        assert len(result["data"]) == 3
+
+    @pytest.mark.anyio
+    async def test_paginate_with_outer_join(self, db_session: AsyncSession):
+        """Paginate with outer join includes all records."""
+        # User with post
+        user1 = await UserCrud.create(
+            db_session,
+            UserCreate(username="with_post", email="with@test.com"),
+        )
+        await PostCrud.create(
+            db_session,
+            PostCreate(title="A Post", author_id=user1.id),
+        )
+
+        # User without post
+        await UserCrud.create(
+            db_session,
+            UserCreate(username="without_post", email="without@test.com"),
+        )
+
+        # Paginate with outer join
+        result = await UserCrud.paginate(
+            db_session,
+            joins=[(Post, Post.author_id == User.id)],
+            outer_join=True,
+            page=1,
+            items_per_page=10,
+        )
+
+        assert result["pagination"]["total_count"] == 2
+        assert len(result["data"]) == 2
+
+    @pytest.mark.anyio
+    async def test_multiple_joins(self, db_session: AsyncSession):
+        """Multiple joins can be applied."""
+        role = await RoleCrud.create(db_session, RoleCreate(name="author_role"))
+        user = await UserCrud.create(
+            db_session,
+            UserCreate(
+                username="multi_join",
+                email="multi@test.com",
+                role_id=role.id,
+            ),
+        )
+        await PostCrud.create(
+            db_session,
+            PostCreate(title="Multi Join Post", author_id=user.id, is_published=True),
+        )
+
+        # Join both Role and Post
+        users = await UserCrud.get_multi(
+            db_session,
+            joins=[
+                (Role, Role.id == User.role_id),
+                (Post, Post.author_id == User.id),
+            ],
+            filters=[Role.name == "author_role", Post.is_published == True],  # noqa: E712
+        )
+        assert len(users) == 1
+        assert users[0].username == "multi_join"