feat: add get_item method to BaseRepository with tests, examples and docs

AGENTS.md

@@ -78,6 +78,16 @@ class ProductRepository(BaseRepository, table=ProductTable, filter_=ProductFilte
 
 Note: `pydantic-filters` from GitHub is required for filter support (PyPI version is broken with pydantic v2). The project currently uses a fork with Python 3.14 lazy-annotations support: `git+https://github.com/OlegYurchik/pydantic-filters.git@fix/compare-to-pydantic-2.12`.
 
+#### Single item retrieval
+
+`get_item(filter_=..., sort=...)` returns the first matching record (or `None` if no records match). It delegates to `get_items` under the hood, so filter and sort semantics are identical:
+
+```python
+user = await user_repository.get_item(filter_=UserFilter(email="alice@example.com"))
+if user is not None:
+    print(user.name)
+```
+
 #### Eager loading (options)
 
 `get_items()` and `update_items()` accept an optional `options` parameter for SQLAlchemy eager loading strategies such as `joinedload` or `selectinload`:

examples/basic_usage.py

@@ -32,6 +32,16 @@ async def main() -> None:
     )
     print(f"Created: {user.name}, {user.email}")
 
+    # Read single item
+    single = await repository.get_item()
+    print(f"Single item: {single.name}, {single.email}")
+
+    # Read by filter
+    filtered = await repository.get_item(
+        filter_=UserFilter(email="alice@example.com"),
+    )
+    print(f"Filtered item: {filtered.name if filtered else None}")
+
     # Read all
     items = [item async for item in repository.get_items()]
     print(f"All items: {[(item.name, item.email) for item in items]}")

examples/container_usage.py

@@ -78,6 +78,10 @@ async def main() -> None:
     print(f"Orders: {[(order.user_id, order.total) for order in orders]}")
     print(f"Orders count: {len(orders)}")
 
+    # Get single user by name
+    single_user = await user_repo.get_item(filter_=UserFilter(name="Alice"))
+    print(f"Single user: {single_user.name if single_user else None}")
+
 
 if __name__ == "__main__":
     asyncio.run(main())

examples/dto_usage.py

@@ -47,6 +47,12 @@ async def main() -> None:
     )
     print(f"Created DTO: {user.model_dump()}")
 
+    # Read single DTO
+    single = await repository.get_item(
+        filter_=UserFilter(email="alice@example.com"),
+    )
+    print(f"Single DTO: {single.model_dump() if single else None}")
+
     # Read all — returned as DTOs
     items = [item async for item in repository.get_items()]
     print(f"Items as DTOs: {[item.model_dump() for item in items]}")

examples/filter_usage.py

@@ -40,8 +40,12 @@ async def main() -> None:
             BookTable(title=f"Book {index}", year=2020 + index),
         )
 
-    # Filter by year = 2025
+    # Get single item by filter
     year_filter = BookFilter(year=2025)
+    single = await repository.get_item(filter_=year_filter)
+    print(f"Single year = 2025: {single.title if single else None}")
+
+    # Filter by year = 2025
     filtered = [item async for item in repository.get_items(filter_=year_filter)]
     print(f"Year = 2025: {[book.title for book in filtered]}")
 

examples/transactions.py

@@ -54,6 +54,11 @@ async def main() -> None:
     count = await repository.get_items_count()
     print(f"Items after nested rollback: {count}")  # 2
 
+    # Read single item inside a transaction
+    async with repository.transaction():
+        item = await repository.get_item(filter_=ProductFilter(name="Laptop"))
+        print(f"Single in transaction: {item.name if item else None}")
+
 
 if __name__ == "__main__":
     asyncio.run(main())

metaorm/repositories.py

@@ -68,13 +68,21 @@ class BaseRepository:
 
         return items_count
 
+    async def get_item(
+        self,
+        filter_: BaseFilter | None = None,
+        sort: BaseSort | None = None,
+    ) -> BaseModel | None:
+        async for item in self.get_items(filter_=filter_, sort=sort):
+            return item
+
     async def get_items(
         self,
         filter_: BaseFilter | None = None,
         pagination: BasePagination | None = None,
         sort: BaseSort | None = None,
         options: Sequence[Any] | None = None,
-    ) -> AsyncGenerator[Any]:
+    ) -> AsyncGenerator[BaseModel]:
         table = self.get_table_type()
         statement = select(table)
         statement = append_to_statement(

tests/test_repositories.py

@@ -177,6 +177,68 @@ class TestBaseRepository:
         assert product_repository.get_dto_type() is None
         assert user_repository.get_dto_type() is User
 
+    async def test_get_item_returns_first_item(
+        self,
+        user_repository: UserRepository,
+    ) -> None:
+        await user_repository.create_item(
+            User(name="Alice", email="alice@example.com"),
+        )
+        await user_repository.create_item(
+            User(name="Bob", email="bob@example.com"),
+        )
+
+        item = await user_repository.get_item()
+
+        assert item is not None
+        assert item.name == "Alice"
+
+    async def test_get_item_returns_none_when_empty(
+        self,
+        user_repository: UserRepository,
+    ) -> None:
+        item = await user_repository.get_item()
+
+        assert item is None
+
+    async def test_get_item_with_filter(
+        self,
+        product_repository_settings: ProductRepository,
+    ) -> None:
+        await product_repository_settings.create_item(
+            ProductTable(name="Alpha", price=10.0),
+        )
+        await product_repository_settings.create_item(
+            ProductTable(name="Beta", price=20.0),
+        )
+
+        item = await product_repository_settings.get_item(
+            filter_=ProductFilter(name="Beta"),
+        )
+
+        assert item is not None
+        assert item.name == "Beta"
+
+    async def test_get_item_with_sort(
+        self,
+        user_repository: UserRepository,
+    ) -> None:
+        await user_repository.create_item(
+            User(name="Charlie", email="c@example.com"),
+        )
+        await user_repository.create_item(
+            User(name="Alice", email="a@example.com"),
+        )
+        await user_repository.create_item(
+            User(name="Bob", email="b@example.com"),
+        )
+
+        sort = BaseSort(sort_by="name", sort_by_order="asc")
+        item = await user_repository.get_item(sort=sort)
+
+        assert item is not None
+        assert item.name == "Alice"
+
     async def test_get_items_count_with_filter(
         self,
         product_repository_settings: ProductRepository,