Repository: saleor/saleor
Pull request: Add CustomerBulkUpdate mutation
I want to merge this change because it adds `CustomerBulkUpdate` mutation. Resolves #12057

<!-- Please mention all relevant issue numbers. -->

# Impact

* [ ] New migrations
* [x] New/Updated API fields or mutations
* [ ] Deprecated API fields or mutations
* [ ] Removed API types, fields, or mutations
* [ ] Documentation needs to be updated

# Pull Request Checklist

<!-- Please keep this section. It will make the maintainer's life easier. -->

* [ ] Privileged queries and mutations are guarded by proper permission checks
* [ ] Database queries are optimized and the number of queries is constant
* [ ] Database migration files are up to date
* [ ] The changes are tested
* [ ] GraphQL schema and type definitions are up to date
* [ ] Changes are mentioned in the changelog


Problem statement:
Bulk customer mutations
Introduce bulk update mutation for customers.

# 1. General Assumptions
  1. Allow updating multiple customers in one bulk mutation.
  2. `ID` or `externalReference` can be used to identify `Customer`.
  3. `CUSTOMER_UPDATED` webhook event should be sent for every updated customer instance.
  4. Correct `CustomerEvents` should be created depending on changed data.

# 2. API changes

```graphql
enum CustomerBulkUpdateErrorCode {
  BLANK
  DUPLICATED_INPUT_ITEM
  GRAPHQL_ERROR
  INVALID
  REQUIRED
  UNIQUE
  NOT_FOUND
  MAX_LENGTH
}

type CustomerBulkUpdateError {
  path: String
  message: String
  code: CustomerBulkUpdateErrorCode!
}

input CustomerBulkUpdateInput {
  id: ID
  externalReference: String
  input: CustomerInput!
}

type CustomerBulkUpdate {
  count: Int!
  results: [CustomerBulkResult!]!
}

type Mutation {
  customerBulkUpdate(
    customers: [CustomerBulkUpdateInput!]!
    errorPolicy: ErrorPolicyEnum = reject_everything
  ): CustomerBulkUpdate
}
```

# 3. Ideas and improvements to-do in a separate task
   1. Move updating `serach_document` for User to async Celeryy task.

Hints from issue discussion:
Mutation return password cannot be blank, but the password is not part of CustomerInput 
variables:
```
{
  "errorPolicy": "IGNORE_FAILED",
  "customers": [
    {
      "id": "VXNlcjo5NzU=",
      "input": {
        "firstName": "Carol"
      }
    }
  ]
}
```
Response:
```
        "customerBulkUpdate": {
            "count": 0,
            "results": [
                {
                    "customer": null,
                    "errors": [
                        {
                            "path": "input.password",
                            "message": "This field cannot be blank.",
                            "code": "BLANK"
                        }
                    ]
                }
            ],
            "errors": []
        }
```

Commit message: Add doc_category to customerBulkUpdate mutation

Patch to review:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index a5954d0d5d1..d72743d5870 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -15,6 +15,7 @@ All notable, unreleased changes to this project will be documented in this file.
 - Add new field `author` to `Manifest` and `App` object types - #12166 by @przlada
 - Add backwards compatibility for `taxCode` field - #12325 by @maarcingebala
 - Support resolving `Order` as an entity in Apollo Federation - #12328 by @binary-koan
+- [Preview] Add `CustomerBulkUpdate` mutation - #12268 by @SzymJ
 
 ### Other changes
 - Create order discounts for all voucher types - #12272 by @IKarbowiak
diff --git a/saleor/account/error_codes.py b/saleor/account/error_codes.py
index 001142e5ae4..b8df987ae82 100644
--- a/saleor/account/error_codes.py
+++ b/saleor/account/error_codes.py
@@ -38,6 +38,17 @@ class AccountErrorCode(Enum):
     ACCOUNT_NOT_CONFIRMED = "account_not_confirmed"
 
 
+class CustomerBulkUpdateErrorCode(Enum):
+    BLANK = "blank"
+    DUPLICATED_INPUT_ITEM = "duplicated_input_item"
+    GRAPHQL_ERROR = "graphql_error"
+    INVALID = "invalid"
+    REQUIRED = "required"
+    UNIQUE = "unique"
+    NOT_FOUND = "not_found"
+    MAX_LENGTH = "max_length"
+
+
 class PermissionGroupErrorCode(Enum):
     ASSIGN_NON_STAFF_MEMBER = "assign_non_staff_member"
     DUPLICATED_INPUT_ITEM = "duplicated_input_item"
