feat(repository): add pk_attr_names property for primary key attribute access

Author: cofin

advanced_alchemy/repository/_async.py

@@ -94,6 +94,9 @@ class SQLAlchemyAsyncRepositoryProtocol(FilterableRepositoryProtocol[ModelT], Pr
     error_messages: Optional[ErrorMessages] = None
     wrap_exceptions: bool = True
 
+    @property
+    def pk_attr_names(self) -> tuple[str, ...]: ...
+
     @property
     def has_composite_pk(self) -> bool: ...
 
@@ -720,6 +723,15 @@ def set_id_attribute_value(
         setattr(item, id_attribute if id_attribute is not None else cls.id_attribute, item_id)
         return item
 
+    @property
+    def pk_attr_names(self) -> tuple[str, ...]:
+        """Get primary key attribute names.
+
+        Returns:
+            Tuple of ORM attribute names for primary key columns.
+        """
+        return self._pk_attr_names
+
     @property
     def has_composite_pk(self) -> bool:
         """Check if model has a composite (multi-column) primary key.

advanced_alchemy/repository/_sync.py

@@ -95,6 +95,9 @@ class SQLAlchemySyncRepositoryProtocol(FilterableRepositoryProtocol[ModelT], Pro
     error_messages: Optional[ErrorMessages] = None
     wrap_exceptions: bool = True
 
+    @property
+    def pk_attr_names(self) -> tuple[str, ...]: ...
+
     @property
     def has_composite_pk(self) -> bool: ...
 
@@ -721,6 +724,15 @@ def set_id_attribute_value(
         setattr(item, id_attribute if id_attribute is not None else cls.id_attribute, item_id)
         return item
 
+    @property
+    def pk_attr_names(self) -> tuple[str, ...]:
+        """Get primary key attribute names.
+
+        Returns:
+            Tuple of ORM attribute names for primary key columns.
+        """
+        return self._pk_attr_names
+
     @property
     def has_composite_pk(self) -> bool:
         """Check if model has a composite (multi-column) primary key.

advanced_alchemy/repository/memory/_async.py

@@ -136,7 +136,7 @@ def _pk_columns(self) -> tuple[Any, ...]:
         return tuple(mapper.primary_key)
 
     @property
-    def _pk_attr_names(self) -> tuple[str, ...]:
+    def pk_attr_names(self) -> tuple[str, ...]:
         """Get primary key attribute names from the model mapper.
 
         Uses mapper.get_property_by_column() to get ORM attribute names,
@@ -167,7 +167,7 @@ def get_primary_key_value(self, instance: ModelT) -> PrimaryKeyType:
             - For single PK: scalar value
             - For composite PK: tuple of values in column order
         """
-        return extract_pk_value_from_instance(instance, self._pk_attr_names)
+        return extract_pk_value_from_instance(instance, self.pk_attr_names)
 
     def has_primary_key_values(self, instance: ModelT) -> bool:
         """Check if all primary key values are set on an instance.
@@ -178,7 +178,7 @@ def has_primary_key_values(self, instance: ModelT) -> bool:
         Returns:
             True if all PK values are non-None, False otherwise.
         """
-        return pk_values_present(instance, self._pk_attr_names)
+        return pk_values_present(instance, self.pk_attr_names)
 
     def _normalize_pk_to_tuple(self, pk_value: PrimaryKeyType) -> tuple[Any, ...]:
         """Normalize a primary key value to a tuple for consistent storage key generation.
@@ -189,7 +189,7 @@ def _normalize_pk_to_tuple(self, pk_value: PrimaryKeyType) -> tuple[Any, ...]:
         Returns:
             Tuple representation of the primary key.
         """
-        return normalize_pk_to_tuple(pk_value, self._pk_attr_names, self.model_type.__name__)
+        return normalize_pk_to_tuple(pk_value, self.pk_attr_names, self.model_type.__name__)
 
     def _get_store_key(self, pk_value: PrimaryKeyType) -> str:
         """Generate a store key from a primary key value.

advanced_alchemy/repository/memory/_sync.py

@@ -137,7 +137,7 @@ def _pk_columns(self) -> tuple[Any, ...]:
         return tuple(mapper.primary_key)
 
     @property
-    def _pk_attr_names(self) -> tuple[str, ...]:
+    def pk_attr_names(self) -> tuple[str, ...]:
         """Get primary key attribute names from the model mapper.
 
         Uses mapper.get_property_by_column() to get ORM attribute names,
@@ -168,7 +168,7 @@ def get_primary_key_value(self, instance: ModelT) -> PrimaryKeyType:
             - For single PK: scalar value
             - For composite PK: tuple of values in column order
         """
-        return extract_pk_value_from_instance(instance, self._pk_attr_names)
+        return extract_pk_value_from_instance(instance, self.pk_attr_names)
 
     def has_primary_key_values(self, instance: ModelT) -> bool:
         """Check if all primary key values are set on an instance.
@@ -179,7 +179,7 @@ def has_primary_key_values(self, instance: ModelT) -> bool:
         Returns:
             True if all PK values are non-None, False otherwise.
         """
-        return pk_values_present(instance, self._pk_attr_names)
+        return pk_values_present(instance, self.pk_attr_names)
 
     def _normalize_pk_to_tuple(self, pk_value: PrimaryKeyType) -> tuple[Any, ...]:
         """Normalize a primary key value to a tuple for consistent storage key generation.
@@ -190,7 +190,7 @@ def _normalize_pk_to_tuple(self, pk_value: PrimaryKeyType) -> tuple[Any, ...]:
         Returns:
             Tuple representation of the primary key.
         """
-        return normalize_pk_to_tuple(pk_value, self._pk_attr_names, self.model_type.__name__)
+        return normalize_pk_to_tuple(pk_value, self.pk_attr_names, self.model_type.__name__)
 
     def _get_store_key(self, pk_value: PrimaryKeyType) -> str:
         """Generate a store key from a primary key value.

advanced_alchemy/service/_async.py

@@ -4,8 +4,6 @@
 should be a SQLAlchemy model.
 """
 
-# pyright: reportPrivateUsage=false
-
 from collections.abc import AsyncIterator, Iterable, Sequence
 from contextlib import asynccontextmanager
 from functools import cached_property
@@ -826,7 +824,7 @@ async def update(
         elif id_attribute is None and self.repository.has_composite_pk:
             # For composite PKs, check if all PK values are present
             if not self.repository.has_primary_key_values(data):
-                pk_names = ", ".join(self.repository._pk_attr_names)  # pyright: ignore[reportUnknownMemberType]
+                pk_names = ", ".join(self.repository.pk_attr_names)
                 msg = f"Could not identify composite PK values. Required attributes: {pk_names}"
                 raise RepositoryError(msg)
         elif (

advanced_alchemy/service/_sync.py

@@ -6,8 +6,6 @@
 should be a SQLAlchemy model.
 """
 
-# pyright: reportPrivateUsage=false
-
 from collections.abc import Iterable, Iterator, Sequence
 from contextlib import contextmanager
 from functools import cached_property
@@ -825,7 +823,7 @@ def update(
         elif id_attribute is None and self.repository.has_composite_pk:
             # For composite PKs, check if all PK values are present
             if not self.repository.has_primary_key_values(data):
-                pk_names = ", ".join(self.repository._pk_attr_names)  # pyright: ignore[reportUnknownMemberType]
+                pk_names = ", ".join(self.repository.pk_attr_names)
                 msg = f"Could not identify composite PK values. Required attributes: {pk_names}"
                 raise RepositoryError(msg)
         elif (

pyproject.toml

@@ -361,7 +361,7 @@ classmethod-decorators = [
 "advanced_alchemy/alembic/templates/*/env.py" = ["INP001"]
 "advanced_alchemy/repository/*.py" = ['C901', 'UP006', 'UP035']
 "advanced_alchemy/repository/memory/*.py" = ['UP006', 'UP035']
-"advanced_alchemy/service/*.py" = ["PLR0911", "SLF001", "UP006", "UP035"]
+"advanced_alchemy/service/*.py" = ["PLR0911", "UP006", "UP035"]
 "examples/flask.py" = ["ANN"]
 "examples/flask/*.py" = ["ANN"]
 "examples/litestar/*.py" = ["PLR6301", "DOC", "B008"]
@@ -467,10 +467,13 @@ module = [
   "s3fs.*",
   "argon2",
   "argon2.*",
-  "dishka",
-  "dishka.*",
 ]
 
+[[tool.mypy.overrides]]
+follow_imports = "skip"
+ignore_missing_imports = true
+module = ["dishka", "dishka.*"]
+
 [[tool.mypy.overrides]]
 follow_imports = "skip"
 ignore_missing_imports = true
@@ -544,10 +547,7 @@ include = ["advanced_alchemy"]
 pythonVersion = "3.9"
 reportMissingTypeStubs = false
 reportPrivateImportUsage = false
-reportPrivateUsage = false
-reportUnknownArgumentType = false
 reportUnknownMemberType = false
-reportUnknownVariableType = false
 reportUnusedFunction = false
 strict = ["advanced_alchemy/**/*"]
 venv = ".venv"