diff --git a/changelogs/fragments/87-docker_image-load-image-ids.yml b/changelogs/fragments/87-docker_image-load-image-ids.yml
new file mode 100644
index 000000000..098b5d733
--- /dev/null
+++ b/changelogs/fragments/87-docker_image-load-image-ids.yml
@@ -0,0 +1,7 @@
+minor_changes:
+- "docker_image - properly support image IDs (hashes) for loading and tagging images (https://github.com/ansible-collections/community.docker/issues/86, https://github.com/ansible-collections/community.docker/pull/87)."
+bugfixes:
+- "docker_image - prevent module failure when removing image that is removed between inspection and removal (https://github.com/ansible-collections/community.docker/pull/87)."
+- "docker_image - prevent module failure when removing non-existant image by ID (https://github.com/ansible-collections/community.docker/pull/87)."
+- "docker_image_info - prevent module failure when image vanishes between listing and inspection (https://github.com/ansible-collections/community.docker/pull/87)."
+- "docker_image_info - prevent module failure when querying non-existant image by ID (https://github.com/ansible-collections/community.docker/pull/87)."
diff --git a/plugins/module_utils/common.py b/plugins/module_utils/common.py
index 9373d25ea..e2a799187 100644
--- a/plugins/module_utils/common.py
+++ b/plugins/module_utils/common.py
@@ -534,6 +534,9 @@ def find_image(self, name, tag):
         if len(images) == 1:
             try:
                 inspection = self.inspect_image(images[0]['Id'])
+            except NotFound:
+                self.log("Image %s:%s not found." % (name, tag))
+                return None
             except Exception as exc:
                 self.fail("Error inspecting image %s:%s - %s" % (name, tag, str(exc)))
             return inspection
@@ -541,7 +544,7 @@ def find_image(self, name, tag):
         self.log("Image %s:%s not found." % (name, tag))
         return None
 
-    def find_image_by_id(self, image_id):
+    def find_image_by_id(self, image_id, accept_missing_image=False):
         '''
         Lookup an image (by ID) and return the inspection results.
         '''
@@ -551,6 +554,11 @@ def find_image_by_id(self, image_id):
         self.log("Find image %s (by ID)" % image_id)
         try:
             inspection = self.inspect_image(image_id)
+        except NotFound as exc:
+            if not accept_missing_image:
+                self.fail("Error inspecting image ID %s - %s" % (image_id, str(exc)))
+            self.log("Image %s not found." % image_id)
+            return None
         except Exception as exc:
             self.fail("Error inspecting image ID %s - %s" % (image_id, str(exc)))
         return inspection
diff --git a/plugins/modules/docker_image.py b/plugins/modules/docker_image.py
index b4d1b0fb4..af5b704aa 100644
--- a/plugins/modules/docker_image.py
+++ b/plugins/modules/docker_image.py
@@ -155,9 +155,9 @@
     default: false
   name:
     description:
-      - "Image name. Name format will be one of: name, repository/name, registry_server:port/name.
-        When pushing or pulling an image the name can optionally include the tag by appending ':tag_name'."
-      - Note that image IDs (hashes) are not supported.
+      - "Image name. Name format will be one of: C(name), C(repository/name), C(registry_server:port/name).
+        When pushing or pulling an image the name can optionally include the tag by appending C(:tag_name)."
+      - Note that image IDs (hashes) are only supported for I(state=absent), and for I(state=present) with I(source=load).
     type: str
     required: yes
   pull:
@@ -344,7 +344,7 @@
         else:
             from docker.auth.auth import resolve_repository_name
         from docker.utils.utils import parse_repository_tag
-        from docker.errors import DockerException
+        from docker.errors import DockerException, NotFound
     except ImportError:
         # missing Docker SDK for Python handled in module_utils.docker.common
         pass
@@ -397,6 +397,12 @@ def __init__(self, client, results):
                 self.name = repo
                 self.tag = repo_tag
 
+        # Sanity check: fail early when we know that something will fail later
+        if self.repository and is_image_name_id(self.repository):
+            self.fail("`repository` must not be an image ID; got: %s" % self.repository)
+        if not self.repository and self.push and is_image_name_id(self.name):
+            self.fail("Cannot push an image by ID; specify `repository` to tag and push the image with ID %s instead" % self.name)
+
         if self.state == 'present':
             self.present()
         elif self.state == 'absent':
