diff --git a/corehq/apps/api/urls.py b/corehq/apps/api/urls.py
index 7a30a6cec112..80b8c1bbe0cf 100644
--- a/corehq/apps/api/urls.py
+++ b/corehq/apps/api/urls.py
@@ -138,11 +138,19 @@ def versioned_apis(api_list):
         messaging_events, name="api_messaging_event_list"),
     url(r'messaging-event/(?P<api_version>v1)/(?P<event_id>\d+)/$',
         messaging_events, name="api_messaging_event_detail"),
+
+    # Case API v0.6 endpoints
     url(r'v0\.6/case/bulk-fetch/$', case_api_bulk_fetch),
+    # Trailing slash optional: https://github.com/dimagi/commcare-hq/pull/29939
+    url(r'v0.6/case/?$', case_api, name='case_api_v0.6'),
+    url(r'v0\.6/case/(?P<case_id>[\w\-,]+)/?$', case_api, name='case_api_v0.6_detail'),
+    path('v0.6/case/ext/<path:external_id>/', case_api),
+    # Case API v2 endpoints
     url(r'case/v2/bulk-fetch/$', case_api_bulk_fetch, name='case_api_bulk_fetch'),
-    # match v0.6/case/ AND v0.6/case/e0ad6c2e-514c-4c2b-85a7-da35bbeb1ff1/ trailing slash optional
-    url(r'v0\.6/case(?:/(?P<case_id>[\w\-,]+))?/?$', case_api),
-    url(r'case/v2(?:/(?P<case_id>[\w\-,]+))?/?$', case_api, name='case_api'),
+    url(r'case/v2/?$', case_api, name='case_api'),
+    url(r'case/v2/(?P<case_id>[\w\-,]+)/?$', case_api, name='case_api_detail'),
+    path('case/v2/ext/<path:external_id>/', case_api, name='case_api_detail_ext'),
+
     path('', include(list(versioned_apis(_OLD_API_LIST)))),
     url(r'^case/attachment/(?P<case_id>[\w\-:]+)/(?P<attachment_id>.*)$', CaseAttachmentAPI.as_view()),
     url(r'^case_attachment/v1/(?P<case_id>[\w\-:]+)/(?P<attachment_id>.*)$', CaseAttachmentAPI.as_view(),
diff --git a/corehq/apps/hqcase/api/updates.py b/corehq/apps/hqcase/api/updates.py
index 5befe3c928af..09285b0ac141 100644
--- a/corehq/apps/hqcase/api/updates.py
+++ b/corehq/apps/hqcase/api/updates.py
@@ -1,21 +1,21 @@
 import uuid
 
-from django.utils.functional import cached_property
 from django.core.exceptions import PermissionDenied
+from django.utils.functional import cached_property
 
 import jsonobject
 from jsonobject.exceptions import BadValueError
 
 from casexml.apps.case.mock import CaseBlock, IndexAttrs
 
+from corehq.apps.es.case_search import CaseSearchES
+from corehq.apps.es.users import UserES
 from corehq.apps.fixtures.utils import is_identifier_invalid
 from corehq.apps.hqcase.utils import CASEBLOCK_CHUNKSIZE, submit_case_blocks
-from corehq.form_processor.models import CommCareCase
-from corehq.sql_db.util import get_db_aliases_for_partitioned_query
-from corehq.apps.es.case_search import CaseSearchES
 from corehq.apps.locations.models import SQLLocation
-from corehq.apps.es.users import UserES
 from corehq.apps.users.models import CouchUser
+from corehq.form_processor.models import CommCareCase
+from corehq.sql_db.util import get_db_aliases_for_partitioned_query
 
 from .core import SubmissionError, UserError
 
@@ -76,7 +76,7 @@ class BaseJsonCaseChange(jsonobject.JsonObject):
     properties = jsonobject.DictProperty(validators=[valid_properties_dict], default={})
     indices = jsonobject.DictProperty(JsonIndex, validators=[valid_indices_dict])
     close = jsonobject.BooleanProperty(default=False)
-    _is_case_creation = False
+    is_new_case = False
 
     _allow_dynamic_properties = False
 
@@ -96,16 +96,22 @@ def wrap(cls, data):
     def get_caseblock(self, case_db):
 
         def get_kwargs(*attribs):
-            return {
+            kwargs = {
                 a: getattr(self, a)
                 for a in attribs
                 if getattr(self, a) is not None
             }
+            if self.is_new_case:
+                required_kwargs = {'case_type', 'case_name', 'owner_id'}
+                if missing := required_kwargs - kwargs.keys():
+                    ext_id = kwargs.get('external_id', '')
+                    raise ValueError(f'{missing} required for new case {ext_id}')
+            return kwargs
 
         return CaseBlock(
             case_id=self.get_case_id(case_db),
             user_id=self.user_id,
-            create=self._is_case_creation,
+            create=self.is_new_case,
             update=dict(self.properties),
             close=self.close,
             index={
@@ -118,10 +124,6 @@ def get_kwargs(*attribs):
             **get_kwargs('case_type', 'case_name', 'external_id', 'owner_id'),
         ).as_text()
 
-    @property
-    def is_new_case(self):
-        return self._is_case_creation
-
 
 class JsonCaseCreation(BaseJsonCaseChange):
     temporary_id = jsonobject.StringProperty()
@@ -131,7 +133,7 @@ class JsonCaseCreation(BaseJsonCaseChange):
     case_type = jsonobject.StringProperty(required=True)
     owner_id = jsonobject.StringProperty(required=True)
 
-    _is_case_creation = True
+    is_new_case = True
 
     @classmethod
     def wrap(cls, data):
@@ -145,7 +147,7 @@ def get_case_id(self, case_db):
 
 
 class JsonCaseUpdate(BaseJsonCaseChange):
-    _is_case_creation = False
+    is_new_case = False
 
     def validate(self, *args, **kwargs):
         super().validate(*args, **kwargs)
@@ -161,12 +163,44 @@ def get_case_id(self, case_db):
         return case_db.get_by_external_id(self.external_id)
 
 
+class JsonCaseUpsert(BaseJsonCaseChange):
+    """Handles UPSERT operations where create/update is determined at lookup time."""
+    external_id = jsonobject.StringProperty(required=True)
+
+    _is_case_creation = ...  # Determined when get_case_id() is called
+
+    @property
+    def is_new_case(self):
+        if self._is_case_creation is ...:
+            raise ValueError('is_new_case has not yet been initialized')
+        return self._is_case_creation
+
+    @classmethod
+    def wrap(cls, data):
+        if 'case_id' in data:
+            raise UserError("UPSERT does not allow case_id to be specified")
+        return super().wrap(data)
+
+    def get_case_id(self, case_db):
+        if self.case_id:
+            return self.case_id
+
+        existing_case_id = case_db.get_upsert_case_id(self.external_id)
+        if existing_case_id:
+            self._is_case_creation = False
+            self.case_id = existing_case_id
+        else:
+            self._is_case_creation = True
+            self.case_id = str(uuid.uuid4())
+        return self.case_id
+
+
 def handle_case_update(domain, data, user, device_id, is_creation, xmlns=None):
     is_bulk = isinstance(data, list)
     if is_bulk:
-        updates = _get_bulk_updates(domain, data, user)
+        updates = _get_bulk_updates(data, user.user_id)
     else:
-        updates = [_get_individual_update(domain, data, user, is_creation)]
+        updates = [_get_individual_update(data, user.user_id, is_creation)]
 
     case_db = CaseIDLookerUpper(domain, updates)
 
@@ -184,9 +218,9 @@ def handle_case_update(domain, data, user, device_id, is_creation, xmlns=None):
         return xform, cases[0]
 
 
-def _get_individual_update(domain, data, user, is_creation):
+def _get_individual_update(data, user_id, is_creation):
     update_class = JsonCaseCreation if is_creation else JsonCaseUpdate
-    data['user_id'] = user.user_id
+    data['user_id'] = user_id
     try:
         update = update_class.wrap(data)
     except BadValueError as e:
@@ -194,7 +228,15 @@ def _get_individual_update(domain, data, user, is_creation):
     return update
 
 
-def _get_bulk_updates(domain, all_data, user):
+def _get_upsert_update(data, user_id):
+    data['user_id'] = user_id
+    try:
+        return JsonCaseUpsert.wrap(data)
+    except BadValueError as err:
+        raise UserError(str(err))
+
+
+def _get_bulk_updates(all_data, user_id):
     if len(all_data) > CASEBLOCK_CHUNKSIZE:
         raise UserError(f"You cannot submit more than {CASEBLOCK_CHUNKSIZE} updates in a single request")
 
@@ -202,10 +244,13 @@ def _get_bulk_updates(domain, all_data, user):
     errors = []
     for i, data in enumerate(all_data, start=1):
         try:
-            is_creation = data.pop('create', None)
-            if is_creation is None:
+            if 'create' not in data:
                 raise UserError("A 'create' flag is required for each update.")
-            updates.append(_get_individual_update(domain, data, user, is_creation))
+            create_flag = data.pop('create')
+            if create_flag is None:
+                updates.append(_get_upsert_update(data, user_id))
+            else:
+                updates.append(_get_individual_update(data, user_id, create_flag))
         except UserError as e:
             errors.append(f'Error in row {i}: {e}')
 
@@ -245,6 +290,10 @@ def get_by_external_id(self, key):
         except KeyError:
             raise UserError(f"Could not find a case with external_id '{key}'")
 
+    def get_upsert_case_id(self, external_id):
+        """Returns case_id if found, None if not found (indicating creation)."""
+        return self._by_external_id.get(external_id)
+
     @cached_property
     def _by_external_id(self):
         ids_in_request = {
@@ -254,7 +303,7 @@ def _by_external_id(self):
 
         ids_to_find = {
             update.external_id for update in self.updates
-            if not update._is_case_creation and not update.case_id
+            if not isinstance(update, JsonCaseCreation) and not update.case_id
         } | {
             index.external_id for update in self.updates for index in update.indices.values()
         }
diff --git a/corehq/apps/hqcase/tests/test_api_updates.py b/corehq/apps/hqcase/tests/test_api_updates.py
new file mode 100644
index 000000000000..6f78bad637d2
--- /dev/null
+++ b/corehq/apps/hqcase/tests/test_api_updates.py
@@ -0,0 +1,330 @@
+import uuid
+
+from django.test import TestCase
+
+import pytest
+
+from casexml.apps.case.mock import CaseBlock
+
+from corehq.apps.domain.shortcuts import create_domain
+from corehq.apps.hqcase.utils import submit_case_blocks
+from corehq.apps.users.models import WebUser
+from corehq.form_processor.tests.utils import FormProcessorTestUtils, sharded
+
+from ..api.core import UserError
+from ..api.updates import (
+    JsonCaseUpsert,
+    _get_bulk_updates,
+    _get_individual_update,
+    handle_case_update,
+)
+
+
+class TestIndividualUpdate:
+
+    def test_individual_update_with_is_creation_true(self):
+        data = {
+            "case_type": "patient",
+            "case_name": "Jane Doe",
+            "owner_id": "test-owner",
+        }
+        update = _get_individual_update(
+            data,
+            'user_id',
+            is_creation=True
+        )
+        assert update.is_new_case
+
+    def test_individual_update_with_is_creation_false(self):
+        data = {
+            "case_id": "existing-case",
+            "case_name": "Updated Name",
+        }
+        update = _get_individual_update(
+            data,
+            'user_id',
+            is_creation=False
+        )
+        assert not update.is_new_case
+        assert update.case_id == "existing-case"
+
+
+class TestBulkUpdates:
+
+    def test_bulk_update_missing_create_flag_raises_error(self):
+        data = [
+            {
+                "case_id": "some-case-id",
+                "case_name": "Test Case",
+            }
+        ]
+        with pytest.raises(
+            UserError,
+            match="Error in row 1: A 'create' flag is required for each update.",
+        ):
+            _get_bulk_updates(data, 'user_id')
+
+    def test_bulk_update_with_create_true(self):
+        data = [
+            {
+                "create": True,
+                "case_type": "patient",
+                "case_name": "Jane Doe",
+                "owner_id": "test-owner",
+            }
+        ]
+        updates = _get_bulk_updates(data, 'user_id')
+        assert len(updates) == 1
+        assert updates[0].is_new_case
+
+    def test_bulk_update_with_create_false(self):
+        data = [
+            {
+                "create": False,
+                "case_id": "existing-case",
+                "case_name": "Updated Name",
+            }
+        ]
+        updates = _get_bulk_updates(data, 'user_id')
+        assert len(updates) == 1
+        assert not updates[0].is_new_case
+        assert updates[0].case_id == "existing-case"
+
+    def test_bulk_update_with_create_none_returns_upsert(self):
+        data = [
+            {
+                "create": None,
+                "case_type": "patient",
+                "case_name": "John Doe",
+                "external_id": "ext-123",
+                "owner_id": "test-owner",
+            }
+        ]
+        updates = _get_bulk_updates(data, 'user_id')
+        assert len(updates) == 1
+        assert isinstance(updates[0], JsonCaseUpsert)
+        with pytest.raises(ValueError):
+            # is_new_case raises ValueError until get_case_id() initializes its value
+            updates[0].is_new_case
+        assert updates[0].external_id == "ext-123"
+
+    def test_bulk_update_with_create_none_without_external_id_raises_error(self):
+        data = [
+            {
+                "create": None,
+                "case_name": "Test Case",
+                "case_type": "patient",
+                "owner_id": "test-owner",
+            }
+        ]
+        with pytest.raises(
+            UserError,
+            match='Error in row 1: Property external_id is required.',
+        ):
+            _get_bulk_updates(data, 'user_id')
+
+    def test_bulk_update_with_create_none_with_case_id_raises_error(self):
+        data = [
+            {
+                "create": None,
+                "case_id": "case-123",
+                "case_type": "patient",
+                "case_name": "Test Case",
+                "external_id": "ext-456",
+                "owner_id": "test-owner",
+            }
+        ]
+        with pytest.raises(
+            UserError,
+            match='Error in row 1: UPSERT does not allow case_id to be specified',
+        ):
+            _get_bulk_updates(data, 'user_id')
+
+    def test_bulk_update_with_error_reports_row_number(self):
+        data = [
+            {
+                "create": True,
+                # Missing required fields for creation (no owner_id)
+                "case_type": "patient",
+                "case_name": "Missing owner_id",
+            },
+        ]
+        with pytest.raises(
+            UserError,
+            match='Error in row 1: Property owner_id is required.',
+        ):
+            _get_bulk_updates(data, 'user_id')
+
+
+@sharded
+class TestUpsertIntegration(TestCase):
+    """End-to-end tests for UPSERT through handle_case_update()."""
+    domain = "test-upsert-integration"
+
+    @classmethod
+    def setUpClass(cls):
+        super().setUpClass()
+        cls.domain_obj = create_domain(cls.domain)
+        cls.web_user = WebUser.create(
+            cls.domain, "testuser", "password", None, None
+        )
+
+    def tearDown(self):
+        FormProcessorTestUtils.delete_all_cases(self.domain)
+        FormProcessorTestUtils.delete_all_xforms(self.domain)
+
+    @classmethod
+    def tearDownClass(cls):
+        cls.web_user.delete(cls.domain, deleted_by=None)
+        cls.domain_obj.delete()
+        super().tearDownClass()
+
+    def test_upsert_creates_case_when_external_id_not_found(self):
+        data = [
+            {
+                "create": None,
+                "case_type": "patient",
+                "case_name": "New Patient",
+                "external_id": "new-ext-id",
+                "owner_id": self.web_user.user_id,
+            }
+        ]
+
+        xform, cases = handle_case_update(
+            self.domain,
+            data,
+            self.web_user,
+            device_id="test",
+            is_creation=None,
+        )
+
+        assert len(cases) == 1
+        case = cases[0]
+        assert case.name == "New Patient"
+        assert case.external_id == "new-ext-id"
+        assert case.type == "patient"
+
+    def test_upsert_with_missing_external_id(self):
+        data = [
+            {
+                "create": None,
+                "case_type": "patient",
+                "case_name": "New Patient",
+                "owner_id": self.web_user.user_id,
+            }
+        ]
+
+        with pytest.raises(UserError):
+            handle_case_update(
+                self.domain,
+                data,
+                self.web_user,
+                device_id="test",
+                is_creation=None,
+            )
+
+    def test_upsert_create_with_missing_data(self):
+        data = [
+            {
+                "create": None,
+                "external_id": "new-ext-id",
+                "owner_id": self.web_user.user_id,
+            }
+        ]
+
+        with pytest.raises(
+            ValueError,
+            match='required for new case new-ext-id',
+        ):
+            handle_case_update(
+                self.domain,
+                data,
+                self.web_user,
+                device_id="test",
+                is_creation=None,
+            )
+
+    def test_upsert_updates_case_when_external_id_exists(self):
+        existing_case_id = str(uuid.uuid4())
+        case_block = CaseBlock(
+            case_id=existing_case_id,
+            case_type="patient",
+            case_name="Original Name",
+            external_id="existing-ext-id",
+            owner_id=self.web_user.user_id,
+            create=True,
+        )
+        submit_case_blocks(case_block.as_text(), domain=self.domain)
+
+        data = [
+            {
+                "create": None,
+                # case_type not required for update
+                "case_name": "Updated Name",
+                "external_id": "existing-ext-id",
+                "owner_id": self.web_user.user_id,
+                "properties": {"status": "updated"},
+            }
+        ]
+
+        xform, cases = handle_case_update(
+            self.domain,
+            data,
+            self.web_user,
+            device_id="test",
+            is_creation=None,
+        )
+
+        assert len(cases) == 1
+        case = cases[0]
+        assert case.case_id == existing_case_id
+        assert case.name == "Updated Name"
+        assert case.get_case_property("status") == "updated"
+
+    def test_bulk_upsert_mixed_creates_and_updates(self):
+        existing_case_id = str(uuid.uuid4())
+        case_block = CaseBlock(
+            case_id=existing_case_id,
+            case_type="patient",
+            case_name="Existing Patient",
+            external_id="ext-existing",
+            owner_id=self.web_user.user_id,
+            create=True,
+        )
+        submit_case_blocks(case_block.as_text(), domain=self.domain)
+
+        data = [
+            {
+                "create": None,
+                "case_type": "patient",
+                "case_name": "Updated Existing",
+                "external_id": "ext-existing",
+                "owner_id": self.web_user.user_id,
+            },
+            {
+                "create": None,
+                "case_type": "patient",
+                "case_name": "Brand New",
+                "external_id": "ext-new",
+                "owner_id": self.web_user.user_id,
+            },
+        ]
+
+        xform, cases = handle_case_update(
+            self.domain,
+            data,
+            self.web_user,
+            device_id="test",
+            is_creation=None,
+        )
+
+        assert len(cases) == 2
+
+        updated_case = next(c for c in cases if c.external_id == "ext-existing")
+        new_case = next(c for c in cases if c.external_id == "ext-new")
+
+        assert updated_case.case_id == existing_case_id
+        assert updated_case.name == "Updated Existing"
+
+        assert new_case.case_id != existing_case_id
+        assert new_case.name == "Brand New"
diff --git a/corehq/apps/hqcase/tests/test_case_api_bulk_get.py b/corehq/apps/hqcase/tests/test_case_api_bulk_get.py
index 3497ad7e6b5b..005c9f58f45f 100644
--- a/corehq/apps/hqcase/tests/test_case_api_bulk_get.py
+++ b/corehq/apps/hqcase/tests/test_case_api_bulk_get.py
@@ -7,6 +7,7 @@
 from casexml.apps.case.mock import CaseBlock
 
 from corehq import privileges
+from corehq.apps.data_dictionary.models import CaseType
 from corehq.apps.domain.shortcuts import create_domain
 from corehq.apps.es.case_search import case_search_adapter
 from corehq.apps.es.tests.utils import (
@@ -14,6 +15,8 @@
     es_test,
     populate_case_search_index,
 )
+from corehq.apps.hqcase.api.core import serialize_es_case
+from corehq.apps.hqcase.api.get_bulk import get_bulk
 from corehq.apps.hqcase.utils import submit_case_blocks
 from corehq.apps.users.models import HqPermissions, UserRole, WebUser
 from corehq.form_processor.tests.utils import FormProcessorTestUtils
@@ -22,7 +25,6 @@
     flag_enabled,
     privilege_enabled,
 )
-from corehq.apps.data_dictionary.models import CaseType
 
 
 @es_test(requires=[case_search_adapter], setup_class=True)
@@ -80,21 +82,6 @@ def _make_case_block(cls, name, case_type='player', external_id=None):
             external_id=external_id,
         )
 
-    def test_get_single_case(self):
-        res = self.client.get(reverse('case_api', args=(self.domain, self.case_ids[0])))
-        self.assertEqual(res.status_code, 200)
-        self.assertEqual(res.json()['case_id'], self.case_ids[0])
-
-    def test_get_single_case_not_found(self):
-        res = self.client.get(reverse('case_api', args=(self.domain, 'fake_id')))
-        self.assertEqual(res.status_code, 404)
-        self.assertEqual(res.json()['error'], "Case 'fake_id' not found")
-
-    def test_get_single_case_on_other_domain(self):
-        res = self.client.get(reverse('case_api', args=(self.domain, self.other_domain_case_id)))
-        self.assertEqual(res.status_code, 404)
-        self.assertEqual(res.json()['error'], f"Case '{self.other_domain_case_id}' not found")
-
     def test_bulk_get(self):
         case_ids = self.case_ids[0:2]
         self._call_get_api_check_results(case_ids, matching=2, missing=0)
@@ -102,13 +89,13 @@ def test_bulk_get(self):
     def test_bulk_get_domain_filter(self):
         case_ids = self.case_ids[0:2] + [self.other_domain_case_id]
         result = self._call_get_api_check_results(case_ids, matching=2, missing=1)
-        self.assertEqual(result['cases'][2]['error'], 'not found')
+        assert result['cases'][2]['error'] == 'not found'
 
     def test_bulk_get_not_found(self):
         case_ids = ['missing1', self.case_ids[1], 'missing2']
         result = self._call_get_api_check_results(case_ids, matching=1, missing=2)
-        self.assertEqual(result['cases'][0]['error'], 'not found')
-        self.assertEqual(result['cases'][2]['error'], 'not found')
+        assert result['cases'][0]['error'] == 'not found'
+        assert result['cases'][2]['error'] == 'not found'
 
     def test_bulk_get_duplicate(self):
         """Duplicate case IDs in the request results in duplicates in the response"""
@@ -167,11 +154,11 @@ def test_bulk_post_case_ids_and_external_ids_missing(self):
         )
 
     def _call_get_api_check_results(self, case_ids, matching=None, missing=None):
-        res = self.client.get(reverse('case_api', args=(self.domain, ','.join(case_ids))))
-        self.assertEqual(res.status_code, 200)
+        res = self.client.get(reverse('case_api_detail', args=(self.domain, ','.join(case_ids))))
+        assert res.status_code == 200
         result = res.json()
         result_case_ids = [case['case_id'] for case in result['cases']]
-        self.assertEqual(result_case_ids, case_ids)
+        assert result_case_ids == case_ids
         self._check_matching_missing(result, matching, missing)
         return result
 
@@ -183,13 +170,13 @@ def _call_post_api_check_results(self, case_ids=None, external_ids=None, matchin
         self._check_matching_missing(result, matching, missing)
 
         total_expected = len(case_ids or []) + len(external_ids or [])
-        self.assertEqual(len(cases), total_expected)
+        assert len(cases) == total_expected
 
         # check for results as well as result order
         if case_ids:
             # case_id results are always at the front
             result_case_ids = [case.get('case_id') for case in cases]
-            self.assertEqual(case_ids, result_case_ids[:len(case_ids)])
+            assert case_ids == result_case_ids[:len(case_ids)]
 
         if external_ids:
             # external_id results are always at the end so reverse the lists and compare
@@ -197,7 +184,7 @@ def _call_post_api_check_results(self, case_ids=None, external_ids=None, matchin
             result_external_ids = list(reversed([
                 case.get('external_id') for case in cases
             ]))
-            self.assertEqual(external_ids, result_external_ids[:len(external_ids)])
+            assert external_ids == result_external_ids[:len(external_ids)]
 
         return result
 
@@ -209,12 +196,137 @@ def _call_post(self, case_ids=None, external_ids=None, expected_status=200):
             data['external_ids'] = external_ids
         url = reverse('case_api_bulk_fetch', args=(self.domain,))
         res = self.client.post(url, data=data, content_type="application/json")
-        self.assertEqual(res.status_code, expected_status, res.json())
+        assert res.status_code == expected_status, res.json()
         return res
 
     def _check_matching_missing(self, result, matching=None, missing=None):
         if matching is not None:
-            self.assertEqual(result['matching_records'], matching)
+            assert result['matching_records'] == matching
 
         if missing is not None:
-            self.assertEqual(result['missing_records'], missing)
+            assert result['missing_records'] == missing
+
+
+@es_test(requires=[case_search_adapter], setup_class=True)
+@disable_quickcache
+@privilege_enabled(privileges.API_ACCESS)
+@flag_enabled('API_THROTTLE_WHITELIST')
+class TestGetBulkFunction(TestCase):
+    domain = 'test-get-bulk-function'
+    maxDiff = None
+
+    @classmethod
+    def setUpClass(cls):
+        super().setUpClass()
+        cls.domain_obj = create_domain(cls.domain)
+        role = UserRole.create(
+            cls.domain, 'edit-data', permissions=HqPermissions(
+                edit_data=True,
+                access_api=True,
+                access_all_locations=True
+            )
+        )
+        cls.web_user = WebUser.create(cls.domain, 'test-user', 'password', None, None, role_id=role.get_id)
+
+        case_blocks = [
+            cls._make_case_block('Test Case 1', external_id='test1'),
+            cls._make_case_block('Test Case 2', external_id='test2'),
+            cls._make_case_block('Test Case 3', external_id='test2'),  # duplicate external_id
+        ]
+        case_search_es_setup(cls.domain, case_blocks)
+        cls.case_ids = [b.case_id for b in case_blocks]
+
+    @classmethod
+    def tearDownClass(cls):
+        cls.web_user.delete(cls.domain, deleted_by=None)
+        cls.domain_obj.delete()
+        FormProcessorTestUtils.delete_all_cases(cls.domain)
+        FormProcessorTestUtils.delete_all_xforms(cls.domain)
+        super().tearDownClass()
+
+    @classmethod
+    def _make_case_block(cls, name, case_type='player', external_id=None):
+        return CaseBlock(
+            case_id=str(uuid.uuid4()),
+            case_type=case_type,
+            case_name=name,
+            external_id=external_id,
+        )
+
+    def test_get_bulk_one_case_found_by_case_id(self):
+        result = get_bulk(self.domain, self.web_user, case_ids=[self.case_ids[0]])
+
+        expected_case = serialize_es_case(case_search_adapter.get(self.case_ids[0]))
+        assert result == {
+            'cases': [expected_case],
+            'matching_records': 1,
+            'missing_records': 0,
+        }
+
+    def test_get_bulk_one_case_found_by_external_id(self):
+        result = get_bulk(self.domain, self.web_user, external_ids=['test1'])
+
+        expected_case = serialize_es_case(case_search_adapter.get(self.case_ids[0]))
+        assert result == {
+            'cases': [expected_case],
+            'matching_records': 1,
+            'missing_records': 0,
+        }
+
+    def test_get_bulk_no_cases_found_by_case_id(self):
+        fake_id = 'fake-case-id-12345'
+        result = get_bulk(self.domain, self.web_user, case_ids=[fake_id])
+
+        assert result == {
+            'cases': [{'case_id': fake_id, 'error': 'not found'}],
+            'matching_records': 0,
+            'missing_records': 1,
+        }
+
+    def test_get_bulk_no_cases_found_by_external_id(self):
+        fake_external_id = 'fake-external-id'
+        result = get_bulk(self.domain, self.web_user, external_ids=[fake_external_id])
+
+        assert result == {
+            'cases': [{'external_id': fake_external_id, 'error': 'not found'}],
+            'matching_records': 0,
+            'missing_records': 1,
+        }
+
+    def test_get_bulk_duplicate_external_id(self):
+        result = get_bulk(self.domain, self.web_user, external_ids=['test2'])
+
+        # When multiple cases have the same external_id, only one is
+        # returned because in `_prepare_result()`, the `results_by_id`
+        # dictionary uses `external_id` as the key.
+        expected_case_1 = serialize_es_case(case_search_adapter.get(self.case_ids[1]))
+        expected_case_2 = serialize_es_case(case_search_adapter.get(self.case_ids[2]))
+
+        possible_results = [
+            {
+                'cases': [expected_case_1],
+                'matching_records': 1,
+                'missing_records': 0,
+            },
+            {
+                'cases': [expected_case_2],
+                'matching_records': 1,
+                'missing_records': 0,
+            },
+        ]
+        assert result in possible_results
+
+    def test_get_bulk_repeats(self):
+        """
+        If case_ids and external_ids refer to the same cases, they will
+        be repeated in `get_bulk()` results.
+        """
+        result = get_bulk(
+            self.domain,
+            self.web_user,
+            case_ids=[self.case_ids[0]],  # Case 1
+            external_ids=['test1'],  # Also case 1
+        )
+        assert result['matching_records'] == 2
+        case_names = [c['case_name'] for c in result['cases']]
+        assert case_names == ['Test Case 1', 'Test Case 1']
diff --git a/corehq/apps/hqcase/tests/test_case_api_get.py b/corehq/apps/hqcase/tests/test_case_api_get.py
new file mode 100644
index 000000000000..f6087f84f47c
--- /dev/null
+++ b/corehq/apps/hqcase/tests/test_case_api_get.py
@@ -0,0 +1,124 @@
+import uuid
+
+from django.test import TestCase
+from django.urls import reverse
+
+from casexml.apps.case.mock import CaseBlock
+
+from corehq import privileges
+from corehq.apps.domain.shortcuts import create_domain
+from corehq.apps.es.case_search import case_search_adapter
+from corehq.apps.es.tests.utils import (
+    case_search_es_setup,
+    es_test,
+    populate_case_search_index,
+)
+from corehq.apps.hqcase.api.core import serialize_es_case
+from corehq.apps.hqcase.utils import submit_case_blocks
+from corehq.apps.users.models import HqPermissions, UserRole, WebUser
+from corehq.form_processor.tests.utils import FormProcessorTestUtils
+from corehq.util.test_utils import (
+    disable_quickcache,
+    flag_enabled,
+    privilege_enabled,
+)
+
+
+@es_test(requires=[case_search_adapter], setup_class=True)
+@disable_quickcache
+@privilege_enabled(privileges.API_ACCESS)
+@flag_enabled('API_THROTTLE_WHITELIST')
+class TestCaseAPIGet(TestCase):
+    domain = 'test-get-cases'
+    maxDiff = None
+
+    @classmethod
+    def setUpClass(cls):
+        super().setUpClass()
+        cls.domain_obj = create_domain(cls.domain)
+        role = UserRole.create(
+            cls.domain, 'edit-data', permissions=HqPermissions(edit_data=True, access_api=True)
+        )
+        cls.web_user = WebUser.create(cls.domain, 'netflix', 'password', None, None, role_id=role.get_id)
+        case_blocks = [
+            cls._make_case_block('Vera Menchik', external_id='vera'),
+            cls._make_case_block('Nona Gaprindashvili', external_id='nona'),
+        ]
+        case_search_es_setup(cls.domain, case_blocks)
+        cls.case_ids = [b.case_id for b in case_blocks]
+
+        xform, [case] = submit_case_blocks(
+            cls._make_case_block('Vera Menchik', external_id='vera').as_text(), domain='other-domain'
+        )
+        populate_case_search_index([case])
+        cls.other_domain_case_id = case.case_id
+
+    def setUp(self):
+        self.client.login(username='netflix', password='password')
+
+    @classmethod
+    def tearDownClass(cls):
+        cls.web_user.delete(cls.domain, deleted_by=None)
+        cls.domain_obj.delete()
+        FormProcessorTestUtils.delete_all_cases(cls.domain)
+        FormProcessorTestUtils.delete_all_xforms(cls.domain)
+        super().tearDownClass()
+
+    @classmethod
+    def _make_case_block(cls, name, case_type='player', external_id=None):
+        return CaseBlock(
+            case_id=str(uuid.uuid4()),
+            case_type=case_type,
+            case_name=name,
+            external_id=external_id,
+        )
+
+    def test_get_single_case(self):
+        res = self.client.get(reverse('case_api_detail', args=(self.domain, self.case_ids[0])))
+        assert res.status_code == 200
+        assert res.json()['case_id'] == self.case_ids[0]
+
+    def test_get_single_case_by_external_id(self):
+        case_id = '11111111-1111-4111-8111-111111111111'
+        external_id = 'ext-get-case'
+        case_block = CaseBlock(
+            case_id=case_id,
+            case_type='player',
+            case_name='Irina Levitina',
+            external_id=external_id,
+            owner_id=self.web_user.get_id,
+            create=True,
+        )
+        _, [case] = submit_case_blocks(case_block.as_text(), domain=self.domain)
+        populate_case_search_index([case])
+        es_doc = case_search_adapter.get(case_id)
+        expected = serialize_es_case(es_doc)
+
+        base_url = reverse('case_api', args=(self.domain,))
+        if not base_url.endswith('/'):
+            base_url += '/'
+        url = f"{base_url}ext/{external_id}/"
+        res = self.client.get(url)
+
+        assert res.status_code == 200
+        assert res.json() == expected
+
+    def test_get_case_by_external_id_not_found(self):
+        base_url = reverse('case_api', args=(self.domain,))
+        if not base_url.endswith('/'):
+            base_url += '/'
+        url = f"{base_url}ext/missing-external-id/"
+        res = self.client.get(url)
+
+        assert res.status_code == 404
+        assert res.json()['error'] == "Case 'missing-external-id' not found"
+
+    def test_get_single_case_not_found(self):
+        res = self.client.get(reverse('case_api_detail', args=(self.domain, 'fake_id')))
+        assert res.status_code == 404
+        assert res.json()['error'] == "Case 'fake_id' not found"
+
+    def test_get_single_case_on_other_domain(self):
+        res = self.client.get(reverse('case_api_detail', args=(self.domain, self.other_domain_case_id)))
+        assert res.status_code == 404
+        assert res.json()['error'] == f"Case '{self.other_domain_case_id}' not found"
diff --git a/corehq/apps/hqcase/tests/test_case_api_permissions.py b/corehq/apps/hqcase/tests/test_case_api_permissions.py
index 3b1d8c242418..6ee64f348cae 100644
--- a/corehq/apps/hqcase/tests/test_case_api_permissions.py
+++ b/corehq/apps/hqcase/tests/test_case_api_permissions.py
@@ -168,7 +168,7 @@ def test_case_api_get_successful(self):
             self.client,
         ):
             case_id = self.case_mapping['user_case']
-            url = reverse('case_api', args=(self.domain, case_id))
+            url = reverse('case_api_detail', args=(self.domain, case_id))
             response = self.client.get(url)
             self.assertEqual(response.status_code, 200)
             json = response.json()
@@ -182,7 +182,7 @@ def test_case_api_get_no_permission(self):
             self.client,
         ):
             case_id = self.case_mapping['restricted_case']
-            url = reverse('case_api', args=(self.domain, case_id))
+            url = reverse('case_api_detail', args=(self.domain, case_id))
             response = self.client.get(url)
             self.assertEqual(response.status_code, 403)
 
@@ -194,7 +194,7 @@ def test_case_api_get_access_all_locations(self):
             self.client,
         ):
             case_id = self.case_mapping['restricted_case']
