feat: Add support for composite primary keys in repositories and services, including a fix for MSSQL tuple().in_() syntax.

Author: cofin

advanced_alchemy/repository/_sync.py

@@ -1247,6 +1247,7 @@ def delete_many(
 
                 for idx in range(0, len(normalized_ids), effective_chunk_size):
                     chunk = normalized_ids[idx : min(idx + effective_chunk_size, len(normalized_ids))]
+                    # MSSQL doesn't support tuple().in_() syntax, use OR of AND conditions instead
                     pk_filter = (
                         or_(*[self._build_pk_filter(pk_tuple) for pk_tuple in chunk])
                         if self._dialect.name == "mssql"

docs/usage/repositories.rst

@@ -208,6 +208,71 @@ Delete Where
         return await repository.delete_where(Post.published.is_(False))
 
 
+Composite Primary Keys
+----------------------
+
+Advanced Alchemy supports models with composite primary keys. For these models, the ``PrimaryKeyType``
+accepts several formats when identifying records:
+
+**Tuple Format** - Pass primary key values as a tuple in column definition order:
+
+.. code-block:: python
+
+    from advanced_alchemy.repository import SQLAlchemyAsyncRepository
+    from sqlalchemy import Column, ForeignKey, Integer, String
+    from sqlalchemy.orm import DeclarativeBase
+
+    class Base(DeclarativeBase):
+        pass
+
+    class UserRole(Base):
+        """Model with composite primary key."""
+        __tablename__ = "user_role"
+        user_id: int = Column(Integer, ForeignKey("user.id"), primary_key=True)
+        role_id: int = Column(Integer, ForeignKey("role.id"), primary_key=True)
+        permissions: str = Column(String, nullable=True)
+
+    class UserRoleRepository(SQLAlchemyAsyncRepository[UserRole]):
+        model_type = UserRole
+
+    async def get_user_role(db_session, user_id: int, role_id: int) -> UserRole:
+        repository = UserRoleRepository(session=db_session)
+        # Tuple format: values in column definition order
+        return await repository.get((user_id, role_id))
+
+**Dict Format** - Pass primary key values as a dictionary with column names as keys:
+
+.. code-block:: python
+
+    async def get_user_role_by_dict(db_session, user_id: int, role_id: int) -> UserRole:
+        repository = UserRoleRepository(session=db_session)
+        # Dict format: explicit column names
+        return await repository.get({"user_id": user_id, "role_id": role_id})
+
+**Bulk Operations with Composite Keys** - Use sequences of tuples or dicts:
+
+.. code-block:: python
+
+    async def delete_user_roles(db_session, role_assignments: list[tuple[int, int]]) -> None:
+        repository = UserRoleRepository(session=db_session)
+        # Delete multiple records by their composite keys
+        await repository.delete_many(role_assignments)
+
+    async def get_multiple_user_roles(db_session) -> list[UserRole]:
+        repository = UserRoleRepository(session=db_session)
+        # Get multiple records using dict format
+        return await repository.get_many([
+            {"user_id": 1, "role_id": 5},
+            {"user_id": 1, "role_id": 6},
+            {"user_id": 2, "role_id": 5},
+        ])
+
+.. note::
+
+    When using tuple format, ensure values are in the same order as the primary key columns
+    are defined on the model. Dict format is more explicit and avoids ordering issues.
+
+
 Transaction Management
 ----------------------
 

docs/usage/services.rst

@@ -92,6 +92,29 @@ Services provide high-level methods for common operations:
         )
         return post_service.to_schema(post, schema_type=PostResponse)
 
+Composite Primary Keys
+~~~~~~~~~~~~~~~~~~~~~~
+
+Services fully support models with composite primary keys using the same formats as repositories.
+Pass primary key values as tuples or dictionaries when using ``get``, ``update``, or ``delete`` methods:
+
+.. code-block:: python
+
+    # Get by composite key (tuple format)
+    user_role = await user_role_service.get((user_id, role_id))
+
+    # Update by composite key (dict format)
+    updated = await user_role_service.update(
+        data={"permissions": "admin"},
+        item_id={"user_id": 1, "role_id": 5},
+    )
+
+    # Delete multiple by composite keys
+    await user_role_service.delete_many([(1, 5), (1, 6), (2, 5)])
+
+See :ref:`Composite Primary Keys <repositories:Composite Primary Keys>` in the Repositories documentation
+for more details on supported formats.
+
 Complex Operations
 -------------------
 

pyproject.toml

@@ -471,6 +471,11 @@ follow_imports = "skip"
 ignore_missing_imports = true
 module = ["pytest", "pytest.*", "_pytest", "_pytest.*"]
 
+[[tool.mypy.overrides]]
+follow_imports = "skip"
+ignore_missing_imports = true
+module = ["sphinx", "sphinx.*"]
+
 [[tool.mypy.overrides]]
 module = "advanced_alchemy._serialization"
 warn_unused_ignores = false