"""Integration tests for the SQLAlchemy Repository implementation using session-based fixtures.""" import asyncio import datetime from collections.abc import Generator from typing import TYPE_CHECKING, Any, Literal, Optional, Union from uuid import UUID import pytest from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import Session from time_machine import travel from advanced_alchemy.exceptions import NotFoundError from advanced_alchemy.filters import ( BeforeAfter, OrderBy, SearchFilter, ) from advanced_alchemy.repository import SQLAlchemyAsyncRepository from advanced_alchemy.repository.memory import ( SQLAlchemyAsyncMockRepository, SQLAlchemySyncMockRepository, ) from advanced_alchemy.service import SQLAlchemyAsyncRepositoryService from tests.helpers import maybe_async # Python 3.9 compatibility for typing.TypeAlias try: # Python >= 3.10 from typing import TypeAlias # type: ignore[attr-defined] except Exception: # Python 3.9 fallback from typing_extensions import TypeAlias # type: ignore[assignment] if TYPE_CHECKING: from time_machine import Coordinates pytestmark = [ pytest.mark.integration, pytest.mark.xdist_group("repository"), ] xfail = pytest.mark.xfail # Type aliases for data and repository/service components RawRecordData: TypeAlias = "list[dict[str, Any]]" RepositoryPKType = Literal["uuid", "bigint"] AnyRepository: TypeAlias = "Union[SQLAlchemyAsyncRepository[Any], SQLAlchemyAsyncMockRepository[Any]]" AnyService: TypeAlias = SQLAlchemyAsyncRepositoryService[Any, "AnyRepository"] # pyright: ignore mock_engines = {"mock_async_engine", "mock_sync_engine"} # Helper functions for repository creation def create_repository( session: "Union[Session, AsyncSession]", model_type: type, repository_class: "Optional[type]" = None ) -> "Any": """Create a repository instance for the given session and model type.""" if repository_class is None: if isinstance(session, AsyncSession): base_repository_class = SQLAlchemyAsyncRepository # type: ignore[assignment] else: from advanced_alchemy.repository import SQLAlchemySyncRepository base_repository_class = SQLAlchemySyncRepository # type: ignore[assignment] else: base_repository_class = repository_class # type: ignore[assignment] # Create a dynamic repository class with the model_type as a class attribute repository_class_name = f"DynamicRepository_{model_type.__name__}" # Add a create method that handles dict data and maps to add for test compatibility async def create(self: Any, data: Any, **kwargs: Any) -> Any: # If data is a dict, convert it to a model instance if isinstance(data, dict): model_instance = model_type(**data) else: model_instance = data return await self.add(model_instance, **kwargs) # Add a create_many method that handles list of dict data async def create_many(self: Any, data: "list[Any]", **kwargs: Any) -> "list[Any]": # Convert dict items to model instances model_instances = [] for item in data: if isinstance(item, dict): model_instances.append(model_type(**item)) else: model_instances.append(item) return await self.add_many(model_instances, **kwargs) # type: ignore[no-any-return] def create_sync(self: Any, data: Any, **kwargs: Any) -> Any: # Sync version for sync repositories if isinstance(data, dict): model_instance = model_type(**data) else: model_instance = data return self.add(model_instance, **kwargs) # Sync version of create_many def create_many_sync(self: Any, data: "list[Any]", **kwargs: Any) -> "list[Any]": # Convert dict items to model instances model_instances = [] for item in data: if isinstance(item, dict): model_instances.append(model_type(**item)) else: model_instances.append(item) return self.add_many(model_instances, **kwargs) # type: ignore[no-any-return] # Choose the right create methods based on repository type create_method = create if isinstance(session, AsyncSession) else create_sync create_many_method = create_many if isinstance(session, AsyncSession) else create_many_sync DynamicRepository = type( repository_class_name, (base_repository_class,), {"model_type": model_type, "create": create_method, "create_many": create_many_method}, ) return DynamicRepository(session=session) def create_service( session: "Union[Session, AsyncSession]", model_type: type, service_class: "Optional[type]" = None ) -> Any: """Create a service instance for the given session and model type.""" # Create a repository first, since services operate on repositories repository = create_repository(session, model_type) if service_class is None: if isinstance(session, AsyncSession): from advanced_alchemy.service import SQLAlchemyAsyncRepositoryService base_service_class = SQLAlchemyAsyncRepositoryService else: from advanced_alchemy.service import SQLAlchemySyncRepositoryService base_service_class = SQLAlchemySyncRepositoryService # type: ignore[assignment] else: base_service_class = service_class # type: ignore[assignment] # Create a dynamic service class that knows about the repository service_class_name = f"DynamicService_{model_type.__name__}" # Set the repository_type to the same type as our dynamic repository repository_type = type(repository) DynamicService = type(service_class_name, (base_service_class,), {"repository_type": repository_type}) # Initialize the service return DynamicService(session=session) # Helper functions for session-based testing def get_model_from_session( session_data: "tuple[Union[Session, AsyncSession], dict[str, type]]", model_name: str ) -> type: """Extract a model type from session data tuple.""" _, models = session_data return models[model_name] def get_repository_from_session( session_data: "tuple[Union[Session, AsyncSession], dict[str, type]]", model_name: str ) -> Any: """Create a repository from session data tuple.""" session, models = session_data model_type = models[model_name] return create_repository(session, model_type) def get_service_from_session( session_data: "tuple[Union[Session, AsyncSession], dict[str, type]]", model_name: str ) -> Any: """Create a service from session data tuple.""" session, models = session_data model_type = models[model_name] return create_service(session, model_type) @pytest.fixture(autouse=True) def _clear_in_memory_db() -> "Generator[None, None, None]": # pyright: ignore[reportUnusedFunction] try: yield finally: SQLAlchemyAsyncMockRepository.__database_clear__() SQLAlchemySyncMockRepository.__database_clear__() @pytest.fixture() def frozen_datetime() -> "Generator[Coordinates, None, None]": with travel(lambda: datetime.datetime.now(datetime.timezone.utc), tick=False) as frozen: yield frozen # Test functions using new session-based pattern async def test_repo_count_method(seeded_test_session_async: "tuple[AsyncSession, dict[str, type]]") -> None: """Test SQLAlchemy count.""" session, models = seeded_test_session_async author_repo = create_repository(session, models["author"]) assert await maybe_async(author_repo.count()) == 2 async def test_repo_count_method_with_filters( seeded_test_session_async: "tuple[AsyncSession, dict[str, type]]", ) -> None: """Test SQLAlchemy count with filters.""" session, models = seeded_test_session_async author_repo = create_repository(session, models["author"]) # Get the first author name from seeded data if hasattr(session, "bind") and getattr(session.bind, "dialect", {}).name == "mock": # Mock repository handling assert ( await maybe_async( author_repo.count( **{author_repo.model_type.name.key: "Agatha Christie"}, ), ) == 1 ) else: # Real repository handling assert ( await maybe_async( author_repo.count( author_repo.model_type.name == "Agatha Christie", ), ) == 1 ) async def test_repo_list_and_count_method(seeded_test_session_async: "tuple[AsyncSession, dict[str, type]]") -> None: """Test SQLAlchemy list with count in asyncpg.""" session, models = seeded_test_session_async author_repo = create_repository(session, models["author"]) data, count = await maybe_async(author_repo.list_and_count()) assert len(data) == 2 assert count == 2 async def test_repo_list_and_count_basic_method( seeded_test_session_async: "tuple[AsyncSession, dict[str, type]]", ) -> None: """Test SQLAlchemy list and count.""" session, models = seeded_test_session_async author_repo = create_repository(session, models["author"]) data, count = await maybe_async(author_repo.list_and_count()) assert len(data) == 2 assert count == 2 async def test_repo_list_method_with_filters(seeded_test_session_async: "tuple[AsyncSession, dict[str, type]]") -> None: """Test SQLAlchemy list with filters.""" session, models = seeded_test_session_async author_repo = create_repository(session, models["author"]) # Test filtering by name if hasattr(session, "bind") and getattr(session.bind, "dialect", {}).name == "mock": # Mock repository handling data = await maybe_async(author_repo.list(**{author_repo.model_type.name.key: "Agatha Christie"})) else: # Real repository handling data = await maybe_async(author_repo.list(author_repo.model_type.name == "Agatha Christie")) assert len(data) == 1 assert data[0].name == "Agatha Christie" async def test_repo_exists_method(seeded_test_session_async: "tuple[AsyncSession, dict[str, type]]") -> None: """Test repository exists method.""" session, models = seeded_test_session_async author_repo = create_repository(session, models["author"]) # Get first author ID authors = await maybe_async(author_repo.list()) first_author_id = authors[0].id assert await maybe_async(author_repo.exists(id=first_author_id)) is True # Test with non-existent ID non_existent_id = UUID("00000000-0000-0000-0000-000000000000") if hasattr(first_author_id, "hex") else 99999 assert await maybe_async(author_repo.exists(id=non_existent_id)) is False async def test_repo_get_method(seeded_test_session_async: "tuple[AsyncSession, dict[str, type]]") -> None: """Test repository get method.""" session, models = seeded_test_session_async author_repo = create_repository(session, models["author"]) # Get first author ID authors = await maybe_async(author_repo.list()) first_author_id = authors[0].id author = await maybe_async(author_repo.get(first_author_id)) assert author.id == first_author_id async def test_repo_get_one_or_none_method(seeded_test_session_async: "tuple[AsyncSession, dict[str, type]]") -> None: """Test repository get_one_or_none method.""" session, models = seeded_test_session_async author_repo = create_repository(session, models["author"]) # Get first author ID authors = await maybe_async(author_repo.list()) first_author_id = authors[0].id author = await maybe_async(author_repo.get_one_or_none(id=first_author_id)) assert author is not None assert author.id == first_author_id # Test with non-existent ID non_existent_id = UUID("00000000-0000-0000-0000-000000000000") if hasattr(first_author_id, "hex") else 99999 author = await maybe_async(author_repo.get_one_or_none(id=non_existent_id)) assert author is None async def test_repo_create_method(seeded_test_session_async: "tuple[AsyncSession, dict[str, type]]") -> None: """Test repository create method.""" session, models = seeded_test_session_async author_repo = create_repository(session, models["author"]) new_author_data = {"name": "Test Author", "dob": datetime.datetime.now(datetime.timezone.utc).date()} new_author = await maybe_async(author_repo.create(new_author_data)) assert new_author.name == "Test Author" assert new_author.id is not None # Verify it was actually created total_count = await maybe_async(author_repo.count()) assert total_count == 3 async def test_repo_update_method(seeded_test_session_async: "tuple[AsyncSession, dict[str, type]]") -> None: """Test repository update method.""" session, models = seeded_test_session_async author_repo = create_repository(session, models["author"]) # Get first author authors = await maybe_async(author_repo.list()) author = authors[0] original_name = author.name original_created_at = author.created_at original_updated_at = author.updated_at # Update the author author.name = "Updated Name" updated_author = await maybe_async(author_repo.update(author)) assert updated_author.name == "Updated Name" assert updated_author.name != original_name assert updated_author.created_at == original_created_at assert updated_author.updated_at > original_updated_at async def test_service_update_with_dict_data_refreshes_timestamp( seeded_test_session_async: "tuple[AsyncSession, dict[str, type]]", frozen_datetime: "Coordinates", ) -> None: """Service update should refresh audit timestamps when supplied with dict payloads.""" session, models = seeded_test_session_async author_service = get_service_from_session((session, models), "author") authors = await maybe_async(author_service.list()) author = authors[0] original_created_at = author.created_at original_updated_at = author.updated_at frozen_datetime.shift(datetime.timedelta(seconds=5)) update_payload = {"name": "Dict Driven Update"} updated_author = await maybe_async(author_service.update(update_payload, item_id=author.id)) assert updated_author.name == "Dict Driven Update" assert updated_author.created_at == original_created_at assert updated_author.updated_at > original_updated_at async def test_repo_update_many_method_stale_data_fix( seeded_test_session_async: "tuple[AsyncSession, dict[str, type]]", ) -> None: """Test repository update_many returns refreshed data from database (Issue #1).""" session, models = seeded_test_session_async author_repo = create_repository(session, models["author"]) # Get first two authors authors = await maybe_async(author_repo.list()) authors = authors[:2] # Create update data that only updates name, leaving other fields unchanged update_data = [{"id": authors[0].id, "name": "Updated Author 1"}, {"id": authors[1].id, "name": "Updated Author 2"}] # Store original created_at/updated_at for comparison original_created_at = authors[0].created_at original_updated_at = authors[0].updated_at # Update using update_many updated_authors = await maybe_async(author_repo.update_many(update_data)) # Critical test: returned objects should have ALL attributes populated from database # This was the bug - returned objects had None for non-updated fields assert len(updated_authors) == 2 for updated_author in updated_authors: # These should be updated assert updated_author.name in ["Updated Author 1", "Updated Author 2"] # These should still be populated from database (not None) assert updated_author.created_at is not None assert updated_author.updated_at is not None assert updated_author.id is not None # updated_at should be newer than before if updated_author.id == authors[0].id: assert updated_author.created_at == original_created_at assert updated_author.updated_at > original_updated_at async def test_repo_update_many_mixed_types(seeded_test_session_async: "tuple[AsyncSession, dict[str, type]]") -> None: """Test repository update_many with mixed input types (dicts and model instances).""" session, models = seeded_test_session_async author_repo = create_repository(session, models["author"]) # Get authors to update authors = await maybe_async(author_repo.list()) authors = authors[:2] # Test mixed input types: dict and model instance authors[1].name = "Updated via Model Instance" update_data = [ {"id": authors[0].id, "name": "Updated via Dict"}, # Dict authors[1], # Model instance ] # This should not raise AttributeError (Issue #3) updated_authors = await maybe_async(author_repo.update_many(update_data)) assert len(updated_authors) == 2 updated_names = {author.name for author in updated_authors} assert "Updated via Dict" in updated_names assert "Updated via Model Instance" in updated_names async def test_repo_delete_method(seeded_test_session_async: "tuple[AsyncSession, dict[str, type]]") -> None: """Test repository delete method.""" session, models = seeded_test_session_async author_repo = create_repository(session, models["author"]) # Get first author authors = await maybe_async(author_repo.list()) author = authors[0] author_id = author.id # Delete the author deleted_author = await maybe_async(author_repo.delete(author_id)) assert deleted_author.id == author_id # Verify it was deleted remaining_authors = await maybe_async(author_repo.list()) assert len(remaining_authors) == 1 remaining_ids = [a.id for a in remaining_authors] assert author_id not in remaining_ids async def test_repo_health_check(seeded_test_session_async: "tuple[AsyncSession, dict[str, type]]") -> None: """Test repository health check.""" session, models = seeded_test_session_async author_repo = create_repository(session, models["author"]) # Health check should not raise an exception - it's a class method that needs session assert await maybe_async(author_repo.check_health(session)) is True # Service tests using new session-based pattern async def test_service_count_method(seeded_test_session_async: "tuple[AsyncSession, dict[str, type]]") -> None: """Test service count method.""" _session, _models = seeded_test_session_async author_service = get_service_from_session(seeded_test_session_async, "author") count = await maybe_async(author_service.count()) assert count == 2 async def test_service_list_method(seeded_test_session_async: "tuple[AsyncSession, dict[str, type]]") -> None: """Test service list method.""" _session, _models = seeded_test_session_async author_service = get_service_from_session(seeded_test_session_async, "author") authors = await maybe_async(author_service.list()) assert len(authors) == 2 async def test_service_get_method(seeded_test_session_async: "tuple[AsyncSession, dict[str, type]]") -> None: """Test service get method.""" _session, _models = seeded_test_session_async author_service = get_service_from_session(seeded_test_session_async, "author") # Get first author ID authors = await maybe_async(author_service.list()) first_author_id = authors[0].id author = await maybe_async(author_service.get(first_author_id)) assert author.id == first_author_id async def test_service_create_method(seeded_test_session_async: "tuple[AsyncSession, dict[str, type]]") -> None: """Test service create method.""" _session, _models = seeded_test_session_async author_service = get_service_from_session(seeded_test_session_async, "author") new_author_data = {"name": "Service Test Author", "dob": datetime.datetime.now(datetime.timezone.utc).date()} new_author = await maybe_async(author_service.create(new_author_data)) assert new_author.name == "Service Test Author" assert new_author.id is not None async def test_service_update_method(seeded_test_session_async: "tuple[AsyncSession, dict[str, type]]") -> None: """Test service update method.""" _session, _models = seeded_test_session_async author_service = get_service_from_session(seeded_test_session_async, "author") # Get first author authors = await maybe_async(author_service.list()) author = authors[0] author_id = author.id original_name = author.name original_created_at = author.created_at original_updated_at = author.updated_at # Update via service - correct parameter order is (data, item_id) author.name = "Service Updated Name" updated_author = await maybe_async(author_service.update(author, item_id=author_id)) assert updated_author.name == "Service Updated Name" assert updated_author.name != original_name assert updated_author.created_at == original_created_at assert updated_author.updated_at > original_updated_at async def test_service_delete_method(seeded_test_session_async: "tuple[AsyncSession, dict[str, type]]") -> None: """Test service delete method.""" _session, _models = seeded_test_session_async author_service = get_service_from_session(seeded_test_session_async, "author") # Get first author authors = await maybe_async(author_service.list()) author_id = authors[0].id # Delete via service deleted_author = await maybe_async(author_service.delete(author_id)) assert deleted_author.id == author_id # Verify deletion remaining_authors = await maybe_async(author_service.list()) assert len(remaining_authors) == 1 # Additional filter tests async def test_repo_filter_before_after(seeded_test_session_async: "tuple[AsyncSession, dict[str, type]]") -> None: """Test repository with BeforeAfter filter.""" session, models = seeded_test_session_async author_repo = create_repository(session, models["author"]) # Test date filtering cutoff_date = datetime.datetime(2023, 4, 1, tzinfo=datetime.timezone.utc) filter_obj = BeforeAfter(field_name="created_at", before=cutoff_date, after=None) authors = await maybe_async(author_repo.list(filter_obj)) # Should get authors created before April 1, 2023 assert len(authors) >= 1 # At least one author should match async def test_repo_filter_search(seeded_test_session_async: "tuple[AsyncSession, dict[str, type]]") -> None: """Test repository with SearchFilter.""" session, models = seeded_test_session_async author_repo = create_repository(session, models["author"]) # Search for 'Christie' in name search_filter = SearchFilter(field_name="name", value="Christie") authors = await maybe_async(author_repo.list(search_filter)) assert len(authors) == 1 assert "Christie" in authors[0].name async def test_repo_filter_order_by(seeded_test_session_async: "tuple[AsyncSession, dict[str, type]]") -> None: """Test repository with OrderBy filter.""" session, models = seeded_test_session_async author_repo = create_repository(session, models["author"]) # Order by name ascending order_filter = OrderBy(field_name="name") authors = await maybe_async(author_repo.list(order_filter)) assert len(authors) == 2 # Verify ordering names = [author.name for author in authors] assert names == sorted(names) # Pagination tests async def test_service_paginated_list(seeded_test_session_async: "tuple[AsyncSession, dict[str, type]]") -> None: """Test service paginated list.""" from advanced_alchemy.filters import LimitOffset _session, _models = seeded_test_session_async author_service = get_service_from_session(seeded_test_session_async, "author") # Test pagination using LimitOffset filter with consistent ordering paginated = await maybe_async(author_service.list(LimitOffset(limit=1, offset=0), OrderBy(field_name="name"))) assert len(paginated) == 1 # Error handling tests async def test_repo_error_messages(seeded_test_session_async: "tuple[AsyncSession, dict[str, type]]") -> None: """Test repository error handling.""" session, models = seeded_test_session_async author_repo = create_repository(session, models["author"]) # Test NotFoundError for non-existent ID non_existent_id = ( UUID("00000000-0000-0000-0000-000000000000") if hasattr(author_repo.model_type.id.type, "python_type") and author_repo.model_type.id.type.python_type == UUID else 99999 ) with pytest.raises(NotFoundError): await maybe_async(author_repo.get(non_existent_id)) # Comprehensive tests for GitHub issue #535 and bug_fix.md issues async def test_service_pydantic_partial_update_github_535( seeded_test_session_async: "tuple[AsyncSession, dict[str, type]]", ) -> None: """Test service update with Pydantic models using exclude_unset for partial updates (GitHub Issue #535).""" pydantic = pytest.importorskip("pydantic") _session, _models = seeded_test_session_async author_service = get_service_from_session(seeded_test_session_async, "author") # Create an author author = await maybe_async(author_service.create({"name": "Original Name", "dob": datetime.date(1990, 1, 1)})) original_dob = author.dob # Create a Pydantic model for partial update with optional fields class AuthorUpdateSchema(pydantic.BaseModel): # type: ignore[name-defined,misc] name: "Optional[str]" = None dob: "Optional[datetime.date]" = None # Partial update with only name field set (dob is unset) partial_update = AuthorUpdateSchema(name="Updated Name") assert partial_update.name == "Updated Name" assert "dob" not in partial_update.model_fields_set # Update via service - should only update name, leave dob unchanged updated_author = await maybe_async(author_service.update(partial_update, item_id=author.id)) # Verify: name was updated, but dob remains unchanged assert updated_author.name == "Updated Name" assert updated_author.dob == original_dob # Should be unchanged assert updated_author.id == author.id async def test_service_msgspec_partial_update_github_535( seeded_test_session_async: "tuple[AsyncSession, dict[str, type]]", ) -> None: """Test service update with msgspec structs using UNSET for partial updates (GitHub Issue #535).""" msgspec = pytest.importorskip("msgspec") _session, _models = seeded_test_session_async author_service = get_service_from_session(seeded_test_session_async, "author") # Create an author author = await maybe_async(author_service.create({"name": "Original Name", "dob": datetime.date(1990, 1, 1)})) original_dob = author.dob # Create a msgspec struct for partial update with UNSET field class AuthorUpdateSchema(msgspec.Struct): # type: ignore[name-defined,misc] name: "str" = msgspec.UNSET dob: "datetime.date" = msgspec.UNSET # Partial update with only name field set (dob is UNSET) partial_update = AuthorUpdateSchema(name="Updated Name") assert partial_update.name == "Updated Name" assert partial_update.dob is msgspec.UNSET # Update via service - should only update name, leave dob unchanged updated_author = await maybe_async(author_service.update(partial_update, item_id=author.id)) # Verify: name was updated, but dob remains unchanged assert updated_author.name == "Updated Name" assert updated_author.dob == original_dob # Should be unchanged assert updated_author.id == author.id async def test_service_update_many_schema_types_github_535( seeded_test_session_async: "tuple[AsyncSession, dict[str, type]]", ) -> None: """Test service update_many with different schema types (GitHub Issue #535).""" pydantic = pytest.importorskip("pydantic") msgspec = pytest.importorskip("msgspec") _session, _models = seeded_test_session_async author_service = get_service_from_session(seeded_test_session_async, "author") # Create multiple authors author1 = await maybe_async(author_service.create({"name": "Author One", "dob": datetime.date(1990, 1, 1)})) author2 = await maybe_async(author_service.create({"name": "Author Two", "dob": datetime.date(1991, 2, 2)})) original_dob1 = author1.dob original_dob2 = author2.dob original_created_at1 = author1.created_at original_created_at2 = author2.created_at original_updated_at1 = author1.updated_at original_updated_at2 = author2.updated_at # Get ID type from model for dynamic schema creation # For Pydantic compatibility, we need to map database-specific types to Python types from uuid import UUID as PythonUUID actual_id_type = type(author1.id) if hasattr(actual_id_type, "__name__") and "UUID" in actual_id_type.__name__: id_type = PythonUUID # Use standard UUID for database UUID types else: id_type = int # type: ignore[assignment] class AuthorUpdateSchema(pydantic.BaseModel): # type: ignore[name-defined,misc] id: id_type # type: ignore[valid-type] name: "Optional[str]" = None dob: "Optional[datetime.date]" = None # Create msgspec schema for partial updates class AuthorUpdateMsgspecSchema(msgspec.Struct): # type: ignore[name-defined,misc] id: id_type # type: ignore[valid-type] name: "str" = msgspec.UNSET dob: "datetime.date" = msgspec.UNSET # Test update_many with mixed schema types (Pydantic, msgspec, dict) update_data = [ AuthorUpdateSchema(id=author1.id, name="Updated Author One"), # Pydantic with UNSET dob AuthorUpdateMsgspecSchema(id=author2.id, name="Updated Author Two"), # msgspec with UNSET dob ] # Sleep to ensure timestamp difference for databases with lower precision await asyncio.sleep(1.1) # Update via service - should only update names, leave dobs unchanged updated_authors = await maybe_async(author_service.update_many(update_data)) # Verify updates assert len(updated_authors) == 2 # Find updated authors by ID updated_author1 = next(a for a in updated_authors if a.id == author1.id) updated_author2 = next(a for a in updated_authors if a.id == author2.id) # Verify: names were updated, but dobs remain unchanged assert updated_author1.name == "Updated Author One" assert updated_author1.dob == original_dob1 # Should be unchanged assert updated_author1.created_at == original_created_at1 assert updated_author1.updated_at > original_updated_at1 assert updated_author2.name == "Updated Author Two" assert updated_author2.dob == original_dob2 # Should be unchanged assert updated_author2.created_at == original_created_at2 assert updated_author2.updated_at > original_updated_at2 async def test_repo_update_many_non_returning_backend_refresh( seeded_test_session_async: "tuple[AsyncSession, dict[str, type]]", ) -> None: """Test repository update_many refreshes data on non-RETURNING backends (Issue #1 from bug_fix.md).""" session, models = seeded_test_session_async author_repo = create_repository(session, models["author"]) # Create multiple authors authors = await maybe_async( author_repo.create_many( [ {"name": "Author A", "dob": datetime.date(1990, 1, 1)}, {"name": "Author B", "dob": datetime.date(1991, 2, 2)}, ] ) ) # Prepare update data with partial changes update_data = [ {"id": authors[0].id, "name": "Updated Author A"}, # Only updating name {"id": authors[1].id, "name": "Updated Author B"}, # Only updating name ] # Update via repository updated_authors = await maybe_async(author_repo.update_many(update_data)) # Verify returned objects have ALL attributes populated (not just updated ones) assert len(updated_authors) == 2 for updated_author in updated_authors: # Verify all attributes are populated from database, not just updated ones assert updated_author.name is not None and "Updated" in updated_author.name assert updated_author.dob is not None # Should be populated from database, not None assert updated_author.id is not None assert updated_author.created_at is not None # Audit fields should be populated assert updated_author.updated_at is not None async def test_service_mixed_input_types_update_many( seeded_test_session_async: "tuple[AsyncSession, dict[str, type]]", ) -> None: """Test service update_many with mixed input types (dicts, Pydantic models, msgspec structs).""" pydantic = pytest.importorskip("pydantic") msgspec = pytest.importorskip("msgspec") _session, _models = seeded_test_session_async author_service = get_service_from_session(seeded_test_session_async, "author") # Create multiple authors authors = await maybe_async( author_service.create_many( [ {"name": "Author 1", "dob": datetime.date(1990, 1, 1)}, {"name": "Author 2", "dob": datetime.date(1991, 2, 2)}, {"name": "Author 3", "dob": datetime.date(1992, 3, 3)}, ] ) ) # Get ID type from model for dynamic schema creation # For Pydantic compatibility, we need to map database-specific types to Python types from uuid import UUID as PythonUUID actual_id_type = type(authors[0].id) if hasattr(actual_id_type, "__name__") and "UUID" in actual_id_type.__name__: id_type = PythonUUID # Use standard UUID for database UUID types else: id_type = int # type: ignore[assignment] # Create schema classes class AuthorUpdatePydantic(pydantic.BaseModel): # type: ignore[name-defined,misc] id: id_type # type: ignore[valid-type] name: "Optional[str]" = None dob: "Optional[datetime.date]" = None class AuthorUpdateMsgspec(msgspec.Struct): # type: ignore[name-defined,misc] id: id_type # type: ignore[valid-type] name: "str" = msgspec.UNSET dob: "datetime.date" = msgspec.UNSET # Test with mixed input types mixed_update_data = [ {"id": authors[0].id, "name": "Dict Updated"}, # Dictionary AuthorUpdatePydantic(id=authors[1].id, name="Pydantic Updated"), # Pydantic model AuthorUpdateMsgspec(id=authors[2].id, name="Msgspec Updated"), # msgspec struct ] # Update via service updated_authors = await maybe_async(author_service.update_many(mixed_update_data)) # Verify all updates worked correctly assert len(updated_authors) == 3 # Create a mapping by ID for verification updated_by_id = {author.id: author for author in updated_authors} # Verify each update matches the expected result assert updated_by_id[authors[0].id].name == "Dict Updated" assert updated_by_id[authors[1].id].name == "Pydantic Updated" assert updated_by_id[authors[2].id].name == "Msgspec Updated" # Verify all attributes are properly populated (not stale/None) for author in updated_authors: assert author.dob is not None assert author.created_at is not None assert author.updated_at is not None async def test_repo_update_with_model_instance_partial_fields_github_560( seeded_test_session_async: "tuple[AsyncSession, dict[str, type]]", ) -> None: """Test repository update with model instances for partial updates (GitHub Issue #560). This test verifies that when updating with a model instance where only some fields are explicitly set, the unset fields do not overwrite existing data with None. """ _session, models = seeded_test_session_async author_model = models["author"] author_repo = get_repository_from_session(seeded_test_session_async, "author") # Create an author with all fields populated author = await maybe_async(author_repo.create({"name": "Original Name", "dob": datetime.date(1990, 1, 1)})) original_dob = author.dob original_id = author.id # Create a partial update using a model instance with only id and name set # This mimics the pattern: Author(id=1, name="Updated Name") # SQLAlchemy initializes 'dob' to None, but it wasn't explicitly set by the user partial_update = author_model(id=original_id, name="Updated Name") # Update via repository - should only update name, leave dob unchanged updated_author = await maybe_async(author_repo.update(partial_update)) # Verify: name was updated, but dob remains unchanged (not overwritten with None) assert updated_author.name == "Updated Name" assert updated_author.dob == original_dob # Should be unchanged assert updated_author.id == original_id async def test_repo_update_many_with_model_instances_partial_fields_github_560( seeded_test_session_async: "tuple[AsyncSession, dict[str, type]]", ) -> None: """Test repository update_many with model instances for partial updates (GitHub Issue #560). This test verifies that update_many correctly handles model instances with partially set fields, preventing None values from overwriting existing data. """ _session, models = seeded_test_session_async author_model = models["author"] author_repo = get_repository_from_session(seeded_test_session_async, "author") # Create multiple authors with all fields populated author1 = await maybe_async(author_repo.create({"name": "Author One", "dob": datetime.date(1990, 1, 1)})) author2 = await maybe_async(author_repo.create({"name": "Author Two", "dob": datetime.date(1991, 2, 2)})) original_dob1 = author1.dob original_dob2 = author2.dob # Create partial updates using model instances with only id and name set partial_updates = [ author_model(id=author1.id, name="Updated One"), author_model(id=author2.id, name="Updated Two"), ] # Update via repository - should only update names, leave dobs unchanged updated_authors = await maybe_async(author_repo.update_many(partial_updates)) # Verify: names were updated, but dobs remain unchanged assert len(updated_authors) == 2 updated_by_id = {author.id: author for author in updated_authors} assert updated_by_id[author1.id].name == "Updated One" assert updated_by_id[author1.id].dob == original_dob1 # Should be unchanged assert updated_by_id[author2.id].name == "Updated Two" assert updated_by_id[author2.id].dob == original_dob2 # Should be unchanged