@@ -412,10 +418,16 @@ def present(self):
 
         :returns None
         '''
-        image = self.client.find_image(name=self.name, tag=self.tag)
+        if is_image_name_id(self.name):
+            image = self.client.find_image_by_id(self.name, accept_missing_image=True)
+        else:
+            image = self.client.find_image(name=self.name, tag=self.tag)
 
         if not image or self.force_source:
             if self.source == 'build':
+                if is_image_name_id(self.name):
+                    self.fail("Image name must not be an image ID for source=build; got: %s" % self.name)
+
                 # Build the image
                 if not os.path.isdir(self.build_path):
                     self.fail("Requested build path %s could not be found or you do not have access." % self.build_path)
@@ -434,13 +446,16 @@ def present(self):
                     self.fail("Error loading image %s. Specified path %s does not exist." % (self.name,
                                                                                              self.load_path))
                 image_name = self.name
-                if self.tag:
+                if self.tag and not is_image_name_id(image_name):
                     image_name = "%s:%s" % (self.name, self.tag)
                 self.results['actions'].append("Loaded image %s from %s" % (image_name, self.load_path))
                 self.results['changed'] = True
                 if not self.check_mode:
                     self.results['image'] = self.load_image()
             elif self.source == 'pull':
+                if is_image_name_id(self.name):
+                    self.fail("Image name must not be an image ID for source=pull; got: %s" % self.name)
+
                 # pull the image
                 self.results['actions'].append('Pulled image %s:%s' % (self.name, self.tag))
                 self.results['changed'] = True
@@ -449,11 +464,13 @@ def present(self):
             elif self.source == 'local':
                 if image is None:
                     name = self.name
-                    if self.tag:
+                    if self.tag and not is_image_name_id(name):
                         name = "%s:%s" % (self.name, self.tag)
                     self.client.fail('Cannot find the image %s locally.' % name)
             if not self.check_mode and image and image['Id'] == self.results['image']['Id']:
                 self.results['changed'] = False
+        else:
+            self.results['image'] = image
 
         if self.archive_path:
             self.archive_image(self.name, self.tag)
@@ -471,7 +488,7 @@ def absent(self):
         '''
         name = self.name
         if is_image_name_id(name):
-            image = self.client.find_image_by_id(name)
+            image = self.client.find_image_by_id(name, accept_missing_image=True)
         else:
             image = self.client.find_image(name, self.tag)
             if self.tag:
@@ -480,6 +497,9 @@ def absent(self):
             if not self.check_mode:
                 try:
                     self.client.remove_image(name, force=self.force_absent)
+                except NotFound:
+                    # If the image vanished while we were trying to remove it, don't fail
+                    pass
                 except Exception as exc:
                     self.fail("Error removing image %s - %s" % (name, str(exc)))
 
@@ -498,33 +518,37 @@ def archive_image(self, name, tag):
         if not tag:
             tag = "latest"
 
-        image = self.client.find_image(name=name, tag=tag)
+        if is_image_name_id(name):
+            image = self.client.find_image_by_id(name, accept_missing_image=True)
+            image_name = name
+        else:
+            image = self.client.find_image(name=name, tag=tag)
+            image_name = "%s:%s" % (name, tag)
+
         if not image:
-            self.log("archive image: image %s:%s not found" % (name, tag))
+            self.log("archive image: image %s not found" % image_name)
             return
 
-        image_name = "%s:%s" % (name, tag)
         self.results['actions'].append('Archived image %s to %s' % (image_name, self.archive_path))
         self.results['changed'] = True
         if not self.check_mode:
             self.log("Getting archive of image %s" % image_name)
             try:
-                image = self.client.get_image(image_name)
+                saved_image = self.client.get_image(image_name)
             except Exception as exc:
                 self.fail("Error getting image %s - %s" % (image_name, str(exc)))
 
             try:
                 with open(self.archive_path, 'wb') as fd:
                     if self.client.docker_py_version >= LooseVersion('3.0.0'):