-            url = reverse('case_api', args=(self.domain, case_id))
+            url = reverse('case_api_detail', args=(self.domain, case_id))
             response = self.client.get(url)
             self.assertEqual(response.status_code, 200)
             json = response.json()
@@ -208,7 +208,7 @@ def test_case_api_update_successful(self):
             self.client,
         ):
             case_id = self.case_mapping['user_case']
-            url = reverse('case_api', args=(self.domain, case_id))
+            url = reverse('case_api_detail', args=(self.domain, case_id))
             response = self.client.put(
                 url,
                 {'properties': self.test_case_property},
@@ -226,7 +226,7 @@ def test_case_api_update_no_permission(self):
             self.client,
         ):
             case_id = self.case_mapping['restricted_case']
-            url = reverse('case_api', args=(self.domain, case_id))
+            url = reverse('case_api_detail', args=(self.domain, case_id))
             response = self.client.put(
                 url,
                 {'properties': self.test_case_property},
@@ -242,7 +242,7 @@ def test_case_api_update_access_all_locations(self):
             self.client,
         ):
             case_id = self.case_mapping['restricted_case']
-            url = reverse('case_api', args=(self.domain, case_id))
+            url = reverse('case_api_detail', args=(self.domain, case_id))
             response = self.client.put(
                 url,
                 {'properties': self.test_case_property},
diff --git a/corehq/apps/hqcase/tests/test_case_api_updates.py b/corehq/apps/hqcase/tests/test_case_api_updates.py
index b364e2e9c1ae..4c3d79566814 100644
--- a/corehq/apps/hqcase/tests/test_case_api_updates.py
+++ b/corehq/apps/hqcase/tests/test_case_api_updates.py
@@ -1,6 +1,10 @@
+from unittest.mock import Mock
+
 from django.test import SimpleTestCase
 
-from corehq.apps.hqcase.api.updates import JsonCaseCreation
+import pytest
+
+from corehq.apps.hqcase.api.updates import JsonCaseCreation, JsonCaseUpsert
 
 
 class GetCaseBlockTests(SimpleTestCase):
@@ -19,8 +23,8 @@ def setUp(self):
     def test_kwargs(self):
         json_case = JsonCaseCreation(self.data)
         case_block = json_case.get_caseblock(case_db=None)
-        self.assertIn('<case_name>Mark Renton</case_name>', case_block)
-        self.assertNotIn('<external_id', case_block)
+        assert '<case_name>Mark Renton</case_name>' in case_block
+        assert '<external_id' not in case_block
 
     def test_kwargs_incl_external_id(self):
         case_data = self.data | {'external_id': 'Ewan McGregor'}
@@ -43,8 +47,86 @@ def test_kwargs_incl_external_id(self):
         #       </update>
         #     </case>
         #
-        self.assertIn('<case_type>trainspotter</case_type>', case_block)
-        self.assertIn('<case_name>Mark Renton</case_name>', case_block)
-        self.assertIn('<owner_id>abc123</owner_id>', case_block)
-        self.assertIn('<external_id>Ewan McGregor</external_id>', case_block)
-        self.assertIn('<age>26</age>', case_block)
+        assert '<case_type>trainspotter</case_type>' in case_block
+        assert '<case_name>Mark Renton</case_name>' in case_block
+        assert '<owner_id>abc123</owner_id>' in case_block
+        assert '<external_id>Ewan McGregor</external_id>' in case_block
+        assert '<age>26</age>' in case_block
+
+
+class JsonCaseUpsertTests(SimpleTestCase):
+
+    def setUp(self):
+        self.data = {
+            'case_name': 'John Doe',
+            'case_type': 'patient',
+            'external_id': 'ext-123',
+            'owner_id': 'owner-abc',
+            'user_id': 'user-xyz',
+            'properties': {'status': 'active'},
+        }
+
+    def test_get_case_id_when_case_exists(self):
+        upsert = JsonCaseUpsert(self.data)
+        case_db = Mock()
+        case_db.get_upsert_case_id.return_value = 'existing-case-id'
+
+        case_id = upsert.get_case_id(case_db)
+
+        assert case_id == 'existing-case-id'
+        assert not upsert._is_case_creation
+        case_db.get_upsert_case_id.assert_called_once_with('ext-123')
+
+    def test_get_case_id_when_case_does_not_exist(self):
+        upsert = JsonCaseUpsert(self.data)
+        case_db = Mock()
+        case_db.get_upsert_case_id.return_value = None
+
+        case_id = upsert.get_case_id(case_db)
+
+        assert case_id is not None
+        assert len(case_id) == 36  # UUID format
+        assert upsert._is_case_creation
+
+    def test_get_case_id_is_idempotent(self):
+        upsert = JsonCaseUpsert(self.data)
+        case_db = Mock()
+        case_db.get_upsert_case_id.return_value = None
+
+        case_id_1 = upsert.get_case_id(case_db)
+        case_id_2 = upsert.get_case_id(case_db)
+
+        assert case_id_1 == case_id_2
+        # Should only call case_db once due to caching
+        case_db.get_upsert_case_id.assert_called_once()
+
+    def test_get_caseblock_for_creation(self):
+        upsert = JsonCaseUpsert(self.data)
+        case_db = Mock()
+        case_db.get_upsert_case_id.return_value = None
+
+        case_block = upsert.get_caseblock(case_db)
+
+        assert '<create>' in case_block
+        assert '<case_type>patient</case_type>' in case_block
+        assert '<case_name>John Doe</case_name>' in case_block
+        assert '<owner_id>owner-abc</owner_id>' in case_block
+
+    def test_get_caseblock_for_update(self):
+        upsert = JsonCaseUpsert(self.data)
+        case_db = Mock()
+        case_db.get_upsert_case_id.return_value = 'existing-case-id'
+
+        case_block = upsert.get_caseblock(case_db)
+
+        assert '<create>' not in case_block
+        assert '<update>' in case_block
+        assert '<status>active</status>' in case_block
+
+    def test_is_new_case_not_initialized(self):
+        upsert = JsonCaseUpsert(self.data)
+        with pytest.raises(
+            ValueError,
+            match='is_new_case has not yet been initialized',
+        ):
+            upsert.is_new_case
diff --git a/corehq/apps/hqcase/tests/test_case_update_api.py b/corehq/apps/hqcase/tests/test_case_update_api.py
index 6006eb965dff..c8e671def4bd 100644
--- a/corehq/apps/hqcase/tests/test_case_update_api.py
+++ b/corehq/apps/hqcase/tests/test_case_update_api.py
@@ -1,5 +1,6 @@
+import json
 import uuid
-from unittest.mock import patch, MagicMock
+from unittest.mock import MagicMock, patch
 
 from django.test import TestCase
 from django.urls import reverse
@@ -80,7 +81,7 @@ def _create_case(self, body):
 
     def _update_case(self, case_id, body):
         return self.client.put(
-            reverse('case_api', args=(self.domain, case_id,)),
+            reverse('case_api_detail', args=(self.domain, case_id,)),
             body,
             content_type="application/json;charset=utf-8",
             HTTP_USER_AGENT="user agent string",
@@ -93,8 +94,8 @@ def _bulk_update_cases(self, body):
     def test_basic_get_list(self):
         with patch('corehq.apps.hqcase.views.get_list', lambda *args: {'example': 'result'}):
             res = self.client.get(reverse('case_api', args=(self.domain,)))
-        self.assertEqual(res.status_code, 200)
-        self.assertEqual(res.json(), {'example': 'result'})
+        assert res.status_code == 200
+        assert res.json() == {'example': 'result'}
 
     def test_create_case(self):
         res = self._create_case({
@@ -108,24 +109,24 @@ def test_create_case(self):
                 'dob': '1948-11-02',
             },
         }).json()
-        self.assertItemsEqual(res.keys(), ['case', 'form_id'])
+        assert set(res.keys()) == set(['case', 'form_id'])
         case = CommCareCase.objects.get_case(res['case']['case_id'], self.domain)
-        self.assertEqual(case.domain, self.domain)
-        self.assertEqual(case.type, 'player')
-        self.assertFalse(case.closed)
-        self.assertEqual(case.name, 'Elizabeth Harmon')
-        self.assertEqual(case.external_id, '1')
-        self.assertEqual(case.owner_id, 'methuen_home')
-        self.assertEqual(case.opened_by, self.web_user.user_id)
-        self.assertEqual(case.dynamic_case_properties(), {
+        assert case.domain == self.domain
+        assert case.type == 'player'
+        assert not case.closed
+        assert case.name == 'Elizabeth Harmon'
+        assert case.external_id == '1'
+        assert case.owner_id == 'methuen_home'
+        assert case.opened_by == self.web_user.user_id
+        assert case.dynamic_case_properties() == {
             'dob': '1948-11-02',
             'sport': 'chess',
-        })
+        }
 
         xform = XFormInstance.objects.get_form(res['form_id'])
-        self.assertEqual(xform.xmlns, 'http://commcarehq.org/case_api')
-        self.assertEqual(xform.metadata.userID, self.web_user.user_id)
-        self.assertEqual(xform.metadata.deviceID, 'user agent string')
+        assert xform.xmlns == 'http://commcarehq.org/case_api'
+        assert xform.metadata.userID == self.web_user.user_id
+        assert xform.metadata.deviceID == 'user agent string'
 
     def test_non_schema_updates(self):
         res = self._create_case({
@@ -136,8 +137,8 @@ def test_non_schema_updates(self):
             'bad_property': "this doesn't fit the schema!",
             'properties': {'sport': 'chess'},
         })
-        self.assertEqual(res.status_code, 400)
-        self.assertEqual(res.json()['error'], "'bad_property' is not a valid field.")
+        assert res.status_code == 400
+        assert res.json()['error'] == "'bad_property' is not a valid field."
 
     def test_empty_case_type(self):
         res = self._create_case({
@@ -146,8 +147,8 @@ def test_empty_case_type(self):
             'owner_id': 'methuen_home',
         }).json()
         case = CommCareCase.objects.get_case(res['case']['case_id'], self.domain)
-        self.assertEqual(case.name, 'Elizabeth Harmon')
-        self.assertEqual(case.type, '')
+        assert case.name == 'Elizabeth Harmon'
+        assert case.type == ''
 
     def test_no_required_updates(self):
         case = self._make_case()
@@ -155,17 +156,17 @@ def test_no_required_updates(self):
         res = self._update_case(case.case_id, {
             'properties': {'rank': '2100'}
         })
-        self.assertEqual(res.status_code, 200)
+        assert res.status_code == 200
 
         case = CommCareCase.objects.get_case(case.case_id, self.domain)
         # Nothing was zeroed out by being omitted
-        self.assertEqual(case.name, 'Elizabeth Harmon')
-        self.assertEqual(case.owner_id, 'methuen_home')
-        self.assertEqual(case.dynamic_case_properties(), {
+        assert case.name == 'Elizabeth Harmon'
+        assert case.owner_id == 'methuen_home'
+        assert case.dynamic_case_properties() == {
             'dob': '1948-11-02',
             'rank': '2100',
             'sport': 'chess',
-        })
+        }
 
     def test_update_case(self):
         case = self._make_case()
@@ -179,18 +180,18 @@ def test_update_case(self):
                 'champion': 'true',
             },
         }).json()
