Bulk Views
Opt-in views for batch collection operations. Import from fastapi_views.views.bulk.
Bulk views follow the same repository pattern as generic views: supply a repository satisfying the AsyncBulkRepository / BulkRepository protocol (a standalone protocol requiring only create_many, bulk_update, update_many and delete_many), plus the action schemas, and the view wires up one route — bulk_route, /bulk by default — with each action selected by the HTTP method:
| Method |
Action |
Repository call |
Success |
POST |
bulk_create |
create_many |
201 |
PUT |
bulk_update |
bulk_update (per item) |
204 |
PATCH |
update_many |
update_many (filtered) |
200 |
DELETE |
bulk_delete |
delete_many (filtered) |
204 |
Every operation is all-or-nothing (one transaction).
repository_options / get_repository_options(action) are defined once on BaseGenericBulkAPIView and reach all four calls. The filtered actions (PATCH, DELETE) build their keyword arguments from the resolved filter and then add the options through merge_repository_options(kwargs, action), which raises TypeError when an option key collides with a resolved filter key; override it to pick a precedence. All four protocol methods declare **kwargs so they can receive the options, and each one's leading parameter is positional-only — an implementation must declare it positional-only too in order to type-check as conforming.
The hooks that receive objects — after_bulk_create(objs) and after_update_many(objs) — use the same parameter name on the sync and async views.
For a complete walkthrough see Bulk actions.
BaseBulkAPIView
Bases: APIView
Common base for bulk views: every bulk action lives on one route.
All four actions are registered under :attr:bulk_route and told apart by
the HTTP method — POST creates, PUT updates per item, PATCH
updates the rows a filter selects, DELETE removes them.
Source code in fastapi_views/views/bulk.py
| class BaseBulkAPIView(APIView):
"""Common base for bulk views: every bulk action lives on one route.
All four actions are registered under :attr:`bulk_route` and told apart by
the HTTP method — ``POST`` creates, ``PUT`` updates per item, ``PATCH``
updates the rows a filter selects, ``DELETE`` removes them.
"""
bulk_route: str = "/bulk"
|
BulkCreateAPIView
Bases: BaseBulkCreateAPIView, Generic[P]
Sync bulk create.
Source code in fastapi_views/views/bulk.py
| class BulkCreateAPIView(BaseBulkCreateAPIView, Generic[P]):
"""Sync bulk create."""
@classmethod
def get_bulk_create_endpoint(cls, status_code: int) -> Endpoint:
schema = cls.get_response_schema(action="bulk_create")
def endpoint(
self: BulkCreateAPIView, *args: P.args, **kwargs: P.kwargs
) -> Response:
objs = self.bulk_create(*args, **kwargs)
if not self.return_on_create:
objs = None
return self.get_response(objs, status_code=status_code, schema=schema)
cls._patch_endpoint_signature(endpoint, cls.bulk_create)
return endpoint
@abstractmethod
def bulk_create(self, *args: P.args, **kwargs: P.kwargs) -> Any:
raise NotImplementedError
|
AsyncBulkCreateAPIView
Bases: BaseBulkCreateAPIView, Generic[P]
Async bulk create.
Source code in fastapi_views/views/bulk.py
| class AsyncBulkCreateAPIView(BaseBulkCreateAPIView, Generic[P]):
"""Async bulk create."""
@classmethod
def get_bulk_create_endpoint(cls, status_code: int) -> Endpoint:
schema = cls.get_response_schema(action="bulk_create")
async def endpoint(
self: AsyncBulkCreateAPIView, *args: P.args, **kwargs: P.kwargs
) -> Response:
objs = await self.bulk_create(*args, **kwargs)
if not self.return_on_create:
objs = None
return self.get_response(objs, status_code=status_code, schema=schema)
cls._patch_endpoint_signature(endpoint, cls.bulk_create)
return endpoint
@abstractmethod
async def bulk_create(self, *args: P.args, **kwargs: P.kwargs) -> Any:
raise NotImplementedError
|
BaseBulkUpdateAPIView
Bases: BaseBulkAPIView
Per-item bulk update: many rows, each with its own values.
Backed by an executemany-style repository call which cannot return
rows, so the route responds with 204 No Content.
Source code in fastapi_views/views/bulk.py
| class BaseBulkUpdateAPIView(BaseBulkAPIView):
"""Per-item bulk update: many rows, each with its own values.
Backed by an ``executemany``-style repository call which cannot return
rows, so the route responds with ``204 No Content``.
"""
@classmethod
def get_api_actions(cls, prefix: str = "") -> Generator[dict[str, Any], None, None]:
status_code = cls.get_status_code("bulk_update", HTTP_204_NO_CONTENT)
yield cls.get_api_action(
prefix=prefix,
path=cls.bulk_route,
endpoint=cls.get_bulk_update_endpoint(status_code),
methods=["PUT"],
status_code=status_code,
response_class=Response,
action="bulk_update",
extra_errors=(NotFound, Conflict),
)
yield from super().get_api_actions(prefix)
@classmethod
@abstractmethod
def get_bulk_update_endpoint(cls, status_code: int) -> Endpoint:
raise NotImplementedError
|
BulkUpdateAPIView
Bases: BaseBulkUpdateAPIView, Generic[P]
Sync per-item bulk update.
Source code in fastapi_views/views/bulk.py
| class BulkUpdateAPIView(BaseBulkUpdateAPIView, Generic[P]):
"""Sync per-item bulk update."""
@classmethod
def get_bulk_update_endpoint(cls, status_code: int) -> Endpoint:
def endpoint(
self: BulkUpdateAPIView, *args: P.args, **kwargs: P.kwargs
) -> Response:
self.bulk_update(*args, **kwargs)
return Response(status_code=status_code)
cls._patch_endpoint_signature(endpoint, cls.bulk_update)
return endpoint
@abstractmethod
def bulk_update(self, *args: P.args, **kwargs: P.kwargs) -> Any:
raise NotImplementedError
|
AsyncBulkUpdateAPIView
Bases: BaseBulkUpdateAPIView, Generic[P]
Async per-item bulk update.
Source code in fastapi_views/views/bulk.py
| class AsyncBulkUpdateAPIView(BaseBulkUpdateAPIView, Generic[P]):
"""Async per-item bulk update."""
@classmethod
def get_bulk_update_endpoint(cls, status_code: int) -> Endpoint:
async def endpoint(
self: AsyncBulkUpdateAPIView, *args: P.args, **kwargs: P.kwargs
) -> Response:
await self.bulk_update(*args, **kwargs)
return Response(status_code=status_code)
cls._patch_endpoint_signature(endpoint, cls.bulk_update)
return endpoint
@abstractmethod
async def bulk_update(self, *args: P.args, **kwargs: P.kwargs) -> Any:
raise NotImplementedError
|
BaseUpdateManyAPIView
Bases: BaseBulkAPIView
Filtered update: apply one set of values to every row a filter selects.
The repository call can use RETURNING, so the route responds with the
updated objects by default.
Source code in fastapi_views/views/bulk.py
| class BaseUpdateManyAPIView(BaseBulkAPIView):
"""Filtered update: apply one set of values to every row a filter selects.
The repository call can use ``RETURNING``, so the route responds with the
updated objects by default.
"""
return_on_update: bool = True
@classmethod
def get_response_schema(cls, action: Action | None = None) -> Any:
if action == "update_many":
return list[cls.response_schema] # type: ignore[name-defined]
return super().get_response_schema(action)
@classmethod
def get_api_actions(cls, prefix: str = "") -> Generator[dict[str, Any], None, None]:
status_code = cls.get_status_code("update_many", HTTP_200_OK)
yield cls.get_api_action(
prefix=prefix,
path=cls.bulk_route,
endpoint=cls.get_update_many_endpoint(status_code),
methods=["PATCH"],
status_code=status_code,
action="update_many",
extra_errors=(Conflict,),
)
yield from super().get_api_actions(prefix)
@classmethod
@abstractmethod
def get_update_many_endpoint(cls, status_code: int) -> Endpoint:
raise NotImplementedError
|
UpdateManyAPIView
Bases: BaseUpdateManyAPIView, Generic[P]
Sync filtered update.
Source code in fastapi_views/views/bulk.py
| class UpdateManyAPIView(BaseUpdateManyAPIView, Generic[P]):
"""Sync filtered update."""
@classmethod
def get_update_many_endpoint(cls, status_code: int) -> Endpoint:
schema = cls.get_response_schema(action="update_many")
def endpoint(
self: UpdateManyAPIView, *args: P.args, **kwargs: P.kwargs
) -> Response:
objs = self.update_many(*args, **kwargs)
if not self.return_on_update:
objs = None
return self.get_response(objs, status_code=status_code, schema=schema)
cls._patch_endpoint_signature(endpoint, cls.update_many)
return endpoint
@abstractmethod
def update_many(self, *args: P.args, **kwargs: P.kwargs) -> Any:
raise NotImplementedError
|
AsyncUpdateManyAPIView
Bases: BaseUpdateManyAPIView, Generic[P]
Async filtered update.
Source code in fastapi_views/views/bulk.py
| class AsyncUpdateManyAPIView(BaseUpdateManyAPIView, Generic[P]):
"""Async filtered update."""
@classmethod
def get_update_many_endpoint(cls, status_code: int) -> Endpoint:
schema = cls.get_response_schema(action="update_many")
async def endpoint(
self: AsyncUpdateManyAPIView, *args: P.args, **kwargs: P.kwargs
) -> Response:
objs = await self.update_many(*args, **kwargs)
if not self.return_on_update:
objs = None
return self.get_response(objs, status_code=status_code, schema=schema)
cls._patch_endpoint_signature(endpoint, cls.update_many)
return endpoint
@abstractmethod
async def update_many(self, *args: P.args, **kwargs: P.kwargs) -> Any:
raise NotImplementedError
|
BulkDestroyAPIView
Bases: BaseBulkDestroyAPIView, Generic[P]
Sync bulk delete.
Source code in fastapi_views/views/bulk.py
| class BulkDestroyAPIView(BaseBulkDestroyAPIView, Generic[P]):
"""Sync bulk delete."""
@classmethod
def get_bulk_delete_endpoint(cls, status_code: int) -> Endpoint:
def endpoint(
self: BulkDestroyAPIView, *args: P.args, **kwargs: P.kwargs
) -> Response:
self.bulk_delete(*args, **kwargs)
return Response(status_code=status_code)
cls._patch_endpoint_signature(endpoint, cls.bulk_delete)
return endpoint
@abstractmethod
def bulk_delete(self, *args: P.args, **kwargs: P.kwargs) -> Any:
raise NotImplementedError
|
AsyncBulkDestroyAPIView
Bases: BaseBulkDestroyAPIView, Generic[P]
Async bulk delete.
Source code in fastapi_views/views/bulk.py
| class AsyncBulkDestroyAPIView(BaseBulkDestroyAPIView, Generic[P]):
"""Async bulk delete."""
@classmethod
def get_bulk_delete_endpoint(cls, status_code: int) -> Endpoint:
async def endpoint(
self: AsyncBulkDestroyAPIView, *args: P.args, **kwargs: P.kwargs
) -> Response:
await self.bulk_delete(*args, **kwargs)
return Response(status_code=status_code)
cls._patch_endpoint_signature(endpoint, cls.bulk_delete)
return endpoint
@abstractmethod
async def bulk_delete(self, *args: P.args, **kwargs: P.kwargs) -> Any:
raise NotImplementedError
|
BaseGenericBulkAPIView
Bases: GenericView
Source code in fastapi_views/views/bulk.py
| class BaseGenericBulkAPIView(GenericView):
repository_options: ClassVar[dict[str, Any]] = {}
def get_repository_options(
self,
action: Action | None = None, # noqa: ARG002
) -> dict[str, Any]:
return self.repository_options
def merge_repository_options(
self, kwargs: dict[str, Any], action: Action | None = None
) -> dict[str, Any]:
"""Add :meth:`get_repository_options` to already built keyword arguments.
Used by the filtered actions, whose repository call already receives the
resolved filter as keyword arguments. A key present in both is a
configuration error — dropping either the filter criterion or the option
would silently change what the request does — so it raises
:class:`TypeError`. Override to pick a precedence instead.
"""
options = self.get_repository_options(action)
if clashing := kwargs.keys() & options.keys():
msg = (
f"{type(self).__name__}: repository_options key(s) "
f"{sorted(clashing)} collide with the filter criteria of action "
f"{action!r}"
)
raise TypeError(msg)
return kwargs | options
|
merge_repository_options(kwargs: dict[str, Any], action: Action | None = None) -> dict[str, Any]
Add :meth:get_repository_options to already built keyword arguments.
Used by the filtered actions, whose repository call already receives the
resolved filter as keyword arguments. A key present in both is a
configuration error — dropping either the filter criterion or the option
would silently change what the request does — so it raises
:class:TypeError. Override to pick a precedence instead.
Source code in fastapi_views/views/bulk.py
| def merge_repository_options(
self, kwargs: dict[str, Any], action: Action | None = None
) -> dict[str, Any]:
"""Add :meth:`get_repository_options` to already built keyword arguments.
Used by the filtered actions, whose repository call already receives the
resolved filter as keyword arguments. A key present in both is a
configuration error — dropping either the filter criterion or the option
would silently change what the request does — so it raises
:class:`TypeError`. Override to pick a precedence instead.
"""
options = self.get_repository_options(action)
if clashing := kwargs.keys() & options.keys():
msg = (
f"{type(self).__name__}: repository_options key(s) "
f"{sorted(clashing)} collide with the filter criteria of action "
f"{action!r}"
)
raise TypeError(msg)
return kwargs | options
|
AsyncGenericBulkCreateAPIView
Bases: BaseGenericBulkCreateAPIView, AsyncBulkCreateAPIView, WithAsyncBulkRepositoryMixin[M]
Async repository-backed bulk create.
Source code in fastapi_views/views/bulk.py
| class AsyncGenericBulkCreateAPIView(
BaseGenericBulkCreateAPIView,
AsyncBulkCreateAPIView,
WithAsyncBulkRepositoryMixin[M],
):
"""Async repository-backed bulk create."""
async def bulk_create(self, items: list[BaseModel]) -> Sequence[M]:
extra = self.get_kwargs("bulk_create")
data = [item.model_dump() | extra for item in items]
await self.before_bulk_create(data)
objs = await self.repository.create_many(
data, **self.get_repository_options("bulk_create")
)
await self.after_bulk_create(objs)
return objs
async def before_bulk_create(self, data: list[dict[str, Any]]) -> None:
"""Hook receiving the validated payloads before the repository call."""
async def after_bulk_create(self, objs: Sequence[M]) -> None:
"""Hook receiving the created objects before the response is built."""
|
before_bulk_create(data: list[dict[str, Any]]) -> None
async
Hook receiving the validated payloads before the repository call.
Source code in fastapi_views/views/bulk.py
| async def before_bulk_create(self, data: list[dict[str, Any]]) -> None:
"""Hook receiving the validated payloads before the repository call."""
|
after_bulk_create(objs: Sequence[M]) -> None
async
Hook receiving the created objects before the response is built.
Source code in fastapi_views/views/bulk.py
| async def after_bulk_create(self, objs: Sequence[M]) -> None:
"""Hook receiving the created objects before the response is built."""
|
GenericBulkCreateAPIView
Bases: BaseGenericBulkCreateAPIView, BulkCreateAPIView, WithBulkRepositoryMixin[M]
Sync repository-backed bulk create.
Source code in fastapi_views/views/bulk.py
| class GenericBulkCreateAPIView(
BaseGenericBulkCreateAPIView,
BulkCreateAPIView,
WithBulkRepositoryMixin[M],
):
"""Sync repository-backed bulk create."""
def bulk_create(self, items: list[BaseModel]) -> Sequence[M]:
extra = self.get_kwargs("bulk_create")
data = [item.model_dump() | extra for item in items]
self.before_bulk_create(data)
objs = self.repository.create_many(
data, **self.get_repository_options("bulk_create")
)
self.after_bulk_create(objs)
return objs
def before_bulk_create(self, data: list[dict[str, Any]]) -> None:
"""Hook receiving the validated payloads before the repository call."""
def after_bulk_create(self, objs: Sequence[M]) -> None:
"""Hook receiving the created objects before the response is built."""
|
before_bulk_create(data: list[dict[str, Any]]) -> None
Hook receiving the validated payloads before the repository call.
Source code in fastapi_views/views/bulk.py
| def before_bulk_create(self, data: list[dict[str, Any]]) -> None:
"""Hook receiving the validated payloads before the repository call."""
|
after_bulk_create(objs: Sequence[M]) -> None
Hook receiving the created objects before the response is built.
Source code in fastapi_views/views/bulk.py
| def after_bulk_create(self, objs: Sequence[M]) -> None:
"""Hook receiving the created objects before the response is built."""
|
AsyncGenericBulkUpdateAPIView
Bases: BaseGenericBulkUpdateAPIView, AsyncBulkUpdateAPIView, WithAsyncBulkRepositoryMixin[M]
Async repository-backed per-item bulk update.
Source code in fastapi_views/views/bulk.py
| class AsyncGenericBulkUpdateAPIView(
BaseGenericBulkUpdateAPIView,
AsyncBulkUpdateAPIView,
WithAsyncBulkRepositoryMixin[M],
):
"""Async repository-backed per-item bulk update."""
async def bulk_update(self, items: list[BaseModel]) -> None:
extra = self.get_kwargs("bulk_update")
data = [item.model_dump() | extra for item in items]
await self.before_bulk_update(data)
await self.repository.bulk_update(
data, **self.get_repository_options("bulk_update")
)
await self.after_bulk_update()
async def before_bulk_update(self, data: list[dict[str, Any]]) -> None:
"""Hook receiving the validated payloads before the repository call."""
async def after_bulk_update(self) -> None:
"""Hook invoked after rows were updated."""
|
before_bulk_update(data: list[dict[str, Any]]) -> None
async
Hook receiving the validated payloads before the repository call.
Source code in fastapi_views/views/bulk.py
| async def before_bulk_update(self, data: list[dict[str, Any]]) -> None:
"""Hook receiving the validated payloads before the repository call."""
|
after_bulk_update() -> None
async
Hook invoked after rows were updated.
Source code in fastapi_views/views/bulk.py
| async def after_bulk_update(self) -> None:
"""Hook invoked after rows were updated."""
|
GenericBulkUpdateAPIView
Bases: BaseGenericBulkUpdateAPIView, BulkUpdateAPIView, WithBulkRepositoryMixin[M]
Sync repository-backed per-item bulk update.
Source code in fastapi_views/views/bulk.py
| class GenericBulkUpdateAPIView(
BaseGenericBulkUpdateAPIView,
BulkUpdateAPIView,
WithBulkRepositoryMixin[M],
):
"""Sync repository-backed per-item bulk update."""
def bulk_update(self, items: list[BaseModel]) -> None:
extra = self.get_kwargs("bulk_update")
data = [item.model_dump() | extra for item in items]
self.before_bulk_update(data)
self.repository.bulk_update(data, **self.get_repository_options("bulk_update"))
self.after_bulk_update()
def before_bulk_update(self, data: list[dict[str, Any]]) -> None:
"""Hook receiving the validated payloads before the repository call."""
def after_bulk_update(self) -> None:
"""Hook invoked after rows were updated."""
|
before_bulk_update(data: list[dict[str, Any]]) -> None
Hook receiving the validated payloads before the repository call.
Source code in fastapi_views/views/bulk.py
| def before_bulk_update(self, data: list[dict[str, Any]]) -> None:
"""Hook receiving the validated payloads before the repository call."""
|
after_bulk_update() -> None
Hook invoked after rows were updated.
Source code in fastapi_views/views/bulk.py
| def after_bulk_update(self) -> None:
"""Hook invoked after rows were updated."""
|
AsyncGenericUpdateManyAPIView
Bases: BaseGenericUpdateManyAPIView, AsyncUpdateManyAPIView, WithAsyncBulkRepositoryMixin[M]
Async filtered update: one set of values applied to the matched rows.
Source code in fastapi_views/views/bulk.py
| class AsyncGenericUpdateManyAPIView(
BaseGenericUpdateManyAPIView,
AsyncUpdateManyAPIView,
WithAsyncBulkRepositoryMixin[M],
):
"""Async filtered update: one set of values applied to the matched rows."""
async def update_many(self, values: BaseModel, filter: BaseFilter) -> Sequence[M]:
data = values.model_dump(exclude_unset=True)
args, kwargs = self.get_filter_args(filter, "update_many")
await self.before_update_many(data)
objs = await self.repository.update_many(
data, *args, **self.merge_repository_options(kwargs, "update_many")
)
await self.after_update_many(objs)
return objs
async def before_update_many(self, values: dict[str, Any]) -> None:
"""Hook receiving the validated values before the repository call."""
async def after_update_many(self, objs: Sequence[M]) -> None:
"""Hook receiving the updated objects before the response is built."""
|
before_update_many(values: dict[str, Any]) -> None
async
Hook receiving the validated values before the repository call.
Source code in fastapi_views/views/bulk.py
| async def before_update_many(self, values: dict[str, Any]) -> None:
"""Hook receiving the validated values before the repository call."""
|
after_update_many(objs: Sequence[M]) -> None
async
Hook receiving the updated objects before the response is built.
Source code in fastapi_views/views/bulk.py
| async def after_update_many(self, objs: Sequence[M]) -> None:
"""Hook receiving the updated objects before the response is built."""
|
GenericUpdateManyAPIView
Bases: BaseGenericUpdateManyAPIView, UpdateManyAPIView, WithBulkRepositoryMixin[M]
Sync filtered update: one set of values applied to the matched rows.
Source code in fastapi_views/views/bulk.py
| class GenericUpdateManyAPIView(
BaseGenericUpdateManyAPIView,
UpdateManyAPIView,
WithBulkRepositoryMixin[M],
):
"""Sync filtered update: one set of values applied to the matched rows."""
def update_many(self, values: BaseModel, filter: BaseFilter) -> Sequence[M]:
data = values.model_dump(exclude_unset=True)
args, kwargs = self.get_filter_args(filter, "update_many")
self.before_update_many(data)
objs = self.repository.update_many(
data, *args, **self.merge_repository_options(kwargs, "update_many")
)
self.after_update_many(objs)
return objs
def before_update_many(self, values: dict[str, Any]) -> None:
"""Hook receiving the validated values before the repository call."""
def after_update_many(self, objs: Sequence[M]) -> None:
"""Hook receiving the updated objects before the response is built."""
|
before_update_many(values: dict[str, Any]) -> None
Hook receiving the validated values before the repository call.
Source code in fastapi_views/views/bulk.py
| def before_update_many(self, values: dict[str, Any]) -> None:
"""Hook receiving the validated values before the repository call."""
|
after_update_many(objs: Sequence[M]) -> None
Hook receiving the updated objects before the response is built.
Source code in fastapi_views/views/bulk.py
| def after_update_many(self, objs: Sequence[M]) -> None:
"""Hook receiving the updated objects before the response is built."""
|
AsyncGenericBulkDestroyAPIView
Bases: BaseGenericBulkDestroyAPIView, AsyncBulkDestroyAPIView, WithAsyncBulkRepositoryMixin[M]
Async bulk delete: resolve the filter, then repository.delete_many.
Source code in fastapi_views/views/bulk.py
| class AsyncGenericBulkDestroyAPIView(
BaseGenericBulkDestroyAPIView,
AsyncBulkDestroyAPIView,
WithAsyncBulkRepositoryMixin[M],
):
"""Async bulk delete: resolve the filter, then ``repository.delete_many``."""
async def bulk_delete(self, filter: BaseFilter) -> None:
await self.before_bulk_delete()
args, kwargs = self.get_filter_args(filter, "bulk_delete")
await self.repository.delete_many(
*args, **self.merge_repository_options(kwargs, "bulk_delete")
)
await self.after_bulk_delete()
async def before_bulk_delete(self) -> None:
"""Hook invoked before rows are deleted."""
async def after_bulk_delete(self) -> None:
"""Hook invoked after rows were deleted."""
|
before_bulk_delete() -> None
async
Hook invoked before rows are deleted.
Source code in fastapi_views/views/bulk.py
| async def before_bulk_delete(self) -> None:
"""Hook invoked before rows are deleted."""
|
after_bulk_delete() -> None
async
Hook invoked after rows were deleted.
Source code in fastapi_views/views/bulk.py
| async def after_bulk_delete(self) -> None:
"""Hook invoked after rows were deleted."""
|
GenericBulkDestroyAPIView
Bases: BaseGenericBulkDestroyAPIView, BulkDestroyAPIView, WithBulkRepositoryMixin[M]
Sync bulk delete: resolve the filter, then repository.delete_many.
Source code in fastapi_views/views/bulk.py
| class GenericBulkDestroyAPIView(
BaseGenericBulkDestroyAPIView,
BulkDestroyAPIView,
WithBulkRepositoryMixin[M],
):
"""Sync bulk delete: resolve the filter, then ``repository.delete_many``."""
def bulk_delete(self, filter: BaseFilter) -> None:
self.before_bulk_delete()
args, kwargs = self.get_filter_args(filter, "bulk_delete")
self.repository.delete_many(
*args, **self.merge_repository_options(kwargs, "bulk_delete")
)
self.after_bulk_delete()
def before_bulk_delete(self) -> None:
"""Hook invoked before rows are deleted."""
def after_bulk_delete(self) -> None:
"""Hook invoked after rows were deleted."""
|
before_bulk_delete() -> None
Hook invoked before rows are deleted.
Source code in fastapi_views/views/bulk.py
| def before_bulk_delete(self) -> None:
"""Hook invoked before rows are deleted."""
|
after_bulk_delete() -> None
Hook invoked after rows were deleted.
Source code in fastapi_views/views/bulk.py
| def after_bulk_delete(self) -> None:
"""Hook invoked after rows were deleted."""
|