diff --git a/CHANGELOG.md b/CHANGELOG.md
index 8347c2b25..8aa644cac 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -41,6 +41,8 @@ The types of changes are:
 * Data seeding for Datadog access tests [#1269](https://github.com/ethyca/fidesops/pull/1269)
 * Added support for one-to-many relationships for param_values in SaaS configs [#1253](https://github.com/ethyca/fidesops/pull/1253)
 * Added erasure endpoints for Shopify connector [#1289](https://github.com/ethyca/fidesops/pull/1289)
+* Adds ability to send email notification upon privacy request completion [#1282](https://github.com/ethyca/fidesops/pull/1282)
+
 ### Docs
 
 * Fix analytics opt out environment variable name [#1170](https://github.com/ethyca/fidesops/pull/1170)
diff --git a/data/config/fidesops.toml b/data/config/fidesops.toml
index 49a64bb02..6dd94aee2 100644
--- a/data/config/fidesops.toml
+++ b/data/config/fidesops.toml
@@ -38,3 +38,6 @@ analytics_id = "internal"
 
 [admin_ui]
 enabled = true
+
+[notifications]
+send_request_completion_notification = true
diff --git a/docs/fidesops/docs/guides/configuration_reference.md b/docs/fidesops/docs/guides/configuration_reference.md
index fa1dcb659..d4287860d 100644
--- a/docs/fidesops/docs/guides/configuration_reference.md
+++ b/docs/fidesops/docs/guides/configuration_reference.md
@@ -47,6 +47,7 @@ The `fidesops.toml` file should specify the following variables:
 | `root_username` | `FIDESOPS__SECURITY__ROOT_USERNAME` | string | root_user | None | If set this can be used in conjunction with `root_password` to log in as a root user without first needing to create a user in the database. |
 | `root_password` | `FIDESOPS__SECURITY__ROOT_PASSWORD` | string | apassword | None | If set this can be used in conjunction with `root_username` to log in as a root user without first needing to create a user in the database. |
 | `root_user_scopes` | `FIDESOPS__SECURITY__ROOT_USER_SCOPES` | list of strings | ["client:create", "client:update"] | All available scopes | The scopes granted to the root user when logging in with `root_username` and `root_password`. |
+| `subject_request_download_link_ttl_seconds` | `FIDESOPS__SECURITY__SUBJECT_REQUEST_DOWNLOAD_LINK_TTL_SECONDS` | int | 86400 | 86400 | Time in seconds for a subject data package download link to remain valid, default to 1 day. |
 | Execution Variables |---|---|---|---|---|
 |`privacy_request_delay_timeout` | `FIDESOPS__EXECUTION__PRIVACY_REQUEST_DELAY_TIMEOUT` | int | 3600 | 3600 | The amount of time to wait for actions delaying privacy requests, for example pre and post processing webhooks.
 |`task_retry_count` | `FIDESOPS__EXECUTION__TASK_RETRY_COUNT` | int | 5 | 0 | The number of times a failed request will be retried
@@ -61,6 +62,9 @@ The `fidesops.toml` file should specify the following variables:
 |`analytics_opt_out` | `FIDESOPS__ROOT_USER__ANALYTICS_OPT_OUT` | bool | False | False | Opt out of sending anonymous usage data to Ethyca to improve the product experience.
 | Admin UI Variables|---|---|---|---|---|
 |`enabled` | `FIDESOPS__ADMIN_UI__ENABLED` | bool | False | True | Toggle whether the Admin UI is served from `/`
+| Fidesops Notification Variables|---|---|---|---|---|
+|`send_request_completion_notification` | `FIDESOPS__NOTIFICATIONS__SEND_REQUEST_COMPLETION_NOTIFICATION` | bool | True | True | Whether a notification will be sent to data subjects upon privacy request completion
+
 
 ### An example `fidesops.toml` configuration file
 
@@ -93,6 +97,7 @@ oauth_root_client_secret = "fidesopsadminsecret"
 log_level = "INFO"
 root_username = "root_user"
 root_password = "Testpassword1!"
+subject_request_download_link_ttl_seconds = 86400
 
 [execution]
 masking_strict = true
@@ -110,6 +115,9 @@ analytics_opt_out = false
 
 [admin_ui]
 enabled = true
+
+[notifications]
+send_request_completion_notification = true
 ```
 
 Note: The configuration is case-sensitive, so the variables must be specified in `lowercase`.
diff --git a/docs/fidesops/docs/guides/email_communications.md b/docs/fidesops/docs/guides/email_communications.md
index 4b82444e6..2a05da4e3 100644
--- a/docs/fidesops/docs/guides/email_communications.md
+++ b/docs/fidesops/docs/guides/email_communications.md
@@ -5,8 +5,9 @@ Fidesops supports configuring third party email servers to handle outbound commu
 
 Supported modes of use:
 
-- Subject Identity Verification - sends a verification code to the user's email address prior to processing a subject request. for more information on identity verification, see the [Privacy Requests](privacy_requests.md#subject-identity-verification) guide.
+- Subject Identity Verification - sends a verification code to the user's email address prior to processing a subject request. For more information on identity verification, see the [Privacy Requests](privacy_requests.md#subject-identity-verification) guide.
 - Erasure Request Email Fulfillment - sends an email to configured third parties to process erasures for a given data subject.  See [creating email Connectors](#email-third-party-services) for more information.
+- Privacy Request Completion Notification - sends an email to user's email address with privacy request completion notification, including a download link to data package, for access requests. For more information on request completion notification, see the [Privacy Requests](privacy_requests.md#request-completion-notification) guide.
 
 ## Prerequisites
 
diff --git a/docs/fidesops/docs/guides/privacy_requests.md b/docs/fidesops/docs/guides/privacy_requests.md
index ead5b2fc1..6f5fe10de 100644
--- a/docs/fidesops/docs/guides/privacy_requests.md
+++ b/docs/fidesops/docs/guides/privacy_requests.md
@@ -54,6 +54,17 @@ to continue privacy request execution.  Until the Privacy Request identity is ve
 ```
 
 
+## Request Completion Notification
+
+By default, a request completion email will be sent to users, along with a link to download their data, if applicable. To change this behavior, set the `send_request_completion_notification`
+variable in your `fidesops.toml`. You must also set up an [EmailConfig](./email_communications.md) that lets fidesops send automated emails
+to your users. If using a custom privacy center, ensure that you intake an email identity, which is required for email notifications throughout fidesops.
+
+!!! Note
+For security purposes, the data package download link is a one-time link and expires in 24 hrs by default. To change TTL, update the `subject_request_download_link_ttl_seconds`
+variable in your `fidesops.toml`.
+
+
 ## Approve and deny Privacy Requests
 
  To review Privacy Requests before they are executed, set the `require_manual_request_approval` variable in your `fidesops.toml` to `TRUE`.
diff --git a/fidesops.toml b/fidesops.toml
index 275c5dc58..e81c12d72 100644
--- a/fidesops.toml
+++ b/fidesops.toml
@@ -29,6 +29,7 @@ drp_jwt_secret = "secret"
 log_level = "INFO"
 root_username = "root_user"
 root_password = "Testpassword1!"
+subject_request_download_link_ttl_seconds = 86400
 
 [execution]
 masking_strict = true
@@ -44,3 +45,6 @@ analytics_opt_out = false
 
 [admin_ui]
 enabled = true
+
+[notifications]
+send_request_completion_notification = false
diff --git a/src/fidesops/ops/common_exceptions.py b/src/fidesops/ops/common_exceptions.py
index 3567a85fb..0cf9a2d81 100644
--- a/src/fidesops/ops/common_exceptions.py
+++ b/src/fidesops/ops/common_exceptions.py
@@ -85,6 +85,10 @@ class StorageConfigNotFoundException(BaseException):
     """Custom Exception - StorageConfig Not Found"""
 
 
+class IdentityNotFoundException(BaseException):
+    """Identity Not Found"""
+
+
 class WebhookOrderException(BaseException):
     """Custom Exception - Issue with webhooks order"""
 
diff --git a/src/fidesops/ops/core/config.py b/src/fidesops/ops/core/config.py
index a13febda6..706b5d060 100644
--- a/src/fidesops/ops/core/config.py
+++ b/src/fidesops/ops/core/config.py
@@ -88,6 +88,7 @@ class FidesopsSecuritySettings(SecuritySettings):
 
     log_level: str = "INFO"
     root_user_scopes: Optional[List[str]] = SCOPE_REGISTRY
+    subject_request_download_link_ttl_seconds: Optional[int] = 86400
 
     @validator("log_level", pre=True)
     def validate_log_level(cls, value: str) -> str:
@@ -149,6 +150,15 @@ class Config:
         env_prefix = "FIDESOPS__ADMIN_UI__"
 
 
+class FidesopsNotificationSettings(FidesSettings):
+    """Configuration settings for data subject and/or data processor notifications"""
+
+    send_request_completion_notification: Optional[bool] = True
+
+    class Config:
+        env_prefix = "FIDESOPS__NOTIFICATIONS__"
+
+
 class FidesopsConfig(FidesSettings):
     """Configuration variables for the FastAPI project"""
 
@@ -158,6 +168,7 @@ class FidesopsConfig(FidesSettings):
     execution: ExecutionSettings
     root_user: RootUserSettings
     admin_ui: AdminUiSettings
+    notifications: FidesopsNotificationSettings
 
     port: int
     is_test_mode: bool = os.getenv("TESTING", "").lower() == "true"
@@ -214,6 +225,7 @@ def log_all_config_values(self) -> None:
         "cors_origins",
         "encoding",
         "oauth_access_token_expire_minutes",
+        "subject_request_download_link_ttl_seconds",
     ],
     "execution": [
         "task_retry_count",
@@ -222,6 +234,7 @@ def log_all_config_values(self) -> None:
         "require_manual_request_approval",
         "subject_identity_verification_required",
     ],
+    "notifications": ["send_request_completion_notification"],
 }
 
 
diff --git a/src/fidesops/ops/email_templates/get_email_template.py b/src/fidesops/ops/email_templates/get_email_template.py
index 833dc36f8..2ecf05af2 100644
--- a/src/fidesops/ops/email_templates/get_email_template.py
+++ b/src/fidesops/ops/email_templates/get_email_template.py
@@ -6,6 +6,8 @@
 from fidesops.ops.common_exceptions import EmailTemplateUnhandledActionType
 from fidesops.ops.email_templates.template_names import (
     EMAIL_ERASURE_REQUEST_FULFILLMENT,
+    PRIVACY_REQUEST_COMPLETE_ACCESS_TEMPLATE,
+    PRIVACY_REQUEST_COMPLETE_DELETION_TEMPLATE,
     SUBJECT_IDENTITY_VERIFICATION_TEMPLATE,
 )
 from fidesops.ops.schemas.email.email import EmailActionType
@@ -25,6 +27,10 @@ def get_email_template(action_type: EmailActionType) -> Template:
         return template_env.get_template(SUBJECT_IDENTITY_VERIFICATION_TEMPLATE)
     if action_type == EmailActionType.EMAIL_ERASURE_REQUEST_FULFILLMENT:
         return template_env.get_template(EMAIL_ERASURE_REQUEST_FULFILLMENT)
+    if action_type == EmailActionType.PRIVACY_REQUEST_COMPLETE_ACCESS:
+        return template_env.get_template(PRIVACY_REQUEST_COMPLETE_ACCESS_TEMPLATE)
+    if action_type == EmailActionType.PRIVACY_REQUEST_COMPLETE_DELETION:
+        return template_env.get_template(PRIVACY_REQUEST_COMPLETE_DELETION_TEMPLATE)
 
     logger.error("No corresponding template linked to the %s", action_type)
     raise EmailTemplateUnhandledActionType(
diff --git a/src/fidesops/ops/email_templates/template_names.py b/src/fidesops/ops/email_templates/template_names.py
index 63e74eace..afcf3cda6 100644
--- a/src/fidesops/ops/email_templates/template_names.py
+++ b/src/fidesops/ops/email_templates/template_names.py
@@ -1,2 +1,4 @@
 SUBJECT_IDENTITY_VERIFICATION_TEMPLATE = "subject_identity_verification.html"
 EMAIL_ERASURE_REQUEST_FULFILLMENT = "erasure_request_email_fulfillment.html"
+PRIVACY_REQUEST_COMPLETE_DELETION_TEMPLATE = "privacy_request_complete_deletion.html"
+PRIVACY_REQUEST_COMPLETE_ACCESS_TEMPLATE = "privacy_request_complete_access.html"
diff --git a/src/fidesops/ops/email_templates/templates/privacy_request_complete_access.html b/src/fidesops/ops/email_templates/templates/privacy_request_complete_access.html
new file mode 100644
index 000000000..72de34102
--- /dev/null
+++ b/src/fidesops/ops/email_templates/templates/privacy_request_complete_access.html
@@ -0,0 +1,26 @@
+<!DOCTYPE html>
+<html lang="en">
+    <head>
+        <meta charset="UTF-8">
+        <title>Privacy Request Complete</title>
+    </head>
+    <body>
+        <main>
+            {% if download_links|length > 1 %}
+            <p>
+                Your data access has been completed and can be downloaded at the below links. For security purposes, these secret links will expire in 24 hours.
+            </p>
+            <ul>
+                {% for link in download_links %}
+                    <li> {{link}} </li>
+                {% endfor %}
+            </ul>
+            {% else %}
+            <p>
+                Your data access has been completed and can be downloaded at {{download_links[0]}}. For security purposes, this secret link will expire in 24 hours.
+            </p>
+
+            {% endif %}
+        </main>
+    </body>
+</html>
\ No newline at end of file
diff --git a/src/fidesops/ops/email_templates/templates/privacy_request_complete_deletion.html b/src/fidesops/ops/email_templates/templates/privacy_request_complete_deletion.html
new file mode 100644
index 000000000..f0232ea4b
--- /dev/null
+++ b/src/fidesops/ops/email_templates/templates/privacy_request_complete_deletion.html
@@ -0,0 +1,14 @@
+<!DOCTYPE html>
+<html lang="en">
+<head>
+    <meta charset="UTF-8">
+    <title>Privacy Request Complete</title>
+</head>
+<body>
+<main>
+    <p>
+        Your privacy request for deletion has been completed.
+    </p>
+</main>
+</body>
+</html>
\ No newline at end of file
diff --git a/src/fidesops/ops/schemas/email/email.py b/src/fidesops/ops/schemas/email/email.py
index f77742acb..05ad08e09 100644
--- a/src/fidesops/ops/schemas/email/email.py
+++ b/src/fidesops/ops/schemas/email/email.py
@@ -1,5 +1,5 @@
 from enum import Enum
-from typing import Any, Dict, Optional, Union
+from typing import Any, Dict, List, Optional, Union
 
 from pydantic import BaseModel, Extra
 
@@ -20,6 +20,8 @@ class EmailActionType(Enum):
     # verify email upon acct creation
     SUBJECT_IDENTITY_VERIFICATION = "subject_identity_verification"
     EMAIL_ERASURE_REQUEST_FULFILLMENT = "email_erasure_fulfillment"
+    PRIVACY_REQUEST_COMPLETE_ACCESS = "privacy_request_complete_access"
+    PRIVACY_REQUEST_COMPLETE_DELETION = "privacy_request_complete_deletion"
 
 
 class EmailTemplateBodyParams(Enum):
@@ -41,6 +43,12 @@ def get_verification_code_ttl_minutes(self) -> int:
         return self.verification_code_ttl_seconds // 60
 
 
+class AccessRequestCompleteBodyParams(BaseModel):
+    """Body params required for privacy request completion access email template"""
+
+    download_links: List[str]
+
+
 class EmailForActionType(BaseModel):
     """Email details that depend on action type"""
 
diff --git a/src/fidesops/ops/service/email/email_dispatch_service.py b/src/fidesops/ops/service/email/email_dispatch_service.py
index d2b44235a..2784897b1 100644
--- a/src/fidesops/ops/service/email/email_dispatch_service.py
+++ b/src/fidesops/ops/service/email/email_dispatch_service.py
@@ -1,5 +1,5 @@
 import logging
-from typing import Any, Optional, Union
+from typing import Any, Optional
 
 import requests
 from requests import Response
@@ -8,14 +8,12 @@
 from fidesops.ops.common_exceptions import EmailDispatchException
 from fidesops.ops.email_templates import get_email_template
 from fidesops.ops.models.email import EmailConfig
-from fidesops.ops.models.privacy_request import EmailRequestFulfillmentBodyParams
 from fidesops.ops.schemas.email.email import (
     EmailActionType,
     EmailForActionType,
     EmailServiceDetails,
     EmailServiceSecrets,
     EmailServiceType,
-    SubjectIdentityVerificationBodyParams,
 )
 from fidesops.ops.util.logger import Pii
 
@@ -26,10 +24,7 @@ def dispatch_email(
     db: Session,
     action_type: EmailActionType,
     to_email: Optional[str],
-    email_body_params: Union[
-        SubjectIdentityVerificationBodyParams,
-        EmailRequestFulfillmentBodyParams,
-    ],
+    email_body_params: Any,
 ) -> None:
     if not to_email:
         raise EmailDispatchException("No email supplied.")
@@ -82,6 +77,22 @@ def _build_email(
                 {"dataset_collection_action_required": body_params}
             ),
         )
+    if action_type == EmailActionType.PRIVACY_REQUEST_COMPLETE_ACCESS:
+        base_template = get_email_template(action_type)
+        return EmailForActionType(
+            subject="Your data is ready to be downloaded",
+            body=base_template.render(
+                {
+                    "download_links": body_params.download_links,
+                }
+            ),
+        )
+    if action_type == EmailActionType.PRIVACY_REQUEST_COMPLETE_DELETION:
+        base_template = get_email_template(action_type)
+        return EmailForActionType(
+            subject="Your data has been deleted",
+            body=base_template.render(),
+        )
     logger.error("Email action type %s is not implemented", action_type)
     raise EmailDispatchException(f"Email action type {action_type} is not implemented")
 
diff --git a/src/fidesops/ops/service/privacy_request/request_runner_service.py b/src/fidesops/ops/service/privacy_request/request_runner_service.py
index b7cd9ef2d..60adea252 100644
--- a/src/fidesops/ops/service/privacy_request/request_runner_service.py
+++ b/src/fidesops/ops/service/privacy_request/request_runner_service.py
@@ -1,7 +1,7 @@
 import logging
 import random
 from datetime import datetime, timedelta
-from typing import ContextManager, Dict, List, Optional, Set
+from typing import Any, ContextManager, Dict, List, Optional, Set
 
 from celery import Task
 from celery.utils.log import get_task_logger
@@ -15,6 +15,7 @@
 from fidesops.ops.common_exceptions import (
     ClientUnsuccessfulException,
     EmailDispatchException,
+    IdentityNotFoundException,
     PrivacyRequestPaused,
 )
 from fidesops.ops.core.config import config
@@ -36,9 +37,15 @@
 from fidesops.ops.models.privacy_request import (
     PrivacyRequest,
     PrivacyRequestStatus,
+    ProvidedIdentityType,
     can_run_checkpoint,
 )
+from fidesops.ops.schemas.email.email import (
+    AccessRequestCompleteBodyParams,
+    EmailActionType,
+)
 from fidesops.ops.service.connectors.email_connector import email_connector_erasure_send
+from fidesops.ops.service.email.email_dispatch_service import dispatch_email
 from fidesops.ops.service.storage.storage_uploader_service import upload
 from fidesops.ops.task.filter_results import filter_data_categories
 from fidesops.ops.task.graph_task import (
@@ -123,11 +130,11 @@ def upload_access_results(
     access_result: Dict[str, List[Row]],
     dataset_graph: DatasetGraph,
     privacy_request: PrivacyRequest,
-) -> None:
+) -> List[str]:
     """Process the data uploads after the access portion of the privacy request has completed"""
+    download_urls: List[str] = []
     if not access_result:
         logging.info("No results returned for access request %s", privacy_request.id)
-
     for rule in policy.get_rules_for_action(action_type=ActionType.access):
         if not rule.storage_destination:
             raise common_exceptions.RuleValidationError(
@@ -145,12 +152,14 @@ def upload_access_results(
             privacy_request.id,
         )
         try:
-            upload(
+            download_url: Optional[str] = upload(
                 db=session,
                 request_id=privacy_request.id,
                 data=filtered_results,
                 storage_key=rule.storage_destination.key,  # type: ignore
             )
+            if download_url:
+                download_urls.append(download_url)
         except common_exceptions.StorageUploadError as exc:
             logging.error(
                 "Error uploading subject access data for rule %s on policy %s and privacy request %s : %s",
@@ -160,6 +169,7 @@ def upload_access_results(
                 Pii(str(exc)),
             )
             privacy_request.status = PrivacyRequestStatus.error
+    return download_urls
 
 
 def queue_privacy_request(
@@ -168,6 +178,7 @@ def queue_privacy_request(
     from_step: Optional[str] = None,
 ) -> str:
     cache: FidesopsRedis = get_cache()
+    logger.info("queueing privacy request")
     task = run_privacy_request.delay(
         privacy_request_id=privacy_request_id,
         from_webhook_id=from_webhook_id,
@@ -207,7 +218,7 @@ async def run_privacy_request(
     from_webhook_id: Optional[str] = None,
     from_step: Optional[str] = None,
 ) -> None:
-    # pylint: disable=too-many-locals, too-many-statements
+    # pylint: disable=too-many-locals, too-many-statements, too-many-return-statements, too-many-branches
     """
     Dispatch a privacy_request into the execution layer by:
         1. Generate a graph from all the currently configured datasets
@@ -260,6 +271,7 @@ async def run_privacy_request(
             dataset_graph = DatasetGraph(*dataset_graphs)
             identity_data = privacy_request.get_cached_identity_data()
             connection_configs = ConnectionConfig.all(db=session)
+            access_result_urls: List[str] = []
 
             if can_run_checkpoint(
                 request_checkpoint=CurrentStep.access, from_checkpoint=resume_step
@@ -272,8 +284,7 @@ async def run_privacy_request(
                     identity=identity_data,
                     session=session,
                 )
-
-                upload_access_results(
+                access_result_urls = upload_access_results(
                     session,
                     policy,
                     access_result,
@@ -343,7 +354,19 @@ async def run_privacy_request(
         )
         if not proceed:
             return
-
+        if config.notifications.send_request_completion_notification:
+            try:
+                initiate_privacy_request_completion_email(
+                    session, policy, access_result_urls, identity_data
+                )
+            except (IdentityNotFoundException, EmailDispatchException) as e:
+                privacy_request.error_processing(db=session)
+                # If dev mode, log traceback
+                await fideslog_graph_failure(
+                    failed_graph_analytics_event(privacy_request, e)
+                )
+                _log_exception(e, config.dev_mode)
+                return
         privacy_request.finished_processing_at = datetime.utcnow()
         AuditLog.create(
             db=session,
@@ -355,8 +378,42 @@ async def run_privacy_request(
             },
         )
         privacy_request.status = PrivacyRequestStatus.complete
-        privacy_request.save(db=session)
         logging.info("Privacy request %s run completed.", privacy_request.id)
+        privacy_request.save(db=session)
+
+
+def initiate_privacy_request_completion_email(
+    session: Session,
+    policy: Policy,
+    access_result_urls: List[str],
+    identity_data: Dict[str, Any],
+) -> None:
+    """
+    :param session: SQLAlchemy Session
+    :param policy: Policy
+    :param access_result_urls: list of urls generated by access request upload
+    :param identity_data: Dict of identity data
+    """
+    if not identity_data.get(ProvidedIdentityType.email.value):
+        raise IdentityNotFoundException(
+            "Identity email was not found, so request completion email could not be sent."
+        )
+    if policy.get_rules_for_action(action_type=ActionType.access):
+        dispatch_email(
+            db=session,
+            action_type=EmailActionType.PRIVACY_REQUEST_COMPLETE_ACCESS,
+            to_email=identity_data.get(ProvidedIdentityType.email.value),
+            email_body_params=AccessRequestCompleteBodyParams(
+                download_links=access_result_urls
+            ),
+        )
+    if policy.get_rules_for_action(action_type=ActionType.erasure):
+        dispatch_email(
+            db=session,
+            action_type=EmailActionType.PRIVACY_REQUEST_COMPLETE_DELETION,
+            to_email=identity_data.get(ProvidedIdentityType.email.value),
+            email_body_params=None,
+        )
 
 
 def initiate_paused_privacy_request_followup(privacy_request: PrivacyRequest) -> None:
diff --git a/src/fidesops/ops/service/storage/storage_authenticator_service.py b/src/fidesops/ops/service/storage/storage_authenticator_service.py
index c28a16380..7a4dd7e69 100644
--- a/src/fidesops/ops/service/storage/storage_authenticator_service.py
+++ b/src/fidesops/ops/service/storage/storage_authenticator_service.py
@@ -1,5 +1,5 @@
 import logging
-from typing import Any, Dict
+from typing import Any
 
 from botocore.exceptions import ClientError
 from requests import RequestException
@@ -7,8 +7,8 @@
 from fidesops.ops.schemas.storage.storage import (
     SUPPORTED_STORAGE_SECRETS,
     S3AuthMethod,
-    StorageSecrets,
     StorageSecretsOnetrust,
+    StorageSecretsS3,
     StorageType,
 )
 from fidesops.ops.util.storage_authenticator import (
@@ -28,10 +28,10 @@ def secrets_are_valid(
     return uploader(secrets)
 
 
-def _s3_authenticator(secrets: Dict[StorageSecrets, Any]) -> bool:
+def _s3_authenticator(secrets: StorageSecretsS3) -> bool:
     """Authenticates secrets for s3, returns true if secrets are valid"""
     try:
-        get_s3_session(S3AuthMethod.SECRET_KEYS, secrets)
+        get_s3_session(S3AuthMethod.SECRET_KEYS.value, secrets.dict())  # type: ignore
         return True
     except ClientError:
         return False
diff --git a/src/fidesops/ops/service/storage/storage_uploader_service.py b/src/fidesops/ops/service/storage/storage_uploader_service.py
index 825e9c5ee..84caebb75 100644
--- a/src/fidesops/ops/service/storage/storage_uploader_service.py
+++ b/src/fidesops/ops/service/storage/storage_uploader_service.py
@@ -21,7 +21,14 @@
 def upload(
     db: Session, *, request_id: str, data: Dict, storage_key: FidesOpsKey
 ) -> str:
-    """Retrieves storage configs and calls appropriate upload method"""
+    """
+    Retrieves storage configs and calls appropriate upload method
+    :param db: SQLAlchemy Session
+    :param request_id: Request id
+    :param data: Dict of data to upload
+    :param storage_key: Key representing where to upload data
+    :return str representing location of upload (url or simply a description of where to find the data)
+    """
     config: Optional[StorageConfig] = StorageConfig.get_by(
         db=db, field="key", value=storage_key
     )
diff --git a/src/fidesops/ops/tasks/storage.py b/src/fidesops/ops/tasks/storage.py
index 32c67814b..feeff9528 100644
--- a/src/fidesops/ops/tasks/storage.py
+++ b/src/fidesops/ops/tasks/storage.py
@@ -11,6 +11,7 @@
 
 import pandas as pd
 import requests
+from boto3 import Session
 from botocore.exceptions import ClientError, ParamValidationError
 from fideslib.cryptography.cryptographic_util import bytes_to_b64_str
 
@@ -28,7 +29,6 @@
 
 logger = logging.getLogger(__name__)
 
-
 LOCAL_FIDES_UPLOAD_DIRECTORY = "fides_uploads"
 
 
@@ -96,6 +96,27 @@ def write_to_in_memory_buffer(
     raise NotImplementedError(f"No handling for response format {resp_format}.")
 
 
+def create_presigned_url_for_s3(
+    s3_client: Session, bucket_name: str, object_name: str
+) -> str:
+    """ "Generate a presigned URL to share an S3 object
+
+    :param s3_client: s3 base client
+    :param bucket_name: string
+    :param object_name: string
+    :return: Presigned URL as string.
+    """
+
+    response = s3_client.generate_presigned_url(
+        "get_object",
+        Params={"Bucket": bucket_name, "Key": object_name},
+        ExpiresIn=config.security.subject_request_download_link_ttl_seconds,
+    )
+
+    # The response contains the presigned URL
+    return response
+
+
 def upload_to_s3(  # pylint: disable=R0913
     storage_secrets: Dict[StorageSecrets, Any],
     data: Dict,
@@ -109,18 +130,28 @@ def upload_to_s3(  # pylint: disable=R0913
     logger.info("Starting S3 Upload of %s", file_key)
     try:
         my_session = get_s3_session(auth_method, storage_secrets)
-
-        s3 = my_session.client("s3")
+        s3_client = my_session.client("s3")
 
         # handles file chunking
-        s3.upload_fileobj(
-            Fileobj=write_to_in_memory_buffer(resp_format, data, request_id),
-            Bucket=bucket_name,
-            Key=file_key,
+        try:
+            s3_client.upload_fileobj(
+                Fileobj=write_to_in_memory_buffer(resp_format, data, request_id),
+                Bucket=bucket_name,
+                Key=file_key,
+            )
+        except Exception as e:
+            logger.error("Encountered error while uploading s3 object: %s", e)
+            raise e
+
+        presigned_url: str = create_presigned_url_for_s3(
+            s3_client, bucket_name, file_key
         )
-        # todo- move to outbound_urn_registry
-        return f"https://{bucket_name}.s3.amazonaws.com/{file_key}"
+
+        return presigned_url
     except ClientError as e:
+        logger.error(
+            "Encountered error while uploading and generating link for s3 object: %s", e
+        )
         raise e
     except ParamValidationError as e:
         raise ValueError(f"The parameters you provided are incorrect: {e}")
@@ -147,7 +178,7 @@ def upload_to_onetrust(
         data=payload,
         headers=headers,
     )
-    return "success"
+    return "onetrust"
 
 
 def _handle_json_encoding(field: Any) -> str:
@@ -169,4 +200,4 @@ def upload_to_local(payload: Dict, file_key: str, request_id: str) -> str:
     with open(filename, "w") as file:  # pylint: disable=W1514
         file.write(data_str)
 
-    return "success"
+    return "your local fides_uploads folder"
diff --git a/src/fidesops/ops/util/storage_authenticator.py b/src/fidesops/ops/util/storage_authenticator.py
index d47a12b58..8a77ccb21 100644
--- a/src/fidesops/ops/util/storage_authenticator.py
+++ b/src/fidesops/ops/util/storage_authenticator.py
@@ -41,8 +41,8 @@ def get_s3_session(
         logger.info("Successfully created automatic session")
         return session
 
-    logger.error("No S3 session created. Auth method used: %s", auth_method)
-    raise ValueError(f"No S3 session created. Auth method used: {auth_method}")
+    logger.error("Auth method not supported for S3: %s", auth_method)
+    raise ValueError(f"Auth method not supported for S3: {auth_method}")
 
 
 def get_onetrust_access_token(client_id: str, client_secret: str, hostname: str) -> str:
diff --git a/tests/ops/api/v1/endpoints/test_config_endpoints.py b/tests/ops/api/v1/endpoints/test_config_endpoints.py
index 95286f1c5..e2c83556d 100644
--- a/tests/ops/api/v1/endpoints/test_config_endpoints.py
+++ b/tests/ops/api/v1/endpoints/test_config_endpoints.py
@@ -35,6 +35,7 @@ def test_get_config(
                             "cors_origins",
                             "encoding",
                             "oauth_access_token_expire_minutes",
+                            "subject_request_download_link_ttl_seconds",
                         ]
                     )
                 )
diff --git a/tests/ops/api/v1/endpoints/test_storage_endpoints.py b/tests/ops/api/v1/endpoints/test_storage_endpoints.py
index a434b8e12..67390d221 100644
--- a/tests/ops/api/v1/endpoints/test_storage_endpoints.py
+++ b/tests/ops/api/v1/endpoints/test_storage_endpoints.py
@@ -554,11 +554,11 @@ def test_put_s3_config_secrets_and_verify(
         response = api_client.put(url, headers=auth_header, json=payload)
         assert 200 == response.status_code
         get_s3_session_mock.assert_called_once_with(
-            S3AuthMethod.SECRET_KEYS,
-            StorageSecretsS3(
-                aws_access_key_id=payload["aws_access_key_id"],
-                aws_secret_access_key=payload["aws_secret_access_key"],
-            ),
+            S3AuthMethod.SECRET_KEYS.value,
+            {
+                "aws_access_key_id": payload["aws_access_key_id"],
+                "aws_secret_access_key": payload["aws_secret_access_key"],
+            },
         )
 
     @mock.patch(
diff --git a/tests/ops/conftest.py b/tests/ops/conftest.py
index 7afe7da4c..439396cee 100644
--- a/tests/ops/conftest.py
+++ b/tests/ops/conftest.py
@@ -255,6 +255,15 @@ def subject_identity_verification_not_required():
     config.execution.subject_identity_verification_required = original_value
 
 
+@pytest.fixture(autouse=True, scope="function")
+def privacy_request_complete_email_notification_disabled():
+    """Disable request completion email for most tests unless overridden"""
+    original_value = config.notifications.send_request_completion_notification
+    config.notifications.send_request_completion_notification = False
+    yield
+    config.notifications.send_request_completion_notification = original_value
+
+
 @pytest.fixture(scope="session", autouse=True)
 def event_loop():
     try:
diff --git a/tests/ops/fixtures/application_fixtures.py b/tests/ops/fixtures/application_fixtures.py
index 8f7a9da10..d18989815 100644
--- a/tests/ops/fixtures/application_fixtures.py
+++ b/tests/ops/fixtures/application_fixtures.py
@@ -17,6 +17,7 @@
 from sqlalchemy.orm.exc import ObjectDeletedError
 
 from fidesops.ops.api.v1.scope_registry import PRIVACY_REQUEST_READ, SCOPE_REGISTRY
+from fidesops.ops.core.config import config
 from fidesops.ops.models.connectionconfig import (
     AccessLevel,
     ConnectionConfig,
@@ -293,6 +294,77 @@ def policy_post_execution_webhooks(
         pass
 
 
+@pytest.fixture(scope="function")
+def access_and_erasure_policy(
+    db: Session,
+    oauth_client: ClientDetail,
+    storage_config: StorageConfig,
+) -> Generator:
+    access_and_erasure_policy = Policy.create(
+        db=db,
+        data={
+            "name": "example access and erasure policy",
+            "key": "example_access_erasure_policy",
+            "client_id": oauth_client.id,
+        },
+    )
+    access_rule = Rule.create(
+        db=db,
+        data={
+            "action_type": ActionType.access.value,
+            "client_id": oauth_client.id,
+            "name": "Access Request Rule",
+            "policy_id": access_and_erasure_policy.id,
+            "storage_destination_id": storage_config.id,
+        },
+    )
+    access_rule_target = RuleTarget.create(
+        db=db,
+        data={
+            "client_id": oauth_client.id,
+            "data_category": DataCategory("user").value,
+            "rule_id": access_rule.id,
+        },
+    )
+    erasure_rule = Rule.create(
+        db=db,
+        data={
+            "action_type": ActionType.erasure.value,
+            "client_id": oauth_client.id,
+            "name": "Erasure Rule",
+            "policy_id": access_and_erasure_policy.id,
+            "masking_strategy": {
+                "strategy": "null_rewrite",
+                "configuration": {},
+            },
+        },
+    )
+
+    erasure_rule_target = RuleTarget.create(
+        db=db,
+        data={
+            "client_id": oauth_client.id,
+            "data_category": DataCategory("user.name").value,
+            "rule_id": erasure_rule.id,
+        },
+    )
+    yield access_and_erasure_policy
+    try:
+        access_rule_target.delete(db)
+        erasure_rule_target.delete(db)
+    except ObjectDeletedError:
+        pass
+    try:
+        access_rule.delete(db)
+        erasure_rule.delete(db)
+    except ObjectDeletedError:
+        pass
+    try:
+        access_and_erasure_policy.delete(db)
+    except ObjectDeletedError:
+        pass
+
+
 @pytest.fixture(scope="function")
 def erasure_policy(
     db: Session,
@@ -811,10 +883,13 @@ def _create_privacy_request_for_policy(
         db=db,
         data=data,
     )
+    email_identity = "test@example.com"
+    identity_kwargs = {"email": email_identity}
+    pr.cache_identity(identity_kwargs)
     pr.persist_identity(
         db=db,
         identity=PrivacyRequestIdentity(
-            email="test@example.com",
+            email=email_identity,
             phone_number="+1 234 567 8910",
         ),
     )
diff --git a/tests/ops/service/privacy_request/request_runner_service_test.py b/tests/ops/service/privacy_request/request_runner_service_test.py
index a9a22fba9..ef61c6f3a 100644
--- a/tests/ops/service/privacy_request/request_runner_service_test.py
+++ b/tests/ops/service/privacy_request/request_runner_service_test.py
@@ -1,7 +1,7 @@
 import time
 from typing import Any, Dict, List, Set
 from unittest import mock
-from unittest.mock import Mock
+from unittest.mock import ANY, Mock, call
 from uuid import uuid4
 
 import pydash
@@ -13,6 +13,8 @@
 
 from fidesops.ops.common_exceptions import (
     ClientUnsuccessfulException,
+    EmailDispatchException,
+    IdentityNotFoundException,
     PrivacyRequestPaused,
 )
 from fidesops.ops.core.config import config
@@ -28,7 +30,11 @@
     PrivacyRequest,
     PrivacyRequestStatus,
 )
-from fidesops.ops.schemas.email.email import EmailForActionType
+from fidesops.ops.schemas.email.email import (
+    AccessRequestCompleteBodyParams,
+    EmailActionType,
+    EmailForActionType,
+)
 from fidesops.ops.schemas.external_https import SecondPartyResponseFormat
 from fidesops.ops.schemas.masking.masking_configuration import (
     HmacMaskingConfiguration,
@@ -56,23 +62,47 @@
 PRIVACY_REQUEST_TASK_TIMEOUT_EXTERNAL = 30
 
 
+@pytest.fixture(scope="function")
+def privacy_request_complete_email_notification_enabled():
+    """Enable request completion email"""
+    original_value = config.notifications.send_request_completion_notification
+    config.notifications.send_request_completion_notification = True
+    yield
+    config.notifications.send_request_completion_notification = original_value
+
+
+@mock.patch(
+    "fidesops.ops.service.privacy_request.request_runner_service.dispatch_email"
+)
 @mock.patch("fidesops.ops.service.privacy_request.request_runner_service.upload")
-def test_policy_upload_called(
+def test_policy_upload_dispatch_email_called(
     upload_mock: Mock,
+    mock_email_dispatch: Mock,
     privacy_request_status_pending: PrivacyRequest,
     run_privacy_request_task,
+    privacy_request_complete_email_notification_enabled,
 ) -> None:
+    upload_mock.return_value = "http://www.data-download-url"
     run_privacy_request_task.delay(privacy_request_status_pending.id).get(
         timeout=PRIVACY_REQUEST_TASK_TIMEOUT
     )
     assert upload_mock.called
+    assert mock_email_dispatch.call_count == 1
 
 
+@mock.patch(
+    "fidesops.ops.service.privacy_request.request_runner_service.dispatch_email"
+)
+@mock.patch("fidesops.ops.service.privacy_request.request_runner_service.upload")
 def test_start_processing_sets_started_processing_at(
+    upload_mock: Mock,
+    mock_email_dispatch: Mock,
     db: Session,
     privacy_request_status_pending: PrivacyRequest,
     run_privacy_request_task,
+    privacy_request_complete_email_notification_enabled,
 ) -> None:
+    upload_mock.return_value = "http://www.data-download-url"
     updated_at = privacy_request_status_pending.updated_at
     assert privacy_request_status_pending.started_processing_at is None
     run_privacy_request_task.delay(privacy_request_status_pending.id).get(
@@ -83,12 +113,22 @@ def test_start_processing_sets_started_processing_at(
     assert privacy_request_status_pending.started_processing_at is not None
     assert privacy_request_status_pending.updated_at > updated_at
 
+    assert mock_email_dispatch.call_count == 1
 
+
+@mock.patch(
+    "fidesops.ops.service.privacy_request.request_runner_service.dispatch_email"
+)
+@mock.patch("fidesops.ops.service.privacy_request.request_runner_service.upload")
 def test_start_processing_doesnt_overwrite_started_processing_at(
+    upload_mock: Mock,
+    mock_email_dispatch: Mock,
     db: Session,
     privacy_request: PrivacyRequest,
     run_privacy_request_task,
+    privacy_request_complete_email_notification_enabled,
 ) -> None:
+    upload_mock.return_value = "http://www.data-download-url"
     before = privacy_request.started_processing_at
     assert before is not None
     updated_at = privacy_request.updated_at
@@ -101,6 +141,8 @@ def test_start_processing_doesnt_overwrite_started_processing_at(
     assert privacy_request.started_processing_at == before
     assert privacy_request.updated_at > updated_at
 
+    assert mock_email_dispatch.call_count == 1
+
 
 @mock.patch(
     "fidesops.ops.service.privacy_request.request_runner_service.upload_access_results"
@@ -110,6 +152,7 @@ def test_halts_proceeding_if_cancelled(
     db: Session,
     privacy_request_status_canceled: PrivacyRequest,
     run_privacy_request_task,
+    privacy_request_complete_email_notification_enabled,
 ) -> None:
     assert privacy_request_status_canceled.status == PrivacyRequestStatus.canceled
     run_privacy_request_task.delay(privacy_request_status_canceled.id).get(
@@ -124,6 +167,12 @@ def test_halts_proceeding_if_cancelled(
     assert not upload_access_results_mock.called
 
 
+@mock.patch(
+    "fidesops.ops.service.privacy_request.request_runner_service.dispatch_email"
+)
+@mock.patch(
+    "fidesops.ops.service.privacy_request.request_runner_service.upload_access_results"
+)
 @mock.patch(
     "fidesops.ops.service.privacy_request.request_runner_service.run_webhooks_and_report_status",
 )
@@ -135,11 +184,15 @@ def test_from_graph_resume_does_not_run_pre_webhooks(
     run_erasure,
     run_access,
     run_webhooks,
+    upload_mock: Mock,
+    mock_email_dispatch,
     db: Session,
     privacy_request: PrivacyRequest,
     run_privacy_request_task,
     erasure_policy,
+    privacy_request_complete_email_notification_enabled,
 ) -> None:
+    upload_mock.return_value = "http://www.data-download-url"
     privacy_request.started_processing_at = None
     privacy_request.policy = erasure_policy
     privacy_request.save(db)
@@ -161,7 +214,12 @@ def test_from_graph_resume_does_not_run_pre_webhooks(
     assert run_access.call_count == 1  # Access request runs
     assert run_erasure.call_count == 1  # Erasure request runs
 
+    assert mock_email_dispatch.call_count == 1
+
 
+@mock.patch(
+    "fidesops.ops.service.privacy_request.request_runner_service.dispatch_email"
+)
 @mock.patch(
     "fidesops.ops.service.privacy_request.request_runner_service.run_webhooks_and_report_status",
 )
@@ -173,10 +231,12 @@ def test_resume_privacy_request_from_erasure(
     run_erasure,
     run_access,
     run_webhooks,
+    mock_email_dispatch,
     db: Session,
     privacy_request: PrivacyRequest,
     run_privacy_request_task,
     erasure_policy,
+    privacy_request_complete_email_notification_enabled,
 ) -> None:
     privacy_request.started_processing_at = None
     privacy_request.policy = erasure_policy
@@ -199,6 +259,8 @@ def test_resume_privacy_request_from_erasure(
     assert run_access.call_count == 0  # Access request skipped
     assert run_erasure.call_count == 1  # Erasure request runs
 
+    assert mock_email_dispatch.call_count == 1
+
 
 def get_privacy_request_results(
     db,
@@ -1773,3 +1835,220 @@ def test_email_connector_no_updates_needed(
         assert (
             mailgun_send.called is False
         ), "Email not sent because no updates are needed. Data category doesn't apply to any of the collections."
+
+
+class TestPrivacyRequestsEmailNotifications:
+    @pytest.fixture(scope="function")
+    def privacy_request_complete_email_notification_enabled(self):
+        """Enable request completion email"""
+        original_value = config.notifications.send_request_completion_notification
+        config.notifications.send_request_completion_notification = True
+        yield
+        config.notifications.send_request_completion_notification = original_value
+
+    @pytest.mark.integration_postgres
+    @pytest.mark.integration
+    @mock.patch(
+        "fidesops.ops.service.privacy_request.request_runner_service.dispatch_email"
+    )
+    def test_email_complete_send_erasure(
+        self,
+        mailgun_send,
+        postgres_integration_db,
+        postgres_example_test_dataset_config,
+        cache,
+        db,
+        generate_auth_header,
+        erasure_policy,
+        read_connection_config,
+        email_config,
+        privacy_request_complete_email_notification_enabled,
+        run_privacy_request_task,
+    ):
+        customer_email = "customer-1@example.com"
+        data = {
+            "requested_at": "2021-08-30T16:09:37.359Z",
+            "policy_key": erasure_policy.key,
+            "identity": {"email": customer_email},
+        }
+
+        pr = get_privacy_request_results(
+            db,
+            erasure_policy,
+            run_privacy_request_task,
+            data,
+        )
+        pr.delete(db=db)
+
+        mailgun_send.assert_called_once()
+
+    @pytest.mark.integration_postgres
+    @pytest.mark.integration
+    @mock.patch(
+        "fidesops.ops.service.privacy_request.request_runner_service.dispatch_email"
+    )
+    @mock.patch("fidesops.ops.service.privacy_request.request_runner_service.upload")
+    def test_email_complete_send_access(
+        self,
+        upload_mock,
+        mailgun_send,
+        postgres_integration_db,
+        postgres_example_test_dataset_config,
+        cache,
+        db,
+        generate_auth_header,
+        policy,
+        read_connection_config,
+        email_config,
+        privacy_request_complete_email_notification_enabled,
+        run_privacy_request_task,
+    ):
+        upload_mock.return_value = "http://www.data-download-url"
+        customer_email = "customer-1@example.com"
+        data = {
+            "requested_at": "2021-08-30T16:09:37.359Z",
+            "policy_key": policy.key,
+            "identity": {"email": customer_email},
+        }
+
+        pr = get_privacy_request_results(
+            db,
+            policy,
+            run_privacy_request_task,
+            data,
+        )
+        pr.delete(db=db)
+
+        mailgun_send.assert_called_once()
+
+    @pytest.mark.integration_postgres
+    @pytest.mark.integration
+    @mock.patch(
+        "fidesops.ops.service.privacy_request.request_runner_service.dispatch_email"
+    )
+    @mock.patch("fidesops.ops.service.privacy_request.request_runner_service.upload")
+    def test_email_complete_send_access_and_erasure(
+        self,
+        upload_mock,
+        mailgun_send,
+        postgres_integration_db,
+        postgres_example_test_dataset_config,
+        cache,
+        db,
+        generate_auth_header,
+        access_and_erasure_policy,
+        read_connection_config,
+        email_config,
+        privacy_request_complete_email_notification_enabled,
+        run_privacy_request_task,
+    ):
+        upload_mock.return_value = "http://www.data-download-url"
+        customer_email = "customer-1@example.com"
+        data = {
+            "requested_at": "2021-08-30T16:09:37.359Z",
+            "policy_key": access_and_erasure_policy.key,
+            "identity": {"email": customer_email},
+        }
+
+        pr = get_privacy_request_results(
+            db,
+            access_and_erasure_policy,
+            run_privacy_request_task,
+            data,
+        )
+        pr.delete(db=db)
+
+        mailgun_send.assert_has_calls(
+            [
+                call(
+                    db=ANY,
+                    action_type=EmailActionType.PRIVACY_REQUEST_COMPLETE_ACCESS,
+                    to_email=customer_email,
+                    email_body_params=AccessRequestCompleteBodyParams(
+                        download_links=[upload_mock.return_value]
+                    ),
+                ),
+                call(
+                    db=ANY,
+                    action_type=EmailActionType.PRIVACY_REQUEST_COMPLETE_DELETION,
+                    to_email=customer_email,
+                    email_body_params=None,
+                ),
+            ],
+            any_order=True,
+        )
+
+    @pytest.mark.integration_postgres
+    @pytest.mark.integration
+    @mock.patch("fidesops.ops.service.email.email_dispatch_service._mailgun_dispatcher")
+    @mock.patch("fidesops.ops.service.privacy_request.request_runner_service.upload")
+    def test_email_complete_send_access_no_email_config(
+        self,
+        upload_mock,
+        mailgun_send,
+        postgres_integration_db,
+        postgres_example_test_dataset_config,
+        cache,
+        db,
+        generate_auth_header,
+        policy,
+        read_connection_config,
+        privacy_request_complete_email_notification_enabled,
+        run_privacy_request_task,
+    ):
+        upload_mock.return_value = "http://www.data-download-url"
+        customer_email = "customer-1@example.com"
+        data = {
+            "requested_at": "2021-08-30T16:09:37.359Z",
+            "policy_key": policy.key,
+            "identity": {"email": customer_email},
+        }
+
+        pr = get_privacy_request_results(
+            db,
+            policy,
+            run_privacy_request_task,
+            data,
+        )
+        db.refresh(pr)
+        assert pr.status == PrivacyRequestStatus.error
+        pr.delete(db=db)
+
+        assert mailgun_send.called is False
+
+    @pytest.mark.integration_postgres
+    @pytest.mark.integration
+    @mock.patch("fidesops.ops.service.email.email_dispatch_service._mailgun_dispatcher")
+    @mock.patch("fidesops.ops.service.privacy_request.request_runner_service.upload")
+    def test_email_complete_send_access_no_email_identity(
+        self,
+        upload_mock,
+        mailgun_send,
+        postgres_integration_db,
+        postgres_example_test_dataset_config,
+        cache,
+        db,
+        generate_auth_header,
+        policy,
+        read_connection_config,
+        privacy_request_complete_email_notification_enabled,
+        run_privacy_request_task,
+    ):
+        upload_mock.return_value = "http://www.data-download-url"
+        data = {
+            "requested_at": "2021-08-30T16:09:37.359Z",
+            "policy_key": policy.key,
+            "identity": {"phone": "1231231233"},
+        }
+
+        pr = get_privacy_request_results(
+            db,
+            policy,
+            run_privacy_request_task,
+            data,
+        )
+        db.refresh(pr)
+        assert pr.status == PrivacyRequestStatus.error
+        pr.delete(db=db)
+
+        assert mailgun_send.called is False