-        self.assertItemsEqual(res.keys(), ['case', 'form_id'])
+        assert set(res.keys()) == set(['case', 'form_id'])
 
         case = CommCareCase.objects.get_case(case.case_id, self.domain)
-        self.assertFalse(case.closed)
-        self.assertEqual(case.name, 'Beth Harmon')
-        self.assertEqual(case.owner_id, 'us_chess_federation')
-        self.assertEqual(case.dynamic_case_properties(), {
+        assert not case.closed
+        assert case.name == 'Beth Harmon'
+        assert case.owner_id == 'us_chess_federation'
+        assert case.dynamic_case_properties() == {
             'champion': 'true',
             'dob': '1948-11-02',
             'rank': '2100',
             'sport': 'chess',
-        })
+        }
 
     def test_can_update_case_type(self):
         case = self._make_case()
@@ -198,16 +199,16 @@ def test_can_update_case_type(self):
             'case_name': 'Beth Harmon',
             'case_type': 'legend',
         })
-        self.assertEqual(res.status_code, 200)
+        assert res.status_code == 200
         case = CommCareCase.objects.get_case(case.case_id, self.domain)
-        self.assertEqual(case.type, 'legend')
+        assert case.type == 'legend'
 
     def test_close_case(self):
         case = self._make_case()
         res = self._update_case(case.case_id, {'close': True})