diff --git a/saleor/graphql/account/bulk_mutations.py b/saleor/graphql/account/bulk_mutations.py
deleted file mode 100644
index 40e85b9b069..00000000000
--- a/saleor/graphql/account/bulk_mutations.py
+++ /dev/null
@@ -1,145 +0,0 @@
-from collections import defaultdict
-from typing import List
-
-import graphene
-from django.core.exceptions import ValidationError
-
-from ...account import models
-from ...account.error_codes import AccountErrorCode
-from ...permission.enums import AccountPermissions
-from ..core import ResolveInfo
-from ..core.doc_category import DOC_CATEGORY_USERS
-from ..core.mutations import BaseBulkMutation, ModelBulkDeleteMutation
-from ..core.types import AccountError, NonNullList, StaffError
-from ..plugins.dataloaders import get_plugin_manager_promise
-from .types import User
-from .utils import CustomerDeleteMixin, StaffDeleteMixin
-
-
-class UserBulkDelete(ModelBulkDeleteMutation):
-    class Arguments:
-        ids = NonNullList(
-            graphene.ID, required=True, description="List of user IDs to delete."
-        )
-
-    class Meta:
-        abstract = True
-
-
-class CustomerBulkDelete(CustomerDeleteMixin, UserBulkDelete):
-    class Meta:
-        description = "Deletes customers."
-        doc_category = DOC_CATEGORY_USERS
-        model = models.User
-        object_type = User
-        permissions = (AccountPermissions.MANAGE_USERS,)
-        error_type_class = AccountError
-        error_type_field = "account_errors"
-
-    @classmethod
-    def perform_mutation(cls, root, info: ResolveInfo, /, **data):
-        count, errors = super().perform_mutation(root, info, **data)
-        cls.post_process(info, count)
-        return count, errors
-
-    @classmethod
-    def bulk_action(cls, info: ResolveInfo, queryset, /):
-        instances = list(queryset)
-        queryset.delete()
-        manager = get_plugin_manager_promise(info.context).get()
-        for instance in instances:
-            manager.customer_deleted(instance)
-
-
-class StaffBulkDelete(StaffDeleteMixin, UserBulkDelete):
-    class Meta:
-        description = (
-            "Deletes staff users. Apps are not allowed to perform this mutation."
-        )
-        doc_category = DOC_CATEGORY_USERS
-        model = models.User
-        object_type = User
-        permissions = (AccountPermissions.MANAGE_STAFF,)
-        error_type_class = StaffError
-        error_type_field = "staff_errors"
-
-    @classmethod
-    def perform_mutation(  # type: ignore[override]
-        cls, _root, info: ResolveInfo, /, *, ids, **data
-    ):
-        instances = cls.get_nodes_or_error(ids, "id", User)
-        errors = cls.clean_instances(info, instances)
-        count = len(instances)
-        if not errors and count:
-            clean_instance_ids = [instance.pk for instance in instances]
-            qs = models.User.objects.filter(pk__in=clean_instance_ids)
-            cls.bulk_action(info, qs, **data)
-        else:
-            count = 0
-        return count, errors
-
-    @classmethod
-    def clean_instances(cls, info: ResolveInfo, users):
-        errors: defaultdict[str, List[ValidationError]] = defaultdict(list)
-
-        requestor = info.context.user
-        cls.check_if_users_can_be_deleted(info, users, "ids", errors)
-        cls.check_if_requestor_can_manage_users(requestor, users, "ids", errors)
-        cls.check_if_removing_left_not_manageable_permissions(
-            requestor, users, "ids", errors
-        )
-        return ValidationError(errors) if errors else {}
-
-    @classmethod
-    def bulk_action(cls, info: ResolveInfo, queryset, /):
-        instances = list(queryset)
-        queryset.delete()
-        manager = get_plugin_manager_promise(info.context).get()
-        for instance in instances:
-            manager.staff_deleted(instance)
-
-
-class UserBulkSetActive(BaseBulkMutation):
-    class Arguments:
-        ids = NonNullList(
-            graphene.ID, required=True, description="List of user IDs to (de)activate)."
-        )
-        is_active = graphene.Boolean(
-            required=True, description="Determine if users will be set active or not."
-        )
-
-    class Meta:
-        description = "Activate or deactivate users."
-        doc_category = DOC_CATEGORY_USERS
-        model = models.User
-        object_type = User
-        permissions = (AccountPermissions.MANAGE_USERS,)
-        error_type_class = AccountError
-        error_type_field = "account_errors"
-
-    @classmethod
-    def clean_instance(cls, info: ResolveInfo, instance):
-        if info.context.user == instance:
-            raise ValidationError(
-                {
-                    "is_active": ValidationError(
-                        "Cannot activate or deactivate your own account.",
-                        code=AccountErrorCode.ACTIVATE_OWN_ACCOUNT.value,
-                    )
-                }
-            )
-        elif instance.is_superuser:
-            raise ValidationError(
-                {
-                    "is_active": ValidationError(
-                        "Cannot activate or deactivate superuser's account.",
-                        code=AccountErrorCode.ACTIVATE_SUPERUSER_ACCOUNT.value,
-                    )
-                }
-            )
-
-    @classmethod
-    def bulk_action(  # type: ignore[override]
-        cls, _info: ResolveInfo, queryset, /, *, is_active
-    ):
-        queryset.update(is_active=is_active)
diff --git a/saleor/graphql/account/bulk_mutations/__init__.py b/saleor/graphql/account/bulk_mutations/__init__.py
new file mode 100644
index 00000000000..eff6ea18d68
--- /dev/null
+++ b/saleor/graphql/account/bulk_mutations/__init__.py
@@ -0,0 +1,11 @@
+from .customer_bulk_delete import CustomerBulkDelete
+from .customer_bulk_update import CustomerBulkUpdate
+from .staff_bulk_delete import StaffBulkDelete
+from .user_bulk_set_active import UserBulkSetActive
+
+__all__ = [
+    "CustomerBulkDelete",
+    "CustomerBulkUpdate",
+    "StaffBulkDelete",
+    "UserBulkSetActive",
+]
diff --git a/saleor/graphql/account/bulk_mutations/customer_bulk_delete.py b/saleor/graphql/account/bulk_mutations/customer_bulk_delete.py
new file mode 100644
index 00000000000..bc5bf3ef42d
--- /dev/null
+++ b/saleor/graphql/account/bulk_mutations/customer_bulk_delete.py
@@ -0,0 +1,46 @@
+import graphene
+
+from ....account import models
+from ....permission.enums import AccountPermissions
+from ...core import ResolveInfo
+from ...core.doc_category import DOC_CATEGORY_USERS
+from ...core.mutations import ModelBulkDeleteMutation
+from ...core.types import AccountError, NonNullList
+from ...plugins.dataloaders import get_plugin_manager_promise
+from ..types import User
+from ..utils import CustomerDeleteMixin
+
+
+class UserBulkDelete(ModelBulkDeleteMutation):
+    class Arguments:
+        ids = NonNullList(
+            graphene.ID, required=True, description="List of user IDs to delete."
+        )
+
+    class Meta:
+        abstract = True
+
+
+class CustomerBulkDelete(CustomerDeleteMixin, UserBulkDelete):
+    class Meta:
+        description = "Deletes customers."
+        doc_category = DOC_CATEGORY_USERS
+        model = models.User
+        object_type = User
+        permissions = (AccountPermissions.MANAGE_USERS,)
+        error_type_class = AccountError
+        error_type_field = "account_errors"
+
+    @classmethod
+    def perform_mutation(cls, root, info: ResolveInfo, /, **data):
+        count, errors = super().perform_mutation(root, info, **data)
+        cls.post_process(info, count)
+        return count, errors
+
+    @classmethod
+    def bulk_action(cls, info: ResolveInfo, queryset, /):
+        instances = list(queryset)
+        queryset.delete()
+        manager = get_plugin_manager_promise(info.context).get()
+        for instance in instances:
+            manager.customer_deleted(instance)
diff --git a/saleor/graphql/account/bulk_mutations/customer_bulk_update.py b/saleor/graphql/account/bulk_mutations/customer_bulk_update.py
new file mode 100644
index 00000000000..477a752d2d5
--- /dev/null
+++ b/saleor/graphql/account/bulk_mutations/customer_bulk_update.py
@@ -0,0 +1,598 @@
+from collections import defaultdict
+from copy import copy
+
+import graphene
+from django.core.exceptions import ValidationError
+from django.db.models import Q
+from graphene.utils.str_converters import to_camel_case
+
+from ....account import models
+from ....account.events import CustomerEvents
+from ....account.search import prepare_user_search_document_value
+from ....checkout import AddressType
+from ....core.tracing import traced_atomic_transaction
+from ....giftcard.utils import assign_user_gift_cards
+from ....order.utils import match_orders_with_new_user
+from ....permission.enums import AccountPermissions
+from ...core.descriptions import ADDED_IN_313, PREVIEW_FEATURE
+from ...core.doc_category import DOC_CATEGORY_USERS
+from ...core.enums import CustomerBulkUpdateErrorCode, ErrorPolicyEnum
+from ...core.mutations import BaseMutation, ModelMutation
+from ...core.types import (
+    BaseInputObjectType,
+    BaseObjectType,
+    CustomerBulkUpdateError,
+    NonNullList,
+)
+from ...core.utils import get_duplicated_values
+from ...core.validators import validate_one_of_args_is_in_mutation
+from ...plugins.dataloaders import get_app_promise, get_plugin_manager_promise
+from ..i18n import I18nMixin
+from ..mutations.base import BILLING_ADDRESS_FIELD, SHIPPING_ADDRESS_FIELD
+from ..mutations.staff import CustomerInput
+from ..types import User
+
+
+class CustomerBulkResult(BaseObjectType):
+    customer = graphene.Field(User, required=False, description="Customer data.")
+    errors = NonNullList(
+        CustomerBulkUpdateError,
+        required=False,
+        description="List of errors that occurred during the update attempt.",
+    )
+
+    class Meta:
+        doc_category = DOC_CATEGORY_USERS
+
+
+class CustomerBulkUpdateInput(BaseInputObjectType):
+    id = graphene.ID(description="ID of a customer to update.", required=False)
+    external_reference = graphene.String(
+        required=False,
+        description="External ID of a customer to update.",
+    )
+    input = CustomerInput(
+        description="Fields required to update a customer.", required=True
+    )
+
+    class Meta:
+        doc_category = DOC_CATEGORY_USERS
+
+
+class CustomerBulkUpdate(BaseMutation, I18nMixin):
+    count = graphene.Int(
+        required=True,
+        default_value=0,
+        description="Returns how many objects were created.",
+    )
+    results = NonNullList(
+        CustomerBulkResult,
+        required=True,
+        default_value=[],
+        description="List of the updated customers.",
+    )
+
+    class Arguments:
+        customers = NonNullList(
+            CustomerBulkUpdateInput,
+            required=True,
+            description="Input list of customers to update.",
+        )
+        error_policy = ErrorPolicyEnum(
+            required=False,
+            description=(
+                "Policies of error handling. DEFAULT: "
+                + ErrorPolicyEnum.REJECT_EVERYTHING.name
+            ),
+        )
+
+    class Meta:
+        description = "Updates customers." + ADDED_IN_313 + PREVIEW_FEATURE
+        doc_category = DOC_CATEGORY_USERS
+        permissions = (AccountPermissions.MANAGE_USERS,)
+        error_type_class = CustomerBulkUpdateError
+
+    @classmethod
+    def format_errors(cls, index, errors, index_error_map, field_prefix=None):
+        for key, value in errors.error_dict.items():
+            for e in value:
+                path = (
+                    to_camel_case(f"{field_prefix}.{key}")
+                    if field_prefix
+                    else to_camel_case(key)
+                )
+                index_error_map[index].append(
+                    CustomerBulkUpdateError(
+                        path=f"input.{path}",
+                        message=e.messages[0],
+                        code=e.code,
+                    )
+                )
+
+    @classmethod
+    def validate_customer(
+        cls, customer_id, external_ref, cleaned_input, index, index_error_map
+    ):
+        errors_count = 0
+        try:
+            validate_one_of_args_is_in_mutation(
+                "id",
+                customer_id,
+                "external_reference",
+                external_ref,
+                use_camel_case=True,
+            )
+        except ValidationError as exc:
+            index_error_map[index].append(
+                CustomerBulkUpdateError(
+                    message=exc.message,
+                    code=CustomerBulkUpdateErrorCode.INVALID.value,
+                )
+            )
+            errors_count += 1
+
+        if customer_id:
+            try:
+                type, customer_id = graphene.Node.from_global_id(customer_id)
+                if type != "User":
+                    index_error_map[index].append(
+                        CustomerBulkUpdateError(
+                            path="id",
+                            message="Invalid customer ID.",
+                            code=CustomerBulkUpdateErrorCode.INVALID.value,
+                        )
+                    )
+                    errors_count += 1
+                else:
+                    cleaned_input["id"] = customer_id
+            except Exception:
+                index_error_map[index].append(
+                    CustomerBulkUpdateError(
+                        path="id",
+                        message="Invalid customer ID.",
+                        code=CustomerBulkUpdateErrorCode.INVALID.value,
+                    )
+                )
+                errors_count += 1
+        return errors_count
+
+    @classmethod
+    def clean_address(
+        cls,
+        address_data,
+        address_type,
+        field,
+        index,
+        index_error_map,
+        format_check=True,
+        required_check=True,
+        enable_normalization=True,
+    ):
+        try:
+            address_form = cls.validate_address_form(
+                address_data,
+                address_type,
+                format_check=format_check,
+                required_check=required_check,
+                enable_normalization=enable_normalization,
+            )
+
+            return address_form.cleaned_data
+        except ValidationError as exc:
+            cls.format_errors(index, exc, index_error_map, field_prefix=field)
+
+    @classmethod
+    def clean_customers(cls, info, customers_input, index_error_map):
+        cleaned_inputs_map: dict = {}
+
+        external_refs = [
+            customer_input["input"]["external_reference"]
+            for customer_input in customers_input
+            if customer_input["input"].get("external_reference")
+        ]
+
+        emails = [
+            customer_input["input"]["email"].lower()
+            for customer_input in customers_input
+            if customer_input["input"].get("email")
+        ]
+
+        duplicated_refs = get_duplicated_values(external_refs)
+        duplicated_emails = get_duplicated_values(emails)
+
+        for index, customer_input in enumerate(customers_input):
+            base_error_count = 0
+            customer_id = customer_input.get("id")
+            customer_external_ref = customer_input.get("external_reference")
+
+            shipping_address_data = customer_input["input"].pop(
+                SHIPPING_ADDRESS_FIELD, None
+            )
+            billing_address_data = customer_input["input"].pop(
+                BILLING_ADDRESS_FIELD, None
+            )
+
+            customer_input["input"] = ModelMutation.clean_input(
+                info, None, customer_input["input"], input_cls=CustomerInput
+            )
+
+            new_external_ref = customer_input["input"].get("external_reference")
+
+            base_error_count += cls.validate_customer(
+                customer_id,
+                customer_external_ref,
+                customer_input,
+                index,
+                index_error_map,
+            )
+
+            if new_external_ref in duplicated_refs:
+                message = "Duplicated externalReference value."
+                index_error_map[index].append(
+                    CustomerBulkUpdateError(
+                        path="input.externalReference",
+                        message=message,
+                        code=CustomerBulkUpdateErrorCode.DUPLICATED_INPUT_ITEM.value,
+                    )
+                )
+                base_error_count += 1
+
+            if shipping_address_data:
+                clean_shipping_address = cls.clean_address(
+                    shipping_address_data,
+                    address_type=AddressType.SHIPPING,
+                    field=SHIPPING_ADDRESS_FIELD,
+                    index=index,
+                    index_error_map=index_error_map,
+                )
+                customer_input["input"][SHIPPING_ADDRESS_FIELD] = clean_shipping_address
+
+            if billing_address_data:
+                clean_billing_address = cls.clean_address(
+                    billing_address_data,
+                    address_type=AddressType.BILLING,
+                    field=BILLING_ADDRESS_FIELD,
+                    index=index,
+                    index_error_map=index_error_map,
+                )
+                customer_input["input"][BILLING_ADDRESS_FIELD] = clean_billing_address
+
+            if email := customer_input.get("email"):
+                customer_input["email"] = email.lower()
+
+                if customer_input["email"] in duplicated_emails:
+                    message = "Duplicated email value."
+                    code = CustomerBulkUpdateErrorCode.DUPLICATED_INPUT_ITEM.value
+                    index_error_map[index].append(
+                        CustomerBulkUpdateError(
+                            path="input.email", message=message, code=code
+                        )
+                    )
+                    base_error_count += 1
+
+            if base_error_count > 0:
+                cleaned_inputs_map[index] = None
+            else:
+                cleaned_inputs_map[index] = customer_input
+
+        return cleaned_inputs_map
+
+    @classmethod
+    def get_customers(cls, cleaned_inputs_map: dict) -> list[models.User]:
+        lookup = Q()
+        for customer_input in cleaned_inputs_map.values():
+            if not customer_input:
+                continue
+
+            single_customer_lookup = Q()
+
+            if customer_id := customer_input.get("id"):
+                single_customer_lookup |= Q(pk=customer_id)
+            else:
+                single_customer_lookup |= Q(
+                    external_reference=customer_input.get("external_reference")
+                )
+            lookup |= single_customer_lookup
+
+        customers = models.User.objects.filter(lookup, is_staff=False)
+        return list(customers)
+
+    @classmethod
+    def _get_customer(cls, customer_id, external_ref):
+        if customer_id:
+            return lambda customer: str(customer.id) == customer_id
+        else:
+            return lambda customer: customer.external_reference == external_ref
+
+    @classmethod
+    def update_address(cls, info, instance, data, field):
+        address = getattr(instance, field) or models.Address()
+        address = cls.construct_instance(address, data)
+        cls.clean_instance(info, address)
+        return address
+
+    @classmethod
+    def update_customers(cls, info, cleaned_inputs_map, index_error_map):
+        instances_data_and_errors_list: list = []
+        customers_list = cls.get_customers(cleaned_inputs_map)
+
+        for index, cleaned_input in cleaned_inputs_map.items():
+            if not cleaned_input:
+                instances_data_and_errors_list.append(
+                    {"instance": None, "errors": index_error_map[index]}
+                )
+                continue
+
+            customer_id = cleaned_input.get("id")
+            external_ref = cleaned_input.get("external_reference")
+            data = cleaned_input["input"]
+            shipping_address_input = data.pop(SHIPPING_ADDRESS_FIELD, None)
+            billing_address_input = data.pop(BILLING_ADDRESS_FIELD, None)
+
+            filtered_customers = list(
+                filter(
+                    cls._get_customer(customer_id, external_ref),
+                    customers_list,
+                )
+            )
+            if filtered_customers:
+                try:
+                    shipping_address = None
+                    billing_address = None
+
+                    old_instance = filtered_customers[0]
+                    new_instance = cls.construct_instance(copy(old_instance), data)
+                    new_instance.full_clean(exclude=["password"])
+
+                    if shipping_address_input:
+                        shipping_address = cls.update_address(
+                            info,
+                            new_instance,
+                            shipping_address_input,
+                            SHIPPING_ADDRESS_FIELD,
+                        )
+                    if billing_address_input:
+                        billing_address = cls.update_address(
+                            info,
+                            new_instance,
+                            billing_address_input,
+                            BILLING_ADDRESS_FIELD,
+                        )
+
+                    instances_data_and_errors_list.append(
+                        {
+                            "instance": new_instance,
+                            "old_instance": old_instance,
+                            "errors": index_error_map[index],
+                            SHIPPING_ADDRESS_FIELD: shipping_address,
+                            BILLING_ADDRESS_FIELD: billing_address,
+                        }
+                    )
+                except ValidationError as exc:
+                    cls.format_errors(index, exc, index_error_map)
+                    instances_data_and_errors_list.append(
+                        {"instance": None, "errors": index_error_map[index]}
+                    )
+            else:
+                index_error_map[index].append(
+                    CustomerBulkUpdateError(
+                        message="Customer was not found.",
+                        code=CustomerBulkUpdateErrorCode.NOT_FOUND.value,
+                    )
+                )
+                instances_data_and_errors_list.append(
+                    {"instance": None, "errors": index_error_map[index]}
+                )
+
+        return instances_data_and_errors_list
+
+    @classmethod
+    def save_customers(cls, instances_data_with_errors_list, manager):
+        old_instances = []
+        customers_to_update = []
+        addresses_to_create = []
+        addresses_to_update = []
+
+        customer_instance_new_addresses_map: dict = defaultdict(list)
+
+        for customer_data in instances_data_with_errors_list:
+            customer = customer_data["instance"]
+
+            if not customer:
+                continue
+
+            customers_to_update.append(customer)
+            old_instances.append(customer_data["old_instance"])
+
+            if shipping_address := customer_data[SHIPPING_ADDRESS_FIELD]:
+                shipping_address = manager.change_user_address(
+                    shipping_address, "shipping", customer, save=False
+                )
+                if customer.default_shipping_address:
+                    addresses_to_update.append(shipping_address)
+                else:
+                    customer.default_shipping_address = shipping_address
+                    addresses_to_create.append(shipping_address)
+                    customer_instance_new_addresses_map[customer].append(
+                        shipping_address
+                    )
+
+            if billing_address := customer_data[BILLING_ADDRESS_FIELD]:
+                billing_address = manager.change_user_address(
+                    billing_address, "billing", customer, save=False
+                )
+
+                if customer.default_billing_address:
+                    addresses_to_update.append(billing_address)
+                else:
+                    customer.default_billing_address = billing_address
+                    addresses_to_create.append(billing_address)
+                    customer_instance_new_addresses_map[customer].append(
+                        shipping_address
+                    )
+
+        models.Address.objects.bulk_create(addresses_to_create)
+        models.Address.objects.bulk_update(
+            addresses_to_update,
+            fields=[
+                "first_name",
+                "last_name",
+                "company_name",
+                "street_address_1",
+                "street_address_2",
+                "city",
+                "city_area",
+                "postal_code",
+                "country",
+                "country_area",
+                "phone",
+            ],
+        )
+
+        models.User.objects.bulk_update(
+            customers_to_update,
+            fields=[
+                "first_name",
+                "last_name",
+                "email",
+                "is_active",
+                "note",
+                "language_code",
+                "external_reference",
+                "updated_at",
+            ],
+        )
+
+        for customer in customers_to_update:
+            if customer in customer_instance_new_addresses_map:
+                customer.addresses.add(*customer_instance_new_addresses_map[customer])
+                # refresh customer default addresses
+                customer.default_billing_address = customer.default_billing_address
+                customer.default_shipping_address = customer.default_shipping_address
+
+            search_document = prepare_user_search_document_value(customer)
+            customer.search_document = search_document
+
+        models.User.objects.bulk_update(
+            customers_to_update,
+            fields=[
+                "default_shipping_address",
+                "default_billing_address",
+                "search_document",
+            ],
+        )
+
+        return customers_to_update, old_instances
+
+    @classmethod
+    def post_save_actions(cls, info, manager, instances, old_instances):
+        customer_events = []
+        app = get_app_promise(info.context).get()
+        staff_user = info.context.user
+
+        for updated_instance, old_instance in zip(instances, old_instances):
+            cls.call_event(manager.customer_updated, updated_instance)
+            new_email = updated_instance.email
+            new_fullname = updated_instance.get_full_name()
+
+            # Compare the data
+            has_new_name = old_instance.get_full_name() != new_fullname
+            has_new_email = old_instance.email != new_email
+            was_activated = not old_instance.is_active and updated_instance.is_active
+            was_deactivated = old_instance.is_active and not updated_instance.is_active
+
+            # Generate the events accordingly
+            if has_new_email:
+                customer_events.append(
+                    models.CustomerEvent(
+                        user=staff_user,
+                        app=app,
+                        order=None,
+                        type=CustomerEvents.EMAIL_ASSIGNED,
+                        parameters={"message": new_email},
+                    )
+                )
+                assign_user_gift_cards(updated_instance)
+                match_orders_with_new_user(updated_instance)
+
+            if has_new_name:
+                customer_events.append(
+                    models.CustomerEvent(
+                        user=staff_user,
+                        app=app,
+                        order=None,
+                        type=CustomerEvents.NAME_ASSIGNED,
+                        parameters={"message": new_fullname},
+                    )
+                )
+            if was_activated:
+                customer_events.append(
+                    models.CustomerEvent(
+                        user=staff_user,
+                        app=app,
+                        type=CustomerEvents.ACCOUNT_ACTIVATED,
+                        parameters={"account_id": updated_instance.id},
+                    )
+                )
+            if was_deactivated:
+                customer_events.append(
+                    models.CustomerEvent(
+                        user=staff_user,
+                        app=app,
+                        type=CustomerEvents.ACCOUNT_DEACTIVATED,
+                        parameters={"account_id": updated_instance.id},
+                    )
+                )
+
+        models.CustomerEvent.objects.bulk_create(customer_events)
+
+    @classmethod
+    def get_results(cls, instances_data_with_errors_list, reject_everything=False):
+        if reject_everything:
+            return [
+                CustomerBulkResult(customer=None, errors=data.get("errors"))
+                for data in instances_data_with_errors_list
+            ]
+        return [
+            CustomerBulkResult(customer=data.get("instance"), errors=data.get("errors"))
+            if data.get("instance")
+            else CustomerBulkResult(customer=None, errors=data.get("errors"))
+            for data in instances_data_with_errors_list
+        ]
+
+    @classmethod
+    @traced_atomic_transaction()
+    def perform_mutation(cls, _root, info, **data):
+        error_policy = data.get("error_policy", ErrorPolicyEnum.REJECT_EVERYTHING.value)
+        index_error_map: dict = defaultdict(list)
+
+        cleaned_inputs_map = cls.clean_customers(
+            info, data["customers"], index_error_map
+        )
+
+        instances_data_with_errors_list = cls.update_customers(
+            info, cleaned_inputs_map, index_error_map
+        )
+
+        if any([bool(error) for error in index_error_map.values()]):
+            if error_policy == ErrorPolicyEnum.REJECT_EVERYTHING.value:
+                results = cls.get_results(instances_data_with_errors_list, True)
+                return CustomerBulkUpdate(count=0, results=results)
+
+            if error_policy == ErrorPolicyEnum.REJECT_FAILED_ROWS.value:
+                for data in instances_data_with_errors_list:
+                    if data["errors"] and data["instance"]:
+                        data["instance"] = None
+
+        manager = get_plugin_manager_promise(info.context).get()
+
+        updated_customers, old_instances = cls.save_customers(
+            instances_data_with_errors_list, manager
+        )
+
+        # prepare and return data
+        results = cls.get_results(instances_data_with_errors_list)
+
+        cls.post_save_actions(info, manager, updated_customers, old_instances)
+
+        return CustomerBulkUpdate(count=len(updated_customers), results=results)
diff --git a/saleor/graphql/account/bulk_mutations/staff_bulk_delete.py b/saleor/graphql/account/bulk_mutations/staff_bulk_delete.py
new file mode 100644
index 00000000000..0df418c69c5
--- /dev/null
+++ b/saleor/graphql/account/bulk_mutations/staff_bulk_delete.py
@@ -0,0 +1,62 @@
+from collections import defaultdict
+from typing import List
+
+from django.core.exceptions import ValidationError
+
+from ....account import models
+from ....permission.enums import AccountPermissions
+from ...core import ResolveInfo
+from ...core.doc_category import DOC_CATEGORY_USERS
+from ...core.types import StaffError
+from ...plugins.dataloaders import get_plugin_manager_promise
+from ..types import User
+from ..utils import StaffDeleteMixin
+from .customer_bulk_delete import UserBulkDelete
+
+
+class StaffBulkDelete(StaffDeleteMixin, UserBulkDelete):
+    class Meta:
+        description = (
+            "Deletes staff users. Apps are not allowed to perform this mutation."
+        )
+        doc_category = DOC_CATEGORY_USERS
+        model = models.User
+        object_type = User
+        permissions = (AccountPermissions.MANAGE_STAFF,)
+        error_type_class = StaffError
+        error_type_field = "staff_errors"
+
+    @classmethod
+    def perform_mutation(  # type: ignore[override]
+        cls, _root, info: ResolveInfo, /, *, ids, **data
+    ):
+        instances = cls.get_nodes_or_error(ids, "id", User)
+        errors = cls.clean_instances(info, instances)
+        count = len(instances)
+        if not errors and count:
+            clean_instance_ids = [instance.pk for instance in instances]
+            qs = models.User.objects.filter(pk__in=clean_instance_ids, is_staff=True)
+            cls.bulk_action(info, qs, **data)
+        else:
+            count = 0
+        return count, errors
+
+    @classmethod
+    def clean_instances(cls, info: ResolveInfo, users):
+        errors: defaultdict[str, List[ValidationError]] = defaultdict(list)
+
+        requestor = info.context.user
+        cls.check_if_users_can_be_deleted(info, users, "ids", errors)
+        cls.check_if_requestor_can_manage_users(requestor, users, "ids", errors)
+        cls.check_if_removing_left_not_manageable_permissions(
+            requestor, users, "ids", errors
+        )
+        return ValidationError(errors) if errors else {}
+
+    @classmethod
+    def bulk_action(cls, info: ResolveInfo, queryset, /):
+        instances = list(queryset)
+        queryset.delete()
+        manager = get_plugin_manager_promise(info.context).get()
+        for instance in instances:
+            manager.staff_deleted(instance)
diff --git a/saleor/graphql/account/bulk_mutations/user_bulk_set_active.py b/saleor/graphql/account/bulk_mutations/user_bulk_set_active.py
new file mode 100644
index 00000000000..0f5d53e2ced
--- /dev/null
+++ b/saleor/graphql/account/bulk_mutations/user_bulk_set_active.py
@@ -0,0 +1,59 @@
+import graphene
+from django.core.exceptions import ValidationError
+
+from ....account import models
+from ....account.error_codes import AccountErrorCode
+from ....permission.enums import AccountPermissions
+from ...core import ResolveInfo
+from ...core.doc_category import DOC_CATEGORY_USERS
+from ...core.mutations import BaseBulkMutation
+from ...core.types import AccountError, NonNullList
+from ..types import User
+
+
+class UserBulkSetActive(BaseBulkMutation):
+    class Arguments:
+        ids = NonNullList(
+            graphene.ID,
+            required=True,
+            description="List of user IDs to activate/deactivate.",
+        )
+        is_active = graphene.Boolean(
+            required=True, description="Determine if users will be set active or not."
+        )
+
+    class Meta:
+        description = "Activate or deactivate users."
+        doc_category = DOC_CATEGORY_USERS
+        model = models.User
+        object_type = User
+        permissions = (AccountPermissions.MANAGE_USERS,)
+        error_type_class = AccountError
+        error_type_field = "account_errors"
+
+    @classmethod
+    def clean_instance(cls, info: ResolveInfo, instance):
+        if info.context.user == instance:
+            raise ValidationError(
+                {
+                    "is_active": ValidationError(
+                        "Cannot activate or deactivate your own account.",
+                        code=AccountErrorCode.ACTIVATE_OWN_ACCOUNT.value,
+                    )
+                }
+            )
+        elif instance.is_superuser:
+            raise ValidationError(
+                {
+                    "is_active": ValidationError(
+                        "Cannot activate or deactivate superuser's account.",
+                        code=AccountErrorCode.ACTIVATE_SUPERUSER_ACCOUNT.value,
+                    )
+                }
+            )
+
+    @classmethod
+    def bulk_action(  # type: ignore[override]
+        cls, _info: ResolveInfo, queryset, /, *, is_active
+    ):
+        queryset.update(is_active=is_active)
diff --git a/saleor/graphql/account/schema.py b/saleor/graphql/account/schema.py
index d79d8a32b8a..6b8429efbd6 100644
--- a/saleor/graphql/account/schema.py
+++ b/saleor/graphql/account/schema.py
@@ -10,7 +10,12 @@
 from ..core.types import FilterInputObjectType
 from ..core.utils import from_global_id_or_error
 from ..core.validators import validate_one_of_args_is_in_query
