feat(v2): add bulk_create to CreateServiceMixin

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: dobrotacreator

docs/v2/services.md

@@ -85,8 +85,8 @@ If `actor_id` is provided and the model does not have the field, a
 
 Each operation has a raw and serialized variant:
 
-- `create_raw`, `update_raw`, `upsert_raw` -> return SQLAlchemy model
-- `create`, `update`, `upsert` -> always serialize to schema
+- `create_raw`, `bulk_create_raw`, `update_raw`, `upsert_raw` -> return SQLAlchemy model
+- `create`, `bulk_create`, `update`, `upsert` -> always serialize to schema
 
 ## Pagination
 
@@ -117,6 +117,27 @@ user_schema = await service.create(session, payload)
 user_model = await service.create_raw(session, payload)
 ```
 
+### Bulk create
+
+```python
+payloads = [
+    UserCreateSchema(email="a@b.com", name="Alice"),
+    UserCreateSchema(email="c@d.com", name="Charlie"),
+]
+
+# Serialized — returns list[ListSchema]
+users = await service.bulk_create(session, payloads)
+
+# Raw — returns list[ModelType]
+models = await service.bulk_create_raw(session, payloads)
+
+# With actor tracking
+users = await service.bulk_create(session, payloads, actor_id=current_user_id)
+```
+
+> M2M relations are not supported in bulk create. Use single `create` calls
+> when you need M2M sync.
+
 ### Actor-aware update (updated_by)
 
 ```python

src/notora/v2/repositories/base.py

@@ -149,6 +149,13 @@ def create(
         options: Iterable[OptionSpec[ModelType]] | None = None,
     ) -> TypedReturnsRows[tuple[ModelType]]: ...
 