-        self.assertEqual(res.status_code, 200)
+        assert res.status_code == 200
         case = CommCareCase.objects.get_case(case.case_id, self.domain)
-        self.assertTrue(case.closed)
+        assert case.closed
 
     def test_create_closed_case(self):
         res = self._create_case({
@@ -216,10 +217,10 @@ def test_create_closed_case(self):
             'owner_id': 'us_chess_federation',
             'close': True,
         })
-        self.assertEqual(res.status_code, 200)
+        assert res.status_code == 200
         case_id = res.json()['case']['case_id']
         case = CommCareCase.objects.get_case(case_id, self.domain)
-        self.assertTrue(case.closed)
+        assert case.closed
 
     def test_update_case_bad_id(self):
         res = self._update_case('notarealcaseid', {
@@ -230,8 +231,8 @@ def test_update_case_bad_id(self):
                 'champion': 'true',
             },
         })
-        self.assertEqual(res.json()['error'], "No case found with ID 'notarealcaseid'")
-        self.assertEqual(CommCareCase.objects.get_case_ids_in_domain(self.domain), [])
+        assert res.json()['error'] == "No case found with ID 'notarealcaseid'"
+        assert CommCareCase.objects.get_case_ids_in_domain(self.domain) == []
 
     def test_update_case_on_other_domain(self):
         case_id = str(uuid.uuid4())