-from .bulk_mutations import CustomerBulkDelete, StaffBulkDelete, UserBulkSetActive
+from .bulk_mutations import (
+    CustomerBulkDelete,
+    CustomerBulkUpdate,
+    StaffBulkDelete,
+    UserBulkSetActive,
+)
 from .enums import CountryCodeEnum
 from .filters import CustomerFilter, PermissionGroupFilter, StaffUserFilter
 from .mutations.account import (
@@ -286,6 +291,7 @@ class AccountMutations(graphene.ObjectType):
     customer_update = CustomerUpdate.Field()
     customer_delete = CustomerDelete.Field()
     customer_bulk_delete = CustomerBulkDelete.Field()
+    customer_bulk_update = CustomerBulkUpdate.Field()
 
     staff_create = StaffCreate.Field()
     staff_update = StaffUpdate.Field()
diff --git a/saleor/graphql/account/tests/mutations/test_customer_bulk_update.py b/saleor/graphql/account/tests/mutations/test_customer_bulk_update.py
new file mode 100644
index 00000000000..712f254dbf2
--- /dev/null
+++ b/saleor/graphql/account/tests/mutations/test_customer_bulk_update.py
@@ -0,0 +1,645 @@
+from unittest.mock import patch
+
+import graphene
+
+from .....account import models
+from .....account.error_codes import CustomerBulkUpdateErrorCode
+from .....account.events import CustomerEvents
+from .....account.search import generate_address_search_document_value
+from ....tests.utils import get_graphql_content
+from ...tests.utils import convert_dict_keys_to_camel_case
+
+CUSTOMER_BULK_UPDATE_MUTATION = """
+    mutation CustomerBulkUpdate(
+        $customers: [CustomerBulkUpdateInput!]!,
+        $errorPolicy: ErrorPolicyEnum
+    ){
+        customerBulkUpdate(customers: $customers, errorPolicy: $errorPolicy){
+            results{
+                errors {
+                    path
+                    message
+                    code
+                }
+                customer{
+                    id
+                    firstName
+                }
+            }
+            count
+        }
+    }
+"""
+
+
+def test_customers_bulk_update_using_ids(
+    staff_api_client,
+    customer_users,
+    permission_manage_users,
+):
+    # given
+    customer_1 = customer_users[0]
+    customer_2 = customer_users[1]
+
+    customer_1_id = graphene.Node.to_global_id("User", customer_1.pk)
+    customer_2_id = graphene.Node.to_global_id("User", customer_2.pk)
+
+    customer_1_new_name = "NewName1"
+    customer_2_new_name = "NewName2"
+
+    customers_input = [
+        {
+            "id": customer_1_id,
+            "input": {
+                "firstName": customer_1_new_name,
+            },
+        },
+        {
+            "id": customer_2_id,
+            "input": {
+                "firstName": customer_2_new_name,
+            },
+        },
+    ]
+
+    variables = {"customers": customers_input}
+
+    # when
+    staff_api_client.user.user_permissions.add(permission_manage_users)
+    response = staff_api_client.post_graphql(CUSTOMER_BULK_UPDATE_MUTATION, variables)
+    content = get_graphql_content(response)
+    data = content["data"]["customerBulkUpdate"]
+
+    # then
+    customer_1.refresh_from_db()
+    customer_2.refresh_from_db()
+    assert not data["results"][0]["errors"]
+    assert not data["results"][1]["errors"]
+    assert data["count"] == 2
+    assert customer_1.first_name == customer_1_new_name
+    assert customer_2.first_name == customer_2_new_name
+
+
+@patch("saleor.plugins.manager.PluginsManager.customer_updated")
+def test_stocks_bulk_update_send_stock_updated_event(
+    customer_updated_webhook,
+    staff_api_client,
+    customer_users,
+    permission_manage_users,
+):
+    # given
+    customer_1 = customer_users[0]
+    customer_2 = customer_users[1]
+
+    customer_1_id = graphene.Node.to_global_id("User", customer_1.pk)
+    customer_2_id = graphene.Node.to_global_id("User", customer_2.pk)
+
+    customer_1_new_name = "NewName1"
+    customer_2_new_name = "NewName2"
+
+    customers_input = [
+        {
+            "id": customer_1_id,
+            "input": {
+                "firstName": customer_1_new_name,
+            },
+        },
+        {
+            "id": customer_2_id,
+            "input": {
+                "firstName": customer_2_new_name,
+            },
+        },
+    ]
+
+    variables = {"customers": customers_input}
+
+    # when
+    staff_api_client.user.user_permissions.add(permission_manage_users)
+    response = staff_api_client.post_graphql(CUSTOMER_BULK_UPDATE_MUTATION, variables)
+    content = get_graphql_content(response)
+    data = content["data"]["customerBulkUpdate"]
+
+    # then
+    assert not data["results"][0]["errors"]
+    assert not data["results"][1]["errors"]
+    assert data["count"] == 2
+    assert customer_updated_webhook.call_count == 2
+
+
+def test_customers_bulk_update_generate_events_when_deactivating(
+    staff_api_client,
+    staff_user,
+    customer_users,
+    permission_manage_users,
+):
+    # given
+    customer_1 = customer_users[0]
+    customer_2 = customer_users[1]
+
+    customer_1_id = graphene.Node.to_global_id("User", customer_1.pk)
+    customer_2_id = graphene.Node.to_global_id("User", customer_2.pk)
+
+    customers_input = [
+        {
+            "id": customer_1_id,
+            "input": {
+                "isActive": False,
+            },
+        },
+        {
+            "id": customer_2_id,
+            "input": {
+                "isActive": False,
+            },
+        },
+    ]
+
+    variables = {"customers": customers_input}
+
+    # when
+    staff_api_client.user.user_permissions.add(permission_manage_users)
+    response = staff_api_client.post_graphql(CUSTOMER_BULK_UPDATE_MUTATION, variables)
+    content = get_graphql_content(response)
+    data = content["data"]["customerBulkUpdate"]
+
+    # then
+    account_deactivated_events = models.CustomerEvent.objects.all()
+    assert not data["results"][0]["errors"]
+    assert not data["results"][1]["errors"]
+    assert data["count"] == 2
+
+    assert len(account_deactivated_events) == 2
+    for event in account_deactivated_events:
+        assert event.type == CustomerEvents.ACCOUNT_DEACTIVATED
+        assert event.user.pk == staff_user.pk
+
+
+def test_customers_bulk_update_generate_events_when_name_change(
+    staff_api_client,
+    staff_user,
+    customer_users,
+    permission_manage_users,
+):
+    # given
+    customer_1 = customer_users[0]
+    customer_2 = customer_users[1]
+
+    customer_1_id = graphene.Node.to_global_id("User", customer_1.pk)
+    customer_2_id = graphene.Node.to_global_id("User", customer_2.pk)
+
+    customer_1_new_name = "NewName1"
+    customer_2_new_name = "NewName2"
+
+    customers_input = [
+        {
+            "id": customer_1_id,
+            "input": {
+                "firstName": customer_1_new_name,
+            },
+        },
+        {
+            "id": customer_2_id,
+            "input": {
+                "firstName": customer_2_new_name,
+            },
+        },
+    ]
+
+    variables = {"customers": customers_input}
+
+    # when
+    staff_api_client.user.user_permissions.add(permission_manage_users)
+    response = staff_api_client.post_graphql(CUSTOMER_BULK_UPDATE_MUTATION, variables)
+    content = get_graphql_content(response)
+    data = content["data"]["customerBulkUpdate"]
+
+    # then
+    name_changed_events = models.CustomerEvent.objects.all()
+    assert not data["results"][0]["errors"]
+    assert not data["results"][1]["errors"]
+    assert data["count"] == 2
+
+    assert len(name_changed_events) == 2
+    for event in name_changed_events:
+        assert event.type == CustomerEvents.NAME_ASSIGNED
+        assert event.user.pk == staff_user.pk
+
+
+def test_customers_bulk_update_generate_events_when_email_change(
+    staff_api_client,
+    staff_user,
+    gift_card,
+    order,
+    customer_user,
+    permission_manage_users,
+):
+    # give
+
+    customer_id = graphene.Node.to_global_id("User", customer_user.pk)
+    customer_new_email = "newemail1@example.com"
+
+    gift_card.created_by = None
+    gift_card.created_by_email = customer_new_email
+    gift_card.save(update_fields=["created_by", "created_by_email"])
+
+    order.user = None
+    order.user_email = customer_new_email
+    order.save(update_fields=["user_email", "user"])
+
+    customers_input = [
+        {
+            "id": customer_id,
+            "input": {
+                "email": customer_new_email,
+            },
+        }
+    ]
+
+    variables = {"customers": customers_input}
+
+    # when
+    staff_api_client.user.user_permissions.add(permission_manage_users)
+    response = staff_api_client.post_graphql(CUSTOMER_BULK_UPDATE_MUTATION, variables)
+    content = get_graphql_content(response)
+    data = content["data"]["customerBulkUpdate"]
+
+    # then
+    gift_card.refresh_from_db()
+    order.refresh_from_db()
+    customer_user.refresh_from_db()
+    email_changed_event = models.CustomerEvent.objects.get()
+    assert not data["results"][0]["errors"]
+    assert data["count"] == 1
+    assert email_changed_event.type == CustomerEvents.EMAIL_ASSIGNED
+    assert email_changed_event.user.pk == staff_user.pk
+    assert gift_card.created_by == customer_user
+    assert gift_card.created_by_email == customer_user.email
+    assert order.user == customer_user
+
+
+def test_customers_bulk_update_using_external_refs(
+    staff_api_client,
+    customer_users,
+    permission_manage_users,
+):
+    # given
+    customer_1 = customer_users[0]
+    customer_2 = customer_users[1]
+
+    customer_1_new_name = "NewName1"
+    customer_2_new_name = "NewName2"
+
+    customers_input = [
+        {
+            "externalReference": customer_1.external_reference,
+            "input": {
+                "firstName": customer_1_new_name,
+            },
+        },
+        {
+            "externalReference": customer_2.external_reference,
+            "input": {
+                "firstName": customer_2_new_name,
+            },
+        },
+    ]
+
+    variables = {"customers": customers_input}
+
+    # when
+    staff_api_client.user.user_permissions.add(permission_manage_users)
+    response = staff_api_client.post_graphql(CUSTOMER_BULK_UPDATE_MUTATION, variables)
+    content = get_graphql_content(response)
+    data = content["data"]["customerBulkUpdate"]
+
+    # then
+    customer_1.refresh_from_db()
+    customer_2.refresh_from_db()
+    assert not data["results"][0]["errors"]
+    assert not data["results"][1]["errors"]
+    assert data["count"] == 2
+    assert customer_1.first_name == customer_1_new_name
+    assert customer_2.first_name == customer_2_new_name
+
+
+def test_customers_bulk_update_when_no_id_or_external_ref_provided(
+    staff_api_client,
+    permission_manage_users,
+):
+    # given
+    customers_input = [
+        {
+            "input": {
+                "firstName": "NewName",
+            }
+        }
+    ]
+
+    variables = {"customers": customers_input}
+
+    # when
+    staff_api_client.user.user_permissions.add(permission_manage_users)
+    response = staff_api_client.post_graphql(CUSTOMER_BULK_UPDATE_MUTATION, variables)
+    content = get_graphql_content(response)
+    data = content["data"]["customerBulkUpdate"]
+
+    # then
+    assert data["results"][0]["errors"]
+    error = data["results"][0]["errors"][0]
+    assert error["message"] == (
+        "At least one of arguments is required: 'id', " "'externalReference'."
+    )
+
+
+def test_customers_bulk_update_when_invalid_id_provided(
+    staff_api_client,
+    permission_manage_users,
+):
+    # given
+    customers_input = [
+        {
+            "id": "WrongID",
+            "input": {
+                "firstName": "NewName",
+            },
+        }
+    ]
+
+    variables = {"customers": customers_input}
+
+    # when
+    staff_api_client.user.user_permissions.add(permission_manage_users)
+    response = staff_api_client.post_graphql(CUSTOMER_BULK_UPDATE_MUTATION, variables)
+    content = get_graphql_content(response)
+    data = content["data"]["customerBulkUpdate"]
+
+    # then
+    assert data["count"] == 0
+    assert data["results"][0]["errors"]
+    error = data["results"][0]["errors"][0]
+    assert error["message"] == "Invalid customer ID."
+
+
+def test_customers_bulk_update_when_customer_not_exists(
+    staff_api_client,
+    permission_manage_users,
+):
+    # given
+    customers_input = [
+        {
+            "externalReference": "WrongRef",
+            "input": {
+                "firstName": "NewName",
+            },
+        }
+    ]
+
+    variables = {"customers": customers_input}
+
+    # when
+    staff_api_client.user.user_permissions.add(permission_manage_users)
+    response = staff_api_client.post_graphql(CUSTOMER_BULK_UPDATE_MUTATION, variables)
+    content = get_graphql_content(response)
+    data = content["data"]["customerBulkUpdate"]
+
+    # then
+    assert data["count"] == 0
+    assert data["results"][0]["errors"]
+    error = data["results"][0]["errors"][0]
+    assert error["code"] == CustomerBulkUpdateErrorCode.NOT_FOUND.name
+    assert error["message"] == "Customer was not found."
+
+
+def test_customers_bulk_update_correct_fields_validation(
+    staff_api_client,
+    customer_user,
+    permission_manage_users,
+):
+    # given
+    customer_id = graphene.Node.to_global_id("User", customer_user.pk)
+
+    new_name = 50 * "NewName1"
+    new_last_name = 30 * "NewLastName2"
+
+    customers_input = [
+        {"id": customer_id, "input": {"firstName": new_name, "lastName": new_last_name}}
+    ]
+    variables = {"customers": customers_input}
+
+    # when
+    staff_api_client.user.user_permissions.add(permission_manage_users)
+    response = staff_api_client.post_graphql(CUSTOMER_BULK_UPDATE_MUTATION, variables)
+    content = get_graphql_content(response)
+    data = content["data"]["customerBulkUpdate"]
+
+    # then
+    errors = data["results"][0]["errors"]
+    assert errors
+    assert errors[0]["code"] == CustomerBulkUpdateErrorCode.MAX_LENGTH.name
+    assert errors[0]["path"] == "input.firstName"
+    assert errors[1]["code"] == CustomerBulkUpdateErrorCode.MAX_LENGTH.name
+    assert errors[1]["path"] == "input.lastName"
+    assert data["count"] == 0
+
+
+def test_customers_bulk_update_with_address(
+    staff_api_client,
+    customer_user,
+    address,
+    permission_manage_users,
+):
+    # given
+    shipping_address, billing_address = (
+        customer_user.default_shipping_address,
+        customer_user.default_billing_address,
+    )
+
+    assert shipping_address
+    assert billing_address
+
+    customer_id = graphene.Node.to_global_id("User", customer_user.pk)
+
+    address_data = convert_dict_keys_to_camel_case(address.as_data())
+    address_data.pop("metadata")
+    address_data.pop("privateMetadata")
+
+    new_street_address = "Updated street address"
+    address_data["streetAddress1"] = new_street_address
+
+    customers_input = [
+        {
+            "id": customer_id,
+            "input": {
+                "defaultBillingAddress": address_data,
+                "defaultShippingAddress": address_data,
+            },
+        }
+    ]
+
+    variables = {"customers": customers_input}
+
+    # when
+    staff_api_client.user.user_permissions.add(permission_manage_users)
+    response = staff_api_client.post_graphql(CUSTOMER_BULK_UPDATE_MUTATION, variables)
+    content = get_graphql_content(response)
+    data = content["data"]["customerBulkUpdate"]
+
+    # then
+
+    shipping_address.refresh_from_db()
+    billing_address.refresh_from_db()
+    customer_user.refresh_from_db()
+
+    assert not data["results"][0]["errors"]
+    assert data["count"] == 1
+    assert billing_address.street_address_1 == new_street_address
+    assert shipping_address.street_address_1 == new_street_address
+    assert (
+        generate_address_search_document_value(billing_address)
+        in customer_user.search_document
+    )
+    assert (
+        generate_address_search_document_value(shipping_address)
+        in customer_user.search_document
+    )
+
+
+def test_customers_bulk_update_with_address_when_no_default(
+    staff_api_client,
+    customer_user,
+    address,
+    permission_manage_users,
+):
+    # given
+    shipping_address = customer_user.default_shipping_address
+
+    customer_user.default_shipping_address = None
+    customer_user.save(update_fields=["default_shipping_address"])
+
+    assert shipping_address
+
+    customer_id = graphene.Node.to_global_id("User", customer_user.pk)
+
+    address_data = convert_dict_keys_to_camel_case(shipping_address.as_data())
+    address_data.pop("metadata")
+    address_data.pop("privateMetadata")
+
+    new_street_address = "Updated street address"
+    address_data["streetAddress1"] = new_street_address
+
+    customers_input = [
+        {
+            "id": customer_id,
+            "input": {
+                "defaultShippingAddress": address_data,
+            },
+        }
+    ]
+
+    variables = {"customers": customers_input}
+
+    # when
+    staff_api_client.user.user_permissions.add(permission_manage_users)
+    response = staff_api_client.post_graphql(CUSTOMER_BULK_UPDATE_MUTATION, variables)
+    content = get_graphql_content(response)
+    data = content["data"]["customerBulkUpdate"]
+
+    # then
+    customer_user.refresh_from_db()
+    assert not data["results"][0]["errors"]
+    assert data["count"] == 1
+
+    assert customer_user.default_shipping_address
+    assert customer_user.default_shipping_address in customer_user.addresses.all()
+
+
+def test_customers_bulk_update_with_invalid_address(
+    staff_api_client,
+    customer_user,
+    address,
+    permission_manage_users,
+):
+    # given
+    shipping_address, billing_address = (
+        customer_user.default_shipping_address,
+        customer_user.default_billing_address,
+    )
+
+    assert shipping_address
+    assert billing_address
+
+    customer_id = graphene.Node.to_global_id("User", customer_user.pk)
+
+    address_data = convert_dict_keys_to_camel_case(address.as_data())
+    address_data.pop("metadata")
+    address_data.pop("privateMetadata")
+    address_data.pop("country")
+
+    new_street_address = "Updated street address"
+    address_data["streetAddress1"] = new_street_address
+
+    customers_input = [
+        {
+            "id": customer_id,
+            "input": {
+                "defaultBillingAddress": address_data,
+            },
+        }
+    ]
+
+    variables = {"customers": customers_input}
+
+    # when
+    staff_api_client.user.user_permissions.add(permission_manage_users)
+    response = staff_api_client.post_graphql(CUSTOMER_BULK_UPDATE_MUTATION, variables)
+    content = get_graphql_content(response)
+    data = content["data"]["customerBulkUpdate"]
+
+    # then
+    assert data["results"][0]["errors"]
+    assert data["count"] == 0
+    error = data["results"][0]["errors"][0]
+    assert error["code"] == CustomerBulkUpdateErrorCode.REQUIRED.name
+    assert error["path"] == "input.defaultBillingAddress.country"
+
+
+def test_customers_bulk_update_with_duplicated_external_ref(
+    staff_api_client,
+    customer_users,
+    permission_manage_users,
+):
+    # given
+    customer_1 = customer_users[0]
+    customer_2 = customer_users[1]
+
+    customer_1_id = graphene.Node.to_global_id("User", customer_1.pk)
+    customer_2_id = graphene.Node.to_global_id("User", customer_2.pk)
+
+    external_ref = "sameRef"
+
+    customers_input = [
+        {"id": customer_1_id, "input": {"externalReference": external_ref}},
+        {"id": customer_2_id, "input": {"externalReference": external_ref}},
+    ]
+
+    variables = {"customers": customers_input}
+
+    # when
+    staff_api_client.user.user_permissions.add(permission_manage_users)
+    response = staff_api_client.post_graphql(CUSTOMER_BULK_UPDATE_MUTATION, variables)
+    content = get_graphql_content(response)
+    data = content["data"]["customerBulkUpdate"]
+
+    # then
+    assert data["count"] == 0
+    assert data["results"][0]["errors"]
+    assert data["results"][1]["errors"]
+    assert (
+        data["results"][0]["errors"][0]["code"]
+        == CustomerBulkUpdateErrorCode.DUPLICATED_INPUT_ITEM.name
+    )
+    assert (
+        data["results"][1]["errors"][0]["code"]
+        == CustomerBulkUpdateErrorCode.DUPLICATED_INPUT_ITEM.name
+    )
diff --git a/saleor/graphql/core/enums.py b/saleor/graphql/core/enums.py
index 63fc90d8cb9..241e5091609 100644
--- a/saleor/graphql/core/enums.py
+++ b/saleor/graphql/core/enums.py
@@ -134,6 +134,9 @@ def error_policy_enum_description(enum):
 AttributeErrorCode = graphene.Enum.from_enum(attribute_error_codes.AttributeErrorCode)
 ChannelErrorCode = graphene.Enum.from_enum(channel_error_codes.ChannelErrorCode)
 CheckoutErrorCode = graphene.Enum.from_enum(checkout_error_codes.CheckoutErrorCode)
+CustomerBulkUpdateErrorCode = graphene.Enum.from_enum(
+    account_error_codes.CustomerBulkUpdateErrorCode
+)
 
 ExternalNotificationTriggerErrorCode = graphene.Enum.from_enum(
     external_notifications_error_codes.ExternalNotificationErrorCodes
diff --git a/saleor/graphql/core/types/__init__.py b/saleor/graphql/core/types/__init__.py
index 3cd18de5327..4e7d2e066ff 100644
--- a/saleor/graphql/core/types/__init__.py
+++ b/saleor/graphql/core/types/__init__.py
@@ -12,6 +12,7 @@
     CollectionChannelListingError,
     CollectionError,
     CountryDisplay,
+    CustomerBulkUpdateError,
     DateRangeInput,
     DateTimeRangeInput,
     DiscountError,
@@ -91,6 +92,7 @@
     "CollectionChannelListingError",
     "CollectionError",
     "CountryDisplay",
+    "CustomerBulkUpdateError",
     "DateRangeInput",
     "DateTimeRangeInput",
     "DiscountError",
diff --git a/saleor/graphql/core/types/common.py b/saleor/graphql/core/types/common.py
index 989c4ee1e37..d34df0aacbd 100644
--- a/saleor/graphql/core/types/common.py
+++ b/saleor/graphql/core/types/common.py
@@ -20,6 +20,7 @@
     ChannelErrorCode,
     CheckoutErrorCode,
     CollectionErrorCode,
+    CustomerBulkUpdateErrorCode,
     DiscountErrorCode,
     ExportErrorCode,
     ExternalNotificationTriggerErrorCode,
@@ -122,6 +123,20 @@ class Meta:
         description = "Represents an error in the input of a mutation."
 
 
+class BulkError(graphene.ObjectType):
+    path = graphene.String(
+        description=(
+            "Path to field that caused the error. A value of `null` indicates that "
+            "the error isn't associated with a particular field."
+        ),
+        required=False,
+    )
+    message = graphene.String(description="The error message.")
+
+    class Meta:
+        description = "Represents an error in the input of a mutation."
+
+
 class AccountError(Error):
     code = AccountErrorCode(description="The error code.", required=True)
     address_type = AddressTypeEnum(  # type: ignore[has-type]
@@ -195,6 +210,10 @@ class CheckoutError(Error):
     )
 
 
+class CustomerBulkUpdateError(BulkError):
+    code = CustomerBulkUpdateErrorCode(description="The error code.", required=True)
+
+
 class ProductWithoutVariantError(Error):
     products = NonNullList(
         graphene.ID,
diff --git a/saleor/graphql/product/bulk_mutations/product_variant_bulk_update.py b/saleor/graphql/product/bulk_mutations/product_variant_bulk_update.py
index 068efd515bc..48905a10f94 100644
--- a/saleor/graphql/product/bulk_mutations/product_variant_bulk_update.py
+++ b/saleor/graphql/product/bulk_mutations/product_variant_bulk_update.py
@@ -663,7 +663,7 @@ def perform_mutation(cls, _root, info, **data):
         )
 
         # check error policy
-        if any([True if error else False for error in index_error_map.values()]):
+        if any([bool(error) for error in index_error_map.values()]):
             if error_policy == ErrorPolicyEnum.REJECT_EVERYTHING.value:
                 results = get_results(instances_data_with_errors_list, True)
                 return ProductVariantBulkUpdate(count=0, results=results)
diff --git a/saleor/graphql/schema.graphql b/saleor/graphql/schema.graphql
index d368d701810..137137e7b77 100644
--- a/saleor/graphql/schema.graphql
+++ b/saleor/graphql/schema.graphql
@@ -15987,6 +15987,23 @@ type Mutation {
     ids: [ID!]!
   ): CustomerBulkDelete @doc(category: "Users")
 
+  """
+  Updates customers.
+  
+  Added in Saleor 3.13.
+  
+  Note: this API is currently in Feature Preview and can be subject to changes at later point. 
+  
+  Requires one of the following permissions: MANAGE_USERS.
+  """
+  customerBulkUpdate(
+    """Input list of customers to update."""
+    customers: [CustomerBulkUpdateInput!]!
+
+    """Policies of error handling. DEFAULT: REJECT_EVERYTHING"""
+    errorPolicy: ErrorPolicyEnum
+  ): CustomerBulkUpdate @doc(category: "Users")
+
   """
   Creates a new staff user. Apps are not allowed to perform this mutation. 
   
@@ -16053,7 +16070,7 @@ type Mutation {
   Requires one of the following permissions: MANAGE_USERS.
   """
   userBulkSetActive(
-    """List of user IDs to (de)activate)."""
+    """List of user IDs to activate/deactivate."""
     ids: [ID!]!
 
     """Determine if users will be set active or not."""
@@ -24165,6 +24182,68 @@ type CustomerBulkDelete @doc(category: "Users") {
   errors: [AccountError!]!
 }
 
+"""
+Updates customers.
+
+Added in Saleor 3.13.
+
+Note: this API is currently in Feature Preview and can be subject to changes at later point. 
+
+Requires one of the following permissions: MANAGE_USERS.
+"""
+type CustomerBulkUpdate @doc(category: "Users") {
+  """Returns how many objects were created."""
+  count: Int!
+
+  """List of the updated customers."""
+  results: [CustomerBulkResult!]!
+  errors: [CustomerBulkUpdateError!]!
+}
+
+type CustomerBulkResult @doc(category: "Users") {
+  """Customer data."""
+  customer: User
+
+  """List of errors that occurred during the update attempt."""
+  errors: [CustomerBulkUpdateError!]
+}
+
+type CustomerBulkUpdateError {
+  """
+  Path to field that caused the error. A value of `null` indicates that the error isn't associated with a particular field.
+  """
+  path: String
+
+  """The error message."""
+  message: String
+
+  """The error code."""
+  code: CustomerBulkUpdateErrorCode!
+}
+
+"""An enumeration."""
+enum CustomerBulkUpdateErrorCode {
+  BLANK
+  DUPLICATED_INPUT_ITEM
+  GRAPHQL_ERROR
+  INVALID
+  REQUIRED
+  UNIQUE
+  NOT_FOUND
+  MAX_LENGTH
+}
+
+input CustomerBulkUpdateInput @doc(category: "Users") {
+  """ID of a customer to update."""
+  id: ID
+
+  """External ID of a customer to update."""
+  externalReference: String
+
+  """Fields required to update a customer."""
+  input: CustomerInput!
+}
+
 """
 Creates a new staff user. Apps are not allowed to perform this mutation. 
 
diff --git a/saleor/graphql/warehouse/bulk_mutations/stock_bulk_update.py b/saleor/graphql/warehouse/bulk_mutations/stock_bulk_update.py
index 6b6e2e5672d..fc8fab68cbd 100644
--- a/saleor/graphql/warehouse/bulk_mutations/stock_bulk_update.py
+++ b/saleor/graphql/warehouse/bulk_mutations/stock_bulk_update.py
@@ -383,7 +383,7 @@ def perform_mutation(cls, _root, info, **data):
             cleaned_inputs_map, index_error_map
         )
 
-        if any([True if error else False for error in index_error_map.values()]):
+        if any([bool(error) for error in index_error_map.values()]):
             if error_policy == ErrorPolicyEnum.REJECT_EVERYTHING.value:
                 results = cls.get_results(instances_data_with_errors_list, True)
                 return StockBulkUpdate(count=0, results=results)
diff --git a/saleor/plugins/anonymize/plugin.py b/saleor/plugins/anonymize/plugin.py
index 81a23478e33..0c3e6a29c41 100644
--- a/saleor/plugins/anonymize/plugin.py
+++ b/saleor/plugins/anonymize/plugin.py
@@ -40,11 +40,13 @@ def change_user_address(
         address_type: Optional[str],
         user: Optional["User"],
         previous_value: "Address",
+        save: bool = True,
     ) -> "Address":
         if address.phone:
             address.phone = ""  # type: ignore
         address = obfuscate_address(address)
-        address.save()
+        if save:
+            address.save()
         return address
 
     def order_created(self, order: "Order", previous_value: Any):
diff --git a/saleor/plugins/base_plugin.py b/saleor/plugins/base_plugin.py
index dca7d45ba66..144d08e874e 100644
--- a/saleor/plugins/base_plugin.py
+++ b/saleor/plugins/base_plugin.py
@@ -373,7 +373,7 @@ def __str__(self):
     channel_status_changed: Callable[["Channel", None], None]
 
     change_user_address: Callable[
-        ["Address", Union[str, None], Union["User", None], "Address"], "Address"
+        ["Address", Union[str, None], Union["User", None], "Address", bool], "Address"
     ]
 
     # Retrieves the balance remaining on a shopper's gift card
diff --git a/saleor/plugins/manager.py b/saleor/plugins/manager.py
index 35642c714f8..176ab59e2b6 100644
--- a/saleor/plugins/manager.py
+++ b/saleor/plugins/manager.py
@@ -216,11 +216,15 @@ def check_payment_balance(self, details: dict, channel_slug: str) -> dict:
         )
 
     def change_user_address(
-        self, address: "Address", address_type: Optional[str], user: Optional["User"]
+        self,
+        address: "Address",
+        address_type: Optional[str],
+        user: Optional["User"],
+        save: bool = True,
     ) -> "Address":
         default_value = address
         return self.__run_method_on_plugins(
-            "change_user_address", default_value, address, address_type, user
+            "change_user_address", default_value, address, address_type, user, save
         )
 
     def calculate_checkout_total(
diff --git a/saleor/tests/fixtures.py b/saleor/tests/fixtures.py
index 449b3656006..0034c9928b3 100644
--- a/saleor/tests/fixtures.py
+++ b/saleor/tests/fixtures.py
@@ -821,6 +821,7 @@ def customer_user(address):  # pylint: disable=W0613
         default_shipping_address=default_address,
         first_name="Leslie",
         last_name="Wade",
+        external_reference="LeslieWade",
         metadata={"key": "value"},
         private_metadata={"secret_key": "secret_value"},
     )
@@ -839,6 +840,7 @@ def customer_user2(address):
         default_shipping_address=default_address,
         first_name="Jane",
         last_name="Doe",
+        external_reference="JaneDoe",
     )
     user.addresses.add(default_address)
     user._password = "password"


Code review comments:
diff_hunk: @@ -0,0 +1,598 @@
+from collections import defaultdict
+from copy import copy
+
+import graphene
+from django.core.exceptions import ValidationError
+from django.db.models import Q
+from graphene.utils.str_converters import to_camel_case
+
+from ....account import models
+from ....account.events import CustomerEvents
+from ....account.search import prepare_user_search_document_value
+from ....checkout import AddressType
+from ....core.tracing import traced_atomic_transaction
+from ....giftcard.utils import assign_user_gift_cards
+from ....order.utils import match_orders_with_new_user
+from ....permission.enums import AccountPermissions
+from ...core.descriptions import ADDED_IN_313, PREVIEW_FEATURE
+from ...core.doc_category import DOC_CATEGORY_USERS
+from ...core.enums import CustomerBulkUpdateErrorCode, ErrorPolicyEnum
+from ...core.mutations import BaseMutation, ModelMutation
+from ...core.types import (
+    BaseInputObjectType,
+    BaseObjectType,
+    CustomerBulkUpdateError,
+    NonNullList,
+)
+from ...core.utils import get_duplicated_values
+from ...core.validators import validate_one_of_args_is_in_mutation
+from ...plugins.dataloaders import get_app_promise, get_plugin_manager_promise
+from ..i18n import I18nMixin
+from ..mutations.base import BILLING_ADDRESS_FIELD, SHIPPING_ADDRESS_FIELD
+from ..mutations.staff import CustomerInput
+from ..types import User
+
+
+class CustomerBulkResult(BaseObjectType):
+    customer = graphene.Field(User, required=False, description="Customer data.")
+    errors = NonNullList(
+        CustomerBulkUpdateError,
+        required=False,
+        description="List of errors that occurred during the update attempt.",
+    )
+
+    class Meta:
+        doc_category = DOC_CATEGORY_USERS
+
+
+class CustomerBulkUpdateInput(BaseInputObjectType):
+    id = graphene.ID(description="ID of a customer to update.", required=False)
+    external_reference = graphene.String(
+        required=False,
+        description="External ID of a customer to update.",
+    )
+    input = CustomerInput(
+        description="Fields required to update a customer.", required=True
+    )
+
+    class Meta:
+        doc_category = DOC_CATEGORY_USERS
+
+
+class CustomerBulkUpdate(BaseMutation, I18nMixin):
+    count = graphene.Int(
+        required=True,
+        default_value=0,
+        description="Returns how many objects were created.",
+    )
+    results = NonNullList(
+        CustomerBulkResult,
+        required=True,
+        default_value=[],
+        description="List of the updated customers.",
+    )
+
+    class Arguments:
+        customers = NonNullList(
+            CustomerBulkUpdateInput,
+            required=True,
+            description="Input list of customers to update.",
+        )
+        error_policy = ErrorPolicyEnum(
+            required=False,
+            description=(
+                "Policies of error handling. DEFAULT: "
+                + ErrorPolicyEnum.REJECT_EVERYTHING.name
+            ),
+        )
+
+    class Meta:
+        description = "Updates customers." + ADDED_IN_313 + PREVIEW_FEATURE
+        doc_category = DOC_CATEGORY_USERS
+        permissions = (AccountPermissions.MANAGE_USERS,)
+        error_type_class = CustomerBulkUpdateError
+
+    @classmethod
+    def format_errors(cls, index, errors, index_error_map, field_prefix=None):
+        for key, value in errors.error_dict.items():
+            for e in value:
+                path = (
+                    to_camel_case(f"{field_prefix}.{key}")
+                    if field_prefix
+                    else to_camel_case(key)
+                )
+                index_error_map[index].append(
+                    CustomerBulkUpdateError(
+                        path=f"input.{path}",
+                        message=e.messages[0],
+                        code=e.code,
+                    )
+                )
+
+    @classmethod
+    def validate_customer(
+        cls, customer_id, external_ref, cleaned_input, index, index_error_map
+    ):
+        errors_count = 0
+        try:
+            validate_one_of_args_is_in_mutation(
+                "id",
+                customer_id,
+                "external_reference",
+                external_ref,
+                use_camel_case=True,
+            )
+        except ValidationError as exc:
+            index_error_map[index].append(
+                CustomerBulkUpdateError(
+                    message=exc.message,
+                    code=CustomerBulkUpdateErrorCode.INVALID.value,
+                )
+            )
+            errors_count += 1
+
+        if customer_id:
+            try:
+                type, customer_id = graphene.Node.from_global_id(customer_id)
+                if type != "User":
+                    index_error_map[index].append(
+                        CustomerBulkUpdateError(
+                            path="id",
+                            message="Invalid customer ID.",
+                            code=CustomerBulkUpdateErrorCode.INVALID.value,
+                        )
+                    )
+                    errors_count += 1
+                else:
+                    cleaned_input["id"] = customer_id
+            except Exception:
+                index_error_map[index].append(
+                    CustomerBulkUpdateError(
+                        path="id",
+                        message="Invalid customer ID.",
+                        code=CustomerBulkUpdateErrorCode.INVALID.value,
+                    )
+                )
+                errors_count += 1
+        return errors_count
+
+    @classmethod
+    def clean_address(
+        cls,
+        address_data,
+        address_type,
+        field,
+        index,
+        index_error_map,
+        format_check=True,
+        required_check=True,
+        enable_normalization=True,
+    ):
+        try:
+            address_form = cls.validate_address_form(
+                address_data,
+                address_type,
+                format_check=format_check,
+                required_check=required_check,
+                enable_normalization=enable_normalization,
+            )
+
+            return address_form.cleaned_data
+        except ValidationError as exc:
+            cls.format_errors(index, exc, index_error_map, field_prefix=field)
+
+    @classmethod
+    def clean_customers(cls, info, customers_input, index_error_map):
+        cleaned_inputs_map: dict = {}
+
+        external_refs = [
+            customer_input["input"]["external_reference"]
+            for customer_input in customers_input
+            if customer_input["input"].get("external_reference")
+        ]
+
+        emails = [
+            customer_input["input"]["email"].lower()
+            for customer_input in customers_input
+            if customer_input["input"].get("email")
+        ]
+
+        duplicated_refs = get_duplicated_values(external_refs)
+        duplicated_emails = get_duplicated_values(emails)
+
+        for index, customer_input in enumerate(customers_input):
+            base_error_count = 0
+            customer_id = customer_input.get("id")
+            customer_external_ref = customer_input.get("external_reference")
+
+            shipping_address_data = customer_input["input"].pop(
+                SHIPPING_ADDRESS_FIELD, None
+            )
+            billing_address_data = customer_input["input"].pop(
+                BILLING_ADDRESS_FIELD, None
+            )
+
+            customer_input["input"] = ModelMutation.clean_input(
+                info, None, customer_input["input"], input_cls=CustomerInput
+            )
+
+            new_external_ref = customer_input["input"].get("external_reference")
+
+            base_error_count += cls.validate_customer(
+                customer_id,
+                customer_external_ref,
+                customer_input,
+                index,
+                index_error_map,
+            )
+
+            if new_external_ref in duplicated_refs:
+                message = "Duplicated externalReference value."
+                index_error_map[index].append(
+                    CustomerBulkUpdateError(
+                        path="input.externalReference",
+                        message=message,
+                        code=CustomerBulkUpdateErrorCode.DUPLICATED_INPUT_ITEM.value,
+                    )
+                )
+                base_error_count += 1
+
+            if shipping_address_data:
+                clean_shipping_address = cls.clean_address(
+                    shipping_address_data,
+                    address_type=AddressType.SHIPPING,
+                    field=SHIPPING_ADDRESS_FIELD,
+                    index=index,
+                    index_error_map=index_error_map,
+                )
+                customer_input["input"][SHIPPING_ADDRESS_FIELD] = clean_shipping_address
+
+            if billing_address_data:
+                clean_billing_address = cls.clean_address(
+                    billing_address_data,
+                    address_type=AddressType.BILLING,
+                    field=BILLING_ADDRESS_FIELD,
+                    index=index,
+                    index_error_map=index_error_map,
+                )
+                customer_input["input"][BILLING_ADDRESS_FIELD] = clean_billing_address
+
+            if email := customer_input.get("email"):
+                customer_input["email"] = email.lower()
+
+                if customer_input["email"] in duplicated_emails:
+                    message = "Duplicated email value."
+                    code = CustomerBulkUpdateErrorCode.DUPLICATED_INPUT_ITEM.value
+                    index_error_map[index].append(
+                        CustomerBulkUpdateError(
+                            path="input.email", message=message, code=code
+                        )
+                    )
+                    base_error_count += 1
+
+            if base_error_count > 0:
+                cleaned_inputs_map[index] = None
+            else:
+                cleaned_inputs_map[index] = customer_input
+
+        return cleaned_inputs_map
+
+    @classmethod
+    def get_customers(cls, cleaned_inputs_map: dict) -> list[models.User]:
+        lookup = Q()
+        for customer_input in cleaned_inputs_map.values():
+            if not customer_input:
+                continue
+
+            single_customer_lookup = Q()
+
+            if customer_id := customer_input.get("id"):
+                single_customer_lookup |= Q(pk=customer_id)
+            else:
+                single_customer_lookup |= Q(
+                    external_reference=customer_input.get("external_reference")
+                )
+            lookup |= single_customer_lookup
+
+        customers = models.User.objects.filter(lookup, is_staff=False)
path: saleor/graphql/account/bulk_mutations/customer_bulk_update.py
text: @user1:
Staff can also be a customer; there is `User.objects.customers()` queryset for that.