diff --git a/alembic/versions/1497926ddec1_add_deleted_environment_role_status.py b/alembic/versions/1497926ddec1_add_deleted_environment_role_status.py
new file mode 100644
index 00000000..ab3b01f8
--- /dev/null
+++ b/alembic/versions/1497926ddec1_add_deleted_environment_role_status.py
@@ -0,0 +1,73 @@
+"""add deleted environment_role status
+
+Revision ID: 1497926ddec1
+Revises: e3d93f9caba7
+Create Date: 2019-10-04 10:44:54.198368
+
+"""
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = "1497926ddec1"  # pragma: allowlist secret
+down_revision = "e3d93f9caba7"  # pragma: allowlist secret
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # op.alter_column("environment_roles", "status", new_column_name="status_old")
+    op.alter_column(
+        "environment_roles",
+        "status",
+        type_=sa.Enum(
+            "PENDING",
+            "COMPLETED",
+            "PENDING_DELETE",
+            "DELETED",
+            name="status",
+            native_enum=False,
+        ),
+        existing_type=sa.Enum(
+            "PENDING",
+            "COMPLETED",
+            "PENDING_DELETE",
+            name="status",
+            native_enum=False,
+        ),
+    )
+
+    # conn = op.get_bind()
+    # conn.execute("UPDATE environment_roles SET status = status_old")
+    # op.drop_column("environment_roles", "status_old")
+
+
+def downgrade():
+    conn = op.get_bind()
+    conn.execute(
+        """
+        UPDATE environment_roles
+        SET status = (CASE WHEN status = 'DELETED' THEN 'PENDING_DELETE' ELSE status END)
+        """
+    )
+
+    op.alter_column(
+        "environment_roles",
+        "status",
+        type_=sa.Enum(
+            "PENDING",
+            "COMPLETED",
+            "PENDING_DELETE",
+            name="status",
+            native_enum=False,
+        ),
+        existing_type=sa.Enum(
+            "PENDING",
+            "COMPLETED",
+            "PENDING_DELETE",
+            "DELETED",
+            name="status",
+            native_enum=False,
+        ),
+    )
diff --git a/atst/app.py b/atst/app.py
index f3e7276f..cd1541b0 100644
--- a/atst/app.py
+++ b/atst/app.py
@@ -8,6 +8,7 @@ from flask_session import Session
 import redis
 from unipath import Path
 from flask_wtf.csrf import CSRFProtect
+import json
 
 from atst.database import db
 from atst.assets import environment as assets_environment
@@ -148,6 +149,15 @@ def set_default_headers(app):  # pragma: no cover
 
 
 def map_config(config):
+    def sqlalchemy_dumps(dct):
+        def _default(self, obj):
+            if isinstance(obj, Enum):
+                return obj.name
+            else:
+                raise TypeError()
+
+        return json.dumps(dct, default=_default)
+
     return {
         **config["default"],
         "ENV": config["default"]["ENVIRONMENT"],
@@ -158,6 +168,7 @@ def map_config(config):
         "PORT": int(config["default"]["PORT"]),
         "SQLALCHEMY_DATABASE_URI": config["default"]["DATABASE_URI"],
         "SQLALCHEMY_TRACK_MODIFICATIONS": False,
+        "SQLALCHEMY_ENGINE_OPTIONS": {"json_serializer": sqlalchemy_dumps},
         "WTF_CSRF_ENABLED": config.getboolean("default", "WTF_CSRF_ENABLED"),
         "PERMANENT_SESSION_LIFETIME": config.getint(
             "default", "PERMANENT_SESSION_LIFETIME"
diff --git a/atst/jobs.py b/atst/jobs.py
index e861d8b5..8e5d5602 100644
--- a/atst/jobs.py
+++ b/atst/jobs.py
@@ -3,7 +3,11 @@ import pendulum
 
 from atst.database import db
 from atst.queue import celery
-from atst.models import EnvironmentJobFailure, EnvironmentRoleJobFailure
+from atst.models import (
+    EnvironmentJobFailure,
+    EnvironmentRoleJobFailure,
+    EnvironmentRole,
+)
 from atst.domain.csp.cloud import CloudProviderInterface, GeneralCSPException
 from atst.domain.environments import Environments
 from atst.domain.environment_roles import EnvironmentRoles
@@ -112,6 +116,19 @@ def do_provision_user(csp: CloudProviderInterface, environment_role_id=None):
             credentials, environment_role, environment_role.role
         )
         environment_role.csp_user_id = csp_user_id
+        environment_role.status = EnvironmentRole.Status.COMPLETED
+        db.session.add(environment_role)
+        db.session.commit()
+
+
+def do_delete_user(csp: CloudProviderInterface, environment_role_id=None):
+    environment_role = EnvironmentRoles.get_by_id(environment_role_id)
+
+    with claim_for_update(environment_role) as environment_role:
+        credentials = environment_role.environment.csp_credentials
+
+        csp.delete_user(credentials, environment_role.csp_user_id)
+        environment_role.status = EnvironmentRole.Status.DELETED
         db.session.add(environment_role)
         db.session.commit()
 
diff --git a/atst/models/environment_role.py b/atst/models/environment_role.py
index 988d19ab..f3e8004f 100644
--- a/atst/models/environment_role.py
+++ b/atst/models/environment_role.py
@@ -40,6 +40,7 @@ class EnvironmentRole(
         PENDING = "pending"
         COMPLETED = "completed"
         PENDING_DELETE = "pending_delete"
+        DELETED = "deleted"
 
     status = Column(SQLAEnum(Status, native_enum=False), default=Status.PENDING)
 
diff --git a/tests/test_jobs.py b/tests/test_jobs.py
index c4d17b5d..47ec1840 100644
--- a/tests/test_jobs.py
+++ b/tests/test_jobs.py
@@ -3,7 +3,6 @@ import pytest
 from uuid import uuid4
 from unittest.mock import Mock
 from threading import Thread
-from time import sleep
 
 from atst.domain.csp.cloud import MockCloudProvider
 from atst.jobs import (
@@ -18,6 +17,7 @@ from atst.jobs import (
     create_environment,
     dispatch_provision_user,
     do_provision_user,
+    do_delete_user,
 )
 from atst.models.utils import claim_for_update
 from atst.domain.exceptions import ClaimFailedException
@@ -370,3 +370,24 @@ def test_do_provision_user(csp, session):
     )
     # I expect that the EnvironmentRole now has a csp_user_id
     assert environment_role.csp_user_id
+
+
+def test_do_delete_user(csp, session):
+    credentials = MockCloudProvider(())._auth_credentials
+    provisioned_environment = EnvironmentFactory.create(
+        cloud_id="cloud_id",
+        root_user_info={"credentials": credentials},
+        baseline_info={},
+    )
+
+    environment_role = EnvironmentRoleFactory.create(
+        environment=provisioned_environment,
+        status=EnvironmentRole.Status.PENDING_DELETE,
+        role="my_role",
+    )
+
+    do_delete_user(csp=csp, environment_role_id=environment_role.id)
+
+    session.refresh(environment_role)
+
+    assert environment_role.status == EnvironmentRole.Status.DELETED