@@ -246,8 +247,8 @@ def test_update_case_on_other_domain(self):
         res = self._update_case(case_id, {
             'owner_id': 'stealing_this_case',
         })
-        self.assertEqual(res.json()['error'], f"No case found with ID '{case_id}'")
-        self.assertEqual(CommCareCase.objects.get_case_ids_in_domain(self.domain), [])
+        assert res.json()['error'] == f"No case found with ID '{case_id}'"
+        assert CommCareCase.objects.get_case_ids_in_domain(self.domain) == []
 
     def test_create_child_case(self):
         parent_case = self._make_case()
@@ -267,18 +268,18 @@ def test_create_child_case(self):
                 },
             },
         }).json()
-        self.assertItemsEqual(res.keys(), ['case', 'form_id'])
+        assert set(res.keys()) == set(['case', 'form_id'])
 
         case = CommCareCase.objects.get_case(res['case']['case_id'], self.domain)
-        self.assertEqual(case.name, 'Harmon/Luchenko')
-        self.assertEqual(case.external_id, '23')
-        self.assertEqual(case.owner_id, 'harmon')
-        self.assertEqual(case.dynamic_case_properties(), {'winner': 'Harmon'})
-        self.assertEqual(case.indices[0].identifier, 'parent')
-        self.assertEqual(case.indices[0].referenced_id, parent_case.case_id)
-        self.assertEqual(case.indices[0].referenced_type, 'player')
-        self.assertEqual(case.indices[0].relationship, 'child')
-        self.assertEqual(case.indices[0].referenced_case.case_id, parent_case.case_id)
+        assert case.name == 'Harmon/Luchenko'
+        assert case.external_id == '23'
+        assert case.owner_id == 'harmon'
+        assert case.dynamic_case_properties() == {'winner': 'Harmon'}
+        assert case.indices[0].identifier == 'parent'
+        assert case.indices[0].referenced_id == parent_case.case_id
+        assert case.indices[0].referenced_type == 'player'
+        assert case.indices[0].relationship == 'child'
+        assert case.indices[0].referenced_case.case_id == parent_case.case_id
 
     def test_set_parent_by_external_id(self):
         parent_case = self._make_case()