-                        for chunk in image:
+                        for chunk in saved_image:
                             fd.write(chunk)
                     else:
-                        for chunk in image.stream(2048, decode_content=False):
+                        for chunk in saved_image.stream(2048, decode_content=False):
                             fd.write(chunk)
             except Exception as exc:
                 self.fail("Error writing image archive %s - %s" % (self.archive_path, str(exc)))
 
-        image = self.client.find_image(name=name, tag=tag)
         if image:
             self.results['image'] = image
 
@@ -537,6 +561,9 @@ def push_image(self, name, tag=None):
         :return: None
         '''
 
+        if is_image_name_id(name):
+            self.fail("Cannot push an image ID: %s" % name)
+
         repository = name
         if not tag:
             repository, tag = parse_repository_tag(name)
@@ -624,7 +651,7 @@ def _extract_output_line(line, output):
             # Make sure we have a string (assuming that line['stream'] and
             # line['status'] are either not defined, falsish, or a string)
             text_line = line.get('stream') or line.get('status') or ''
-            output.append(text_line)
+            output.extend(text_line.splitlines())
 
     def build_image(self):
         '''
@@ -745,27 +772,39 @@ def load_image(self):
         if has_output:
             # We can only do this when we actually got some output from Docker daemon
             loaded_images = set()
+            loaded_image_ids = set()
             for line in load_output:
                 if line.startswith('Loaded image:'):
                     loaded_images.add(line[len('Loaded image:'):].strip())
+                if line.startswith('Loaded image ID:'):
+                    loaded_image_ids.add(line[len('Loaded image ID:'):].strip().lower())
 
-            if not loaded_images:
+            if not loaded_images and not loaded_image_ids:
                 self.client.fail("Detected no loaded images. Archive potentially corrupt?", stdout='\n'.join(load_output))
 
-            expected_image = '%s:%s' % (self.name, self.tag)
-            if expected_image not in loaded_images:
+            if is_image_name_id(self.name):
+                expected_image = self.name.lower()
+                found_image = expected_image not in loaded_image_ids
+            else:
+                expected_image = '%s:%s' % (self.name, self.tag)
+                found_image = expected_image not in loaded_images
+            if found_image:
                 self.client.fail(
                     "The archive did not contain image '%s'. Instead, found %s." % (
-                        expected_image, ', '.join(["'%s'" % image for image in sorted(loaded_images)])),
+                        expected_image,
+                        ', '.join(sorted(["'%s'" % image for image in loaded_images] + list(loaded_image_ids)))),
                     stdout='\n'.join(load_output))
             loaded_images.remove(expected_image)
 
             if loaded_images:
                 self.client.module.warn(
                     "The archive contained more images than specified: %s" % (
-                        ', '.join(["'%s'" % image for image in sorted(loaded_images)]), ))
+                        ', '.join(sorted(["'%s'" % image for image in loaded_images] + list(loaded_image_ids))), ))
 
-        return self.client.find_image(self.name, self.tag)
+        if is_image_name_id(self.name):
+            return self.client.find_image_by_id(self.name, accept_missing_image=True)
+        else:
+            return self.client.find_image(self.name, self.tag)
 
 
 def main():
diff --git a/plugins/modules/docker_image_info.py b/plugins/modules/docker_image_info.py
index 5d855fa29..86c7b093d 100644
--- a/plugins/modules/docker_image_info.py
+++ b/plugins/modules/docker_image_info.py
@@ -167,7 +167,7 @@
 
 try:
     from docker import utils
-    from docker.errors import DockerException
+    from docker.errors import DockerException, NotFound
 except ImportError:
     # missing Docker SDK for Python handled in ansible.module_utils.docker.common
     pass
@@ -215,7 +215,7 @@ def get_facts(self):
         for name in names:
             if is_image_name_id(name):
                 self.log('Fetching image %s (ID)' % (name))
-                image = self.client.find_image_by_id(name)
+                image = self.client.find_image_by_id(name, accept_missing_image=True)
             else:
                 repository, tag = utils.parse_repository_tag(name)
                 if not tag:
@@ -232,6 +232,8 @@ def get_all_images(self):
         for image in images:
             try:
                 inspection = self.client.inspect_image(image['Id'])