+    def bulk_create(
+        self,
+        payload: Sequence[dict[str, Any]],
+        *,
+        options: Iterable[OptionSpec[ModelType]] | None = None,
+    ) -> TypedReturnsRows[tuple[ModelType]]: ...
+
     def create_or_skip(
         self,
         payload: dict[str, Any],

src/notora/v2/services/mixins/create.py

@@ -46,6 +46,41 @@ async def create_raw(
             await self.sync_m2m_relations(session, self._extract_pk(entity), relation_payload)
         return entity
 
+    async def bulk_create_raw(
+        self,
+        session: AsyncSession,
+        data: Sequence[PydanticModel | dict[str, Any]],
+        *,
+        actor_id: Any | None = None,
+        options: Iterable[OptionSpec[ModelType]] | None = None,
+    ) -> list[ModelType]:
+        payloads = [
+            self._apply_updated_by(
+                self._dump_payload(item, exclude_unset=False),
+                actor_id,
+            )
+            for item in data
+        ]
+        query = self.repo.bulk_create(payloads, options=options)
+        return await self.execute_for_many(session, query)
+
+    async def bulk_create(
+        self,
+        session: AsyncSession,
+        data: Sequence[PydanticModel | dict[str, Any]],
+        *,
+        actor_id: Any | None = None,
+        options: Iterable[OptionSpec[ModelType]] | None = None,
+        schema: type[ListSchema] | None = None,
+    ) -> list[ListSchema]:
+        entities = await self.bulk_create_raw(
+            session,
+            data,
+            actor_id=actor_id,
+            options=options,
+        )
+        return self.serialize_many(entities, schema=schema)
+
     async def create(
         self,
         session: AsyncSession,

src/notora/v2/services/mixins/create.pyi

@@ -58,6 +58,34 @@ class CreateServiceMixin[
         options: Iterable[OptionSpec[ModelType]] | None = None,
         schema: type[SchemaT],
     ) -> SchemaT: ...
+    async def bulk_create_raw(
+        self,
+        session: AsyncSession,
+        data: Sequence[PydanticModel | dict[str, Any]],
+        *,
+        actor_id: Any | None = None,
+        options: Iterable[OptionSpec[ModelType]] | None = None,
+    ) -> list[ModelType]: ...
+    @overload
+    async def bulk_create(
+        self,
+        session: AsyncSession,
+        data: Sequence[PydanticModel | dict[str, Any]],
+        *,
+        actor_id: Any | None = None,
+        options: Iterable[OptionSpec[ModelType]] | None = None,
+        schema: None = ...,
+    ) -> list[ListSchema]: ...
+    @overload
+    async def bulk_create[SchemaT: BaseResponseSchema](
+        self,
+        session: AsyncSession,
+        data: Sequence[PydanticModel | dict[str, Any]],
+        *,
+        actor_id: Any | None = None,
+        options: Iterable[OptionSpec[ModelType]] | None = None,
+        schema: type[SchemaT],
+    ) -> list[SchemaT]: ...
 
 class CreateOrSkipServiceMixin[
     PKType,

src/notora/v2/services/mixins/executor.py

@@ -79,6 +79,14 @@ async def execute_optional(
         result = await self.execute(session, statement)
         return cast(ModelType | None, result.unique().scalar_one_or_none())
 
+    async def execute_for_many(
+        self,
+        session: AsyncSession,
+        statement: TypedReturnsRows[tuple[ModelType]],
+    ) -> list[ModelType]:
+        result = await self.execute(session, statement)
+        return list(cast('ScalarResult[ModelType]', result.unique().scalars()).all())
+
     def _translate_integrity_error(self, err: exc.IntegrityError) -> Exception:
         if match := self._fk_constraint_pattern.match(err.args[0]):
             fk_name = match.group('fk_name')

tests/v2/test_integration/test_service_crud.py

@@ -3,7 +3,7 @@
 import pytest
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from notora.v2.exceptions.common import NotFoundError
+from notora.v2.exceptions.common import AlreadyExistsError, NotFoundError
 from notora.v2.repositories.params import PaginationParams, QueryParams
 from notora.v2.schemas.base import PaginatedResponseSchema
 from tests.v2.test_integration.mocks.model import V2User
@@ -182,3 +182,68 @@ async def test_service_retrieve_create_or_skip_upsert_and_delete(
 
     with pytest.raises(NotFoundError):
         await user_service.retrieve(db_session, created.id)
+
+
+async def test_service_bulk_create(
+    db_session: AsyncSession,
+    user_service: V2UserService,
+) -> None:
+    payloads = [
+        _create_user_payload('bulk1@ex.com', 'Bulk1'),
+        _create_user_payload('bulk2@ex.com', 'Bulk2'),
+        _create_user_payload('bulk3@ex.com', 'Bulk3'),
+    ]
+    created = await user_service.bulk_create(db_session, payloads)
+    await db_session.commit()
+
+    assert len(created) == len(payloads)
+    assert all(isinstance(item, V2UserResponseSchema) for item in created)
+    assert {item.email for item in created} == {'bulk1@ex.com', 'bulk2@ex.com', 'bulk3@ex.com'}
+
+
+async def test_service_bulk_create_raw(
+    db_session: AsyncSession,
+    user_service: V2UserService,
+) -> None:
+    payloads = [
+        _create_user_payload('raw1@ex.com', 'Raw1'),
+        _create_user_payload('raw2@ex.com', 'Raw2'),
+    ]
+    created = await user_service.bulk_create_raw(db_session, payloads)
+    await db_session.commit()
+
+    assert len(created) == len(payloads)
+    assert all(isinstance(item, V2User) for item in created)
+
+
+async def test_service_bulk_create_with_actor_id(
+    db_session: AsyncSession,
+    user_service: V2UserService,
+) -> None:
+    actor_id = uuid4()
+    payloads = [
+        _create_user_payload('actor1@ex.com', 'Actor1'),
+        _create_user_payload('actor2@ex.com', 'Actor2'),
+    ]
+    created = await user_service.bulk_create(db_session, payloads, actor_id=actor_id)
+    await db_session.commit()
+
+    assert len(created) == len(payloads)
+    assert all(item.updated_by == actor_id for item in created)
+
+
+async def test_service_bulk_create_duplicate_raises(
+    db_session: AsyncSession,
+    user_service: V2UserService,
+) -> None:
+    payload = _create_user_payload('dup@ex.com', 'Original')
+    await user_service.create(db_session, payload)
+    await db_session.commit()
+
+    with pytest.raises(AlreadyExistsError):
+        await user_service.bulk_create(
+            db_session,
+            [
+                _create_user_payload('dup@ex.com', 'Duplicate'),
+            ],
+        )