@@ -294,10 +295,10 @@ def test_set_parent_by_external_id(self):
                 },
             },
         }).json()
-        self.assertItemsEqual(res.keys(), ['case', 'form_id'])
+        assert set(res.keys()) == set(['case', 'form_id'])
 
         case = CommCareCase.objects.get_case(res['case']['case_id'], self.domain)
-        self.assertEqual(case.indices[0].referenced_id, parent_case.case_id)
+        assert case.indices[0].referenced_id == parent_case.case_id
 
     def test_set_parent_by_bad_external_id(self):
         res = self._create_case({
@@ -312,7 +313,7 @@ def test_set_parent_by_bad_external_id(self):
                 },
             },
         }).json()
-        self.assertEqual(res['error'], "Could not find a case with external_id 'MISSING'")
+        assert res['error'] == "Could not find a case with external_id 'MISSING'"
 
     def test_set_parent_missing_field(self):
         parent_case = self._make_case()
@@ -328,14 +329,13 @@ def test_set_parent_missing_field(self):
                 },
             },
         }).json()
-        self.assertEqual(res['error'], "Property relationship is required when creating or updating case indices")
+        assert res['error'] == "Property relationship is required when creating or updating case indices"
 
     def test_delete_index(self):
         # aka remove child case
         parent_case = self._make_case()
         child_case = self._make_case(parent_id=parent_case.case_id)
-        self.assertEqual([c.case_id for c in parent_case.get_subcases()],
-                         [child_case.case_id])
+        assert [c.case_id for c in parent_case.get_subcases()] == [child_case.case_id]
 
         res = self._update_case(child_case.case_id, {
             'indices': {
@@ -346,13 +346,13 @@ def test_delete_index(self):
                 },
             },
         })
-        self.assertEqual(res.status_code, 200)
+        assert res.status_code == 200
 
         parent_case = CommCareCase.objects.get_case(parent_case.case_id, self.domain)
-        self.assertEqual(parent_case.get_subcases(), [])
+        assert parent_case.get_subcases() == []
 
         child_case = CommCareCase.objects.get_case(child_case.case_id, self.domain)
-        self.assertEqual(child_case.get_index('parent').referenced_id, '')
+        assert child_case.get_index('parent').referenced_id == ''
 
     def test_bulk_action(self):
         existing_case = self._make_case()
@@ -380,13 +380,13 @@ def test_bulk_action(self):
             },
         ]).json()
         #  only returns a single form ID - chunking should happen in the client
-        self.assertItemsEqual(res.keys(), ['cases', 'form_id'])
+        assert set(res.keys()) == set(['cases', 'form_id'])
 
         updated_case = CommCareCase.objects.get_case(existing_case.case_id, self.domain)
-        self.assertEqual(updated_case.name, 'Beth Harmon')
+        assert updated_case.name == 'Beth Harmon'
 
         new_case = CommCareCase.objects.get_case_by_external_id(self.domain, 'jolene')
-        self.assertEqual(new_case.name, 'Jolene')
+        assert new_case.name == 'Jolene'
 
     def test_bulk_without_create_flag(self):
         res = self._bulk_update_cases([{
@@ -395,9 +395,8 @@ def test_bulk_without_create_flag(self):
             'case_name': 'Jolene',
             'owner_id': 'methuen_home',
         }])
-        self.assertEqual(res.status_code, 400)
-        self.assertIn("A 'create' flag is required for each update.",
-                      res.json()['error'])
+        assert res.status_code == 400
+        assert "A 'create' flag is required for each update." in res.json()['error']
 
     def test_attempt_create_with_case_id(self):
         res = self._bulk_update_cases([{
@@ -407,20 +406,16 @@ def test_attempt_create_with_case_id(self):
             'owner_id': 'methuen_home',
             'case_id': 'somethingmalicious',
         }])
-        self.assertEqual(res.status_code, 400)
-        self.assertIn("You cannot specify case_id when creating a new case",
-                      res.json()['error'])
+        assert res.status_code == 400
+        assert "You cannot specify case_id when creating a new case" in res.json()['error']
 
     def test_bulk_update_too_big(self):
         res = self._bulk_update_cases([
             {'create': True, 'case_name': f'case {i}', 'case_type': 'player'}
             for i in range(103)
         ])
-        self.assertEqual(res.status_code, 400)
-        self.assertEqual(
-            res.json(),
-            {'error': "You cannot submit more than 100 updates in a single request"}
-        )
+        assert res.status_code == 400
+        assert res.json() == {'error': "You cannot submit more than 100 updates in a single request"}
 
     def test_update_with_bad_case_id(self):
         res = self._bulk_update_cases([
@@ -439,8 +434,8 @@ def test_update_with_bad_case_id(self):
                 'owner_id': 'methuen_home',
             },
         ])
-        self.assertEqual(res.json()['error'], "No case found with ID 'notarealcaseid'")
-        self.assertEqual(CommCareCase.objects.get_case_ids_in_domain(self.domain), [])
+        assert res.json()['error'] == "No case found with ID 'notarealcaseid'"
+        assert CommCareCase.objects.get_case_ids_in_domain(self.domain) == []
 
     def test_create_parent_and_child_together(self):
         res = self._bulk_update_cases([
@@ -471,10 +466,10 @@ def test_create_parent_and_child_together(self):
                 },
             },
         ])
-        self.assertEqual(res.status_code, 200)
+        assert res.status_code == 200
         parent = CommCareCase.objects.get_case_by_external_id(self.domain, 'beth')
         child = CommCareCase.objects.get_case_by_external_id(self.domain, 'harmon-luchenko')
-        self.assertEqual(parent.case_id, child.get_index('parent').referenced_id)
+        assert parent.case_id == child.get_index('parent').referenced_id
 
     def test_create_child_with_no_parent(self):
         res = self._bulk_update_cases([
@@ -496,8 +491,8 @@ def test_create_child_with_no_parent(self):
                 },
             },
         ])
-        self.assertEqual(res.status_code, 400)
-        self.assertEqual(res.json()['error'], "Could not find a case with temporary_id 'MISSING'")
+        assert res.status_code == 400
+        assert res.json()['error'] == "Could not find a case with temporary_id 'MISSING'"
 
     def test_index_reference_to_uncreated_external_id(self):
         res = self._bulk_update_cases([
@@ -524,10 +519,10 @@ def test_index_reference_to_uncreated_external_id(self):
                 },
             },
         ])
-        self.assertEqual(res.status_code, 200)
+        assert res.status_code == 200
         parent = CommCareCase.objects.get_case_by_external_id(self.domain, 'beth')
         child = CommCareCase.objects.get_case_by_external_id(self.domain, 'harmon-luchenko')