+            except NotFound:
+                pass
             except Exception as exc:
                 self.fail("Error inspecting image %s - %s" % (image['Id'], str(exc)))
             results.append(inspection)
diff --git a/tests/integration/targets/docker_image/tasks/tests/basic.yml b/tests/integration/targets/docker_image/tasks/tests/basic.yml
index 374ea6afb..acb45ea0b 100644
--- a/tests/integration/targets/docker_image/tasks/tests/basic.yml
+++ b/tests/integration/targets/docker_image/tasks/tests/basic.yml
@@ -70,13 +70,66 @@
     force_tag: yes
   register: tag_3
 
+- name: Tag image with ID instead of name
+  docker_image:
+    source: local
+    name: "{{ present_1.image.Id }}"
+    repository: "{{ docker_test_image_hello_world_base }}:alias"
+  register: tag_4
+
 - assert:
     that:
     - tag_1 is changed
     - tag_2 is not changed
     - tag_3 is not changed
+    - tag_4 is not changed
 
 - name: Cleanup alias tag
   docker_image:
     name: "{{ docker_test_image_hello_world_base }}:alias"
     state: absent
+
+- name: Tag image with ID instead of name (use ID for repository, must fail)
+  docker_image:
+    source: local
+    name: "{{ docker_test_image_hello_world }}"
+    repository: "{{ present_1.image.Id }}"
+  register: fail_1
+  ignore_errors: true
+
+- name: Push image with ID (must fail)
+  docker_image:
+    source: local
+    name: "{{ present_1.image.Id }}"
+    push: true
+  register: fail_2
+  ignore_errors: true
+
+- name: Pull image ID (must fail)
+  docker_image:
+    source: pull
+    name: "{{ present_1.image.Id }}"
+    force_source: true
+  register: fail_3
+  ignore_errors: true
+
+- name: buildargs
+  docker_image:
+    source: build
+    name: "{{ present_1.image.Id }}"
+    build:
+      path: "{{ output_dir }}/files"
+    force_source: true
+  register: fail_4
+  ignore_errors: yes
+
+- assert:
+    that:
+    - fail_1 is failed
+    - "'`repository` must not be an image ID' in fail_1.msg"
+    - fail_2 is failed
+    - "'Cannot push an image by ID' in fail_2.msg"
+    - fail_3 is failed
+    - "'Image name must not be an image ID for source=pull' in fail_3.msg"
+    - fail_4 is failed
+    - "'Image name must not be an image ID for source=build' in fail_4.msg"
diff --git a/tests/integration/targets/docker_image/tasks/tests/options.yml b/tests/integration/targets/docker_image/tasks/tests/options.yml
index a966ea2c1..246f95fee 100644
--- a/tests/integration/targets/docker_image/tasks/tests/options.yml
+++ b/tests/integration/targets/docker_image/tasks/tests/options.yml
@@ -241,6 +241,13 @@
     source: pull
   register: archive_image
 
+- name: Archive image by ID
+  docker_image:
+    name: "{{ archive_image.image.Id }}"
+    archive_path: "{{ output_dir }}/image_id.tar"
+    source: pull
+  register: archive_image_id
+
 - name: Create invalid archive
   copy:
     dest: "{{ output_dir }}/image-invalid.tar"
@@ -290,6 +297,13 @@
     api_version: "1.22"
   register: load_image_4
 
+- name: load image (ID, idempotency)
+  docker_image:
+    name: "{{ archive_image.image.Id }}"
+    load_path: "{{ output_dir }}/image_id.tar"
+    source: load
+  register: load_image_5
+
 - assert:
     that:
     - load_image is changed
@@ -302,6 +316,7 @@
     - '"Detected no loaded images. Archive potentially corrupt?" == load_image_3.msg'
     - load_image_4 is changed
     - "'The API version of your Docker daemon is < 1.23, which does not return the image loading result from the Docker daemon. Therefore, we cannot verify whether the expected image was loaded, whether multiple images where loaded, or whether the load actually succeeded. You should consider upgrading your Docker daemon.' in load_image_4.warnings"
+    - load_image_5 is not changed
 
 ####################################################################
 ## build.path ######################################################