-        self.assertEqual(parent.case_id, child.get_index('parent').referenced_id)
+        assert parent.case_id == child.get_index('parent').referenced_id
 
     def test_update_by_external_id(self):
         case = self._make_case()
@@ -539,9 +534,9 @@ def test_update_by_external_id(self):
             },
             content_type="application/json;charset=utf-8",
         )
-        self.assertEqual(res.status_code, 200)
+        assert res.status_code == 200
         case = CommCareCase.objects.get_case(case.case_id, self.domain)
-        self.assertEqual(case.dynamic_case_properties().get('champion'), 'true')
+        assert case.dynamic_case_properties().get('champion') == 'true'
 
     def test_update_by_external_id_doesnt_exist(self):
         res = self.client.put(
@@ -552,8 +547,32 @@ def test_update_by_external_id_doesnt_exist(self):
             },
             content_type="application/json;charset=utf-8",
         )
-        self.assertEqual(res.status_code, 400)
-        self.assertEqual(res.json()['error'], "Could not find a case with external_id 'notarealcaseid'")
+        assert res.status_code == 400
+        assert res.json()['error'] == "Could not find a case with external_id 'notarealcaseid'"
+
+    def test_update_by_external_id_with_bad_create(self):
+        case = self._make_case()
+        res = self.client.put(
+            reverse('case_api', args=(self.domain,)),
+            {
+                'external_id': case.external_id,
+                'properties': {'champion': 'true'},
+                'create': True,  # Only valid in bulk operations
+            },
+            content_type="application/json;charset=utf-8",
+        )
+        assert res.status_code == 400
+        assert res.json()['error'] == "'create' is not a valid field."
+
+    def test_single_update_with_list(self):
+        case = self._make_case()
+        res = self.client.put(
+            reverse('case_api_detail_ext', args=(self.domain, case.external_id)),
+            [{'properties': {'champion': 'true'}}],
+            content_type="application/json;charset=utf-8",
+        )
+        assert res.status_code == 400
+        assert res.json()['error'] == 'Payload must be a single JSON object'
 
     def test_bulk_update_by_external_id(self):
         case = self._make_case()
@@ -562,9 +581,9 @@ def test_bulk_update_by_external_id(self):
             'external_id': case.external_id,
             'owner_id': 'us_chess_federation',
         }])
-        self.assertEqual(res.status_code, 200)
+        assert res.status_code == 200
         case = CommCareCase.objects.get_case(case.case_id, self.domain)
-        self.assertEqual(case.owner_id, 'us_chess_federation')
+        assert case.owner_id == 'us_chess_federation'
 
     def test_bulk_update_external_id_doesnt_exist(self):
         res = self._bulk_update_cases([{
@@ -572,13 +591,13 @@ def test_bulk_update_external_id_doesnt_exist(self):
             'external_id': 'notarealcaseid',
             'owner_id': 'us_chess_federation',
         }])
-        self.assertEqual(res.status_code, 400)
-        self.assertEqual(res.json()['error'], "Could not find a case with external_id 'notarealcaseid'")
+        assert res.status_code == 400
+        assert res.json()['error'] == "Could not find a case with external_id 'notarealcaseid'"
 
     def test_non_json_data(self):
         res = self._create_case("this isn't json")
-        self.assertEqual(res.status_code, 400)
-        self.assertEqual(res.json(), {'error': "Payload must be valid JSON"})
+        assert res.status_code == 400
+        assert res.json() == {'error': "Payload must be valid JSON"}
 
     def test_missing_required_field(self):
         res = self._create_case({
@@ -587,8 +606,8 @@ def test_missing_required_field(self):
             'owner_id': 'methuen_home',
             'properties': {'dob': '1948-11-02'},
         })
-        self.assertEqual(res.status_code, 400)
-        self.assertEqual(res.json(), {'error': "Property case_name is required."})
+        assert res.status_code == 400
+        assert res.json() == {'error': "Property case_name is required."}
 
     def test_invalid_properties(self):
         res = self._create_case({
@@ -600,10 +619,10 @@ def test_invalid_properties(self):
                 'age': 72,  # Can't pass integers
             },
         })
-        self.assertEqual(res.status_code, 400)
-        self.assertEqual(res.json(), {
+        assert res.status_code == 400
+        assert res.json() == {
             'error': "Error with case property 'age'. Values must be strings, received '72'"
-        })
+        }
 
     def test_non_xml_properties(self):
         res = self._create_case({
@@ -612,9 +631,9 @@ def test_non_xml_properties(self):
             'owner_id': 'methuen_home',
             'properties': {'not good': 'tsk tsk'},
         })
-        self.assertEqual(res.status_code, 400)
+        assert res.status_code == 400
         msg = "Error with case property 'not good'. Case property names must be valid XML identifiers."
-        self.assertEqual(res.json()['error'], msg)
+        assert res.json()['error'] == msg
 
     def test_non_xml_index_name(self):
         parent_case = self._make_case()
@@ -630,10 +649,10 @@ def test_non_xml_index_name(self):
                 },
             },
         })
-        self.assertEqual(res.status_code, 400)
+        assert res.status_code == 400
         msg = ("Error with index 'Robert'); DROP TABLE students;--'. "
                "Index names must be valid XML identifiers.")
-        self.assertEqual(res.json()['error'], msg)
+        assert res.json()['error'] == msg
 
     def test_bad_index_reference(self):
         res = self._create_case({
@@ -652,38 +671,40 @@ def test_bad_index_reference(self):
                 },
             },
         })
-        self.assertEqual(res.status_code, 400)
-        self.assertIn("InvalidCaseIndex", res.json()['error'])
+        assert res.status_code == 400
+        assert "InvalidCaseIndex" in res.json()['error']
         form = XFormInstance.objects.get_form(res.json()['form_id'])
-        self.assertEqual(form.is_error, True)
+        assert form.is_error
 
     def test_unset_external_id(self):
         case = self._make_case()
-        self.assertEqual(case.external_id, '1')
+        assert case.external_id == '1'
 
         res = self._update_case(case.case_id, {
             'external_id': '',
         })
-        self.assertEqual(res.status_code, 200)
+        assert res.status_code == 200
         case = CommCareCase.objects.get_case(case.case_id, self.domain)
-        self.assertEqual(case.external_id, '')
+        assert case.external_id == ''
 
     def test_omitting_external_id_doesnt_clear_it(self):
         case = self._make_case()
-        self.assertEqual(case.external_id, '1')
+        assert case.external_id == '1'
 
         res = self._update_case(case.case_id, {
             'properties': {'champion': 'true'},
         })
-        self.assertEqual(res.status_code, 200)
+        assert res.status_code == 200
         case = CommCareCase.objects.get_case(case.case_id, self.domain)
-        self.assertEqual(case.external_id, '1')
+        assert case.external_id == '1'
 
     def test_urls_without_trailing_slash(self):
         case_id = self._make_case().case_id
         urls = [
             (self.client.post, reverse('case_api', args=(self.domain,)).rstrip("/")),
-            (self.client.put, reverse('case_api', args=(self.domain, case_id)).rstrip("/")),
+            (self.client.put, reverse('case_api_detail', args=(self.domain, case_id)).rstrip("/")),
+            (self.client.post, reverse('case_api_v0.6', args=(self.domain,)).rstrip("/")),
+            (self.client.put, reverse('case_api_v0.6_detail', args=(self.domain, case_id)).rstrip("/")),
         ]
         for request_fn, url in urls:
             res = request_fn(
@@ -693,7 +714,7 @@ def test_urls_without_trailing_slash(self):
                 HTTP_USER_AGENT="user agent string",
             )
             # These requests should return a 400 because of the bad body, not a 301 redirect
-            self.assertEqual(res.status_code, 400)
+            assert res.status_code == 400
 
     def test_case_name_twice(self):
         case = self._make_case()
@@ -706,6 +727,64 @@ def test_case_name_twice(self):
                 'champion': 'true',
             },
         })
-        self.assertEqual(res.status_code, 400)
+        assert res.status_code == 400
         msg = "Error with case property 'case_name'. This must be specified at the top level."
-        self.assertEqual(res.json()['error'], msg)
+        assert res.json()['error'] == msg
+
+    @patch("corehq.apps.hqcase.views.handle_case_update")
+    def test_post_without_external_id_calls_with_is_creation_true(self, mock_handle_update):
+        mock_handle_update.return_value = self._get_update_return()
+
+        with patch(
+            "corehq.apps.hqcase.views.serialize_case",
+            return_value={"case_id": "new-case"},
+        ):
+            response = self.client.post(
+                reverse("case_api", args=(self.domain,)),
+                json.dumps(
+                    {
+                        "case_type": "patient",
+                        "case_name": "Jane Doe",
+                    }
+                ),
+                content_type="application/json",
+                HTTP_USER_AGENT="test-agent",
+            )
+
+        assert response.status_code == 200
+        mock_handle_update.assert_called_once()
+        call_kwargs = mock_handle_update.call_args[1]
+        assert call_kwargs["is_creation"]
+
+    @patch("corehq.apps.hqcase.views.handle_case_update")
+    def test_put_with_case_id_calls_with_is_creation_false(self, mock_handle_update):
+        mock_handle_update.return_value = self._get_update_return()
+
+        with patch(
+            "corehq.apps.hqcase.views.serialize_case",
+            return_value={"case_id": "case-123"},
+        ):
+            response = self.client.put(
+                f"/a/{self.domain}/api/case/v2/case-123/",
+                json.dumps(
+                    {
+                        "case_name": "Updated Name",
+                        "external_id": "ext-456",
+                    }
+                ),
+                content_type="application/json",
+                HTTP_USER_AGENT="test-agent",
+            )
+
+        assert response.status_code == 200
+        mock_handle_update.assert_called_once()
+        call_kwargs = mock_handle_update.call_args[1]
+        assert not call_kwargs["is_creation"]
+
+    @staticmethod
+    def _get_update_return():
+        """Helper method to create mock return value for handle_case_update"""
+        mock_xform = MagicMock(spec=XFormInstance)
+        mock_xform.form_id = "test-form-id"
+        mock_case = MagicMock(spec=CommCareCase)
+        return mock_xform, mock_case
diff --git a/corehq/apps/hqcase/tests/test_external_id_url_encoding.py b/corehq/apps/hqcase/tests/test_external_id_url_encoding.py
new file mode 100644
index 000000000000..65c2229f7a5f
--- /dev/null
+++ b/corehq/apps/hqcase/tests/test_external_id_url_encoding.py
@@ -0,0 +1,85 @@
+"""
+Test that the path converter allows any characters in the Case API
+`external_id` parameter.
+"""
+from urllib.parse import unquote
+
+from django.urls import resolve
+
+import pytest
+
+from corehq.apps.hqcase.views import case_api
+
+
+@pytest.mark.parametrize('url_path,expected_external_id', [
+    ('hello-world', 'hello-world'),
+    ('hello%20world', 'hello world'),
+    ('hello%40world', 'hello@world'),
+    ('hello%23world', 'hello#world'),
+    ('hello%3Fworld', 'hello?world'),
+    ('hello%26world', 'hello&world'),
+    ('hello%3Dworld', 'hello=world'),
+    ('hello%2Bworld', 'hello+world'),
+    ('hello%2Fworld', 'hello/world'),
+    ('hello%E2%80%94world%2F', 'hello—world/'),
+])
+def test_external_id_url_decoding_v2(url_path, expected_external_id):
+    url = f'/a/test-domain/api/case/v2/ext/{url_path}/'
+    match = resolve(url)
+
+    assert match.func == case_api
+    assert match.kwargs['domain'] == 'test-domain'
+    assert unquote(match.kwargs['external_id']) == expected_external_id
+
+
+@pytest.mark.parametrize('url_path,expected_external_id', [
+    ('hello-world', 'hello-world'),
+    ('hello%2Fworld', 'hello/world'),
+    ('hello%E2%80%94world%2F', 'hello—world/'),
+])
+def test_external_id_url_decoding_v0_6(url_path, expected_external_id):
+    url = f'/a/test-domain/api/v0.6/case/ext/{url_path}/'
+    match = resolve(url)
+
+    assert match.func == case_api
+    assert match.kwargs['domain'] == 'test-domain'
+    assert unquote(match.kwargs['external_id']) == expected_external_id
+
+
+@pytest.mark.parametrize('url,expected_case_id', [
+    # v2 endpoints
+    ('/a/test-domain/api/case/v2/', None),
+    ('/a/test-domain/api/case/v2/simple-case-id/', 'simple-case-id'),
+    ('/a/test-domain/api/case/v2/simple-case-id', 'simple-case-id'),  # no trailing slash
+    ('/a/test-domain/api/case/v2/case-with-uuid-123/', 'case-with-uuid-123'),
+    ('/a/test-domain/api/case/v2/id1,id2,id3/', 'id1,id2,id3'),  # commas supported
+    # v0.6 endpoints
+    ('/a/test-domain/api/v0.6/case/', None),
+    ('/a/test-domain/api/v0.6/case/simple-case-id/', 'simple-case-id'),
+    ('/a/test-domain/api/v0.6/case/simple-case-id', 'simple-case-id'),
+    ('/a/test-domain/api/v0.6/case/id1,id2,id3/', 'id1,id2,id3'),  # commas supported
+])
+def test_case_id_url_patterns(url, expected_case_id):
+    """Test that case_id URL patterns work with and without case_id parameter"""
+    match = resolve(url)
+
+    assert match.func == case_api
+    assert match.kwargs['domain'] == 'test-domain'
+    assert match.kwargs.get('case_id') == expected_case_id
+
+
+@pytest.mark.parametrize('url', [
+    # Invalid case_id values should not match the pattern [\w\-,]+
+    '/a/test-domain/api/case/v2/id with spaces/',
+    '/a/test-domain/api/case/v2/id@special/',
+    '/a/test-domain/api/case/v2/id#hash/',
+    '/a/test-domain/api/case/v2/id/with/slash/',
+    '/a/test-domain/api/v0.6/case/id with spaces/',
+    '/a/test-domain/api/v0.6/case/id@special/',
+])
+def test_case_id_rejects_invalid_characters(url):
+    """Test that case_id pattern rejects invalid characters"""
+    from django.urls.exceptions import Resolver404
+
+    with pytest.raises(Resolver404):
+        resolve(url)
diff --git a/corehq/apps/hqcase/views.py b/corehq/apps/hqcase/views.py
index 145e980d82b5..134617377fe6 100644
--- a/corehq/apps/hqcase/views.py
+++ b/corehq/apps/hqcase/views.py
@@ -91,15 +91,19 @@ def delete_cases(self, request, domain):
 @requires_privilege_with_fallback(privileges.API_ACCESS)
 @api_throttle
 @location_safe
-def case_api(request, domain, case_id=None):
+def case_api(request, domain, case_id=None, external_id=None):
     if request.method == 'GET' and case_id:
         return _handle_get(request, case_id)
+    if request.method == 'GET' and external_id:
+        return _handle_ext_get(request, external_id)
     if request.method == 'GET' and not case_id:
         return _handle_list_view(request)
     if request.method == 'POST' and not case_id:
-        return _handle_case_update(request, is_creation=True)
+        return _handle_case_put_post(request, is_creation=True)
+    if request.method == 'PUT' and external_id:
+        return _handle_ext_put(request, external_id)
     if request.method == 'PUT':
-        return _handle_case_update(request, is_creation=False, case_id=case_id)
+        return _handle_case_put_post(request, is_creation=False, case_id=case_id)
     return JsonResponse({'error': "Request method not allowed"}, status=405)
 
 
@@ -144,6 +148,25 @@ def _get_single_case(request, case_id):
     return JsonResponse(serialize_es_case(case))
 
 
+def _handle_ext_get(request, external_id):
+    try:
+        bulk_fetch_results_dict = get_bulk(
+            request.domain,
+            request.couch_user,
+            case_ids=[],
+            external_ids=[external_id],
+        )
+    except UserError as err:
+        return JsonResponse({'error': str(err)}, status=400)
+    case = bulk_fetch_results_dict['cases'][0]
+    if case.get('error') == 'not found':
+        return JsonResponse(
+            {'error': f"Case '{external_id}' not found"},
+            status=404,
+        )
+    return JsonResponse(case)
+
+
 def _handle_bulk_fetch(request):
     try:
         data = json.loads(request.body.decode('utf-8'))
@@ -169,7 +192,48 @@ def _handle_list_view(request):
     return JsonResponse(res)
 
 
-def _handle_case_update(request, is_creation, case_id=None):
+def _handle_ext_put(request, external_id):
+    try:
+        data = json.loads(request.body.decode('utf-8'))
+    except (UnicodeDecodeError, json.JSONDecodeError):
+        return JsonResponse({'error': "Payload must be valid JSON"}, status=400)
+    if not isinstance(data, dict):
+        return JsonResponse(
+            {'error': "Payload must be a single JSON object"},
+            status=400,
+        )
+    if 'external_id' not in data:
+        data['external_id'] = external_id
+
+    try:
+        bulk_fetch_results_dict = get_bulk(
+            request.domain,
+            request.couch_user,
+            case_ids=[],
+            external_ids=[external_id],
+        )
+    except UserError as err:
+        return JsonResponse({'error': str(err)}, status=400)
+    case = bulk_fetch_results_dict['cases'][0]
+    if case.get('error') == 'not found':
+        is_creation = True
+    else:
+        is_creation = False
+        if 'case_id' not in data:
+            data['case_id'] = case['case_id']
+        elif data['case_id'] != case['case_id']:
+            return JsonResponse(
+                {'error': 'The given value of "case_id" does not match the '
+                          'existing value for the case identified by '
+                          f'external_id = "{external_id}". "case_id" is '
+                          'read-only and cannot be modified.'},
+                status=400,
+            )
+
+    return _handle_case_update(request, data, is_creation)
+
+
+def _handle_case_put_post(request, is_creation, case_id=None):
     try:
         data = json.loads(request.body.decode('utf-8'))
     except (UnicodeDecodeError, json.JSONDecodeError):
@@ -178,6 +242,10 @@ def _handle_case_update(request, is_creation, case_id=None):
     if not is_creation and case_id and 'case_id' not in data:
         data['case_id'] = case_id
 
+    return _handle_case_update(request, data, is_creation)
+
+
+def _handle_case_update(request, data, is_creation):
     try:
         xform, case_or_cases = handle_case_update(
             domain=request.domain,
diff --git a/docs/api/cases-v2.rst b/docs/api/cases-v2.rst
index 0ada0c97dab2..7730863d23b5 100644
--- a/docs/api/cases-v2.rst
+++ b/docs/api/cases-v2.rst
@@ -498,6 +498,27 @@ xform_id  ID of the (single) form generated to update all cases
 cases     Serialized version of the new state of the cases provided
 ========= =========================================================
 
+Upsert
+^^^^^^
+
+If a case update includes a value for "external_id" and does not include
+a value for "case_id", then the "create" field may be omitted, and the
+API will upsert the case based on the value of "external_id". This
+operation is slower than if the "create" field is included. Ensure that
+the upserted cases in a single request are unique, and that concurrent
+requests cannot include the same cases.
+
+.. DANGER::
+   If a single request tries to upsert the same case more than once,
+   **duplicate cases will be created!** This is because CommCare HQ
+   searches for all upserted cases by their external IDs before any
+   changes are made.
+
+.. WARNING::
+   This operation is vulnerable to a race condition where, if two
+   simultaneous requests try to upsert the same case, two cases could be
+   created.
+
 .. _limitations-1:
 
 Limitations