We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
New connection creation flow does not handle schema creation failure scenarios
## Description
1. Create a database manually, say `test_db`, and create a schema called `Library Management` in it.
1. In Mathesar, create a new connection to the database.
1. While creating the connection, in the "New PostgreSQL Database Connection" modal, check the "Library Management" schema in the "Schemas to install" field.
   ![Screenshot 2024-01-26 at 1 01 21 PM](https://github.com/mathesar-foundation/mathesar/assets/11850603/ec5c109b-d779-4d56-96a0-1c209e3deb3d)
1. Click on 'Add Connection'.
1. An error is thrown:
   - `"(psycopg2.errors.DuplicateSchema) schema \"Library Management\" already exists\n\n[SQL: CREATE SCHEMA \"Library Management\";]\n(Background on this error at: https://sqlalche.me/e/14/f405)"`.
   - This is expected.
1. The modal remains open. Uncheck the "Library Management" schema, and try clicking on 'Add Connection' again.
1. Another error is thrown:
   - `duplicate key value violates unique constraint "mathesar_database_name_key" DETAIL: Key (name)=(pavish_install_test_1) already exists.`
   - From the user perpective, there's no clear next step for them to take when this happens.
1. Close the modal and refresh the page, notice that the connection has been created already.
1. Open the connection, and notice that it shows `Schemas (0)`.
1. Forcing a manual reflection using the "Sync external changes" button shows the public schema and existing tables.

### Note:
* I performed the above steps 1-3 only as a means to force an error in the schema creation flow. While testing, I encountered an error which occurred due to an unknown reason. 
* This issue primarily focuses on steps 4-10.

## Expected behavior
* When an error occurs, there should not be an inconsistent middle state in the DB vs the UI.
* When schema creation fails, we should try to revert back to the original state, or we should make it extra clear on the UI that it's a multi-step process.

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `mathesar/utils/connections.py`
Content:
```
1 """Utilities to help with creating and managing connections in Mathesar."""
2 from psycopg2.errors import DuplicateSchema
3 from sqlalchemy.exc import OperationalError
4 from mathesar.models.base import Database
5 from db import install, connection as dbconn
6 from mathesar.state import reset_reflection
7 from demo.install.library_dataset import load_library_dataset
8 from demo.install.movies_dataset import load_movies_dataset
9 
10 
11 class BadInstallationTarget(Exception):
12     """Raise when an attempt is made to install on a disallowed target"""
13     pass
14 
15 
16 def copy_connection_from_preexisting(
17         connection, nickname, db_name, create_db, sample_data
18 ):
19     if connection['connection_type'] == 'internal_database':
20         db_model = Database.create_from_settings_key('default')
21     elif connection['connection_type'] == 'user_database':
22         db_model = Database.current_objects.get(id=connection['id'])
23         db_model.id = None
24     else:
25         raise KeyError("connection_type")
26     root_db = db_model.db_name
27     return _save_and_install(
28         db_model, db_name, root_db, nickname, create_db, sample_data
29     )
30 
31 
32 def create_connection_from_scratch(
33         user, password, host, port, nickname, db_name, sample_data
34 ):
35     db_model = Database(username=user, password=password, host=host, port=port)
36     root_db = db_name
37     return _save_and_install(
38         db_model, db_name, root_db, nickname, False, sample_data
39     )
40 
41 
42 def create_connection_with_new_user(
43         connection, user, password, nickname, db_name, create_db, sample_data
44 ):
45     db_model = copy_connection_from_preexisting(
46         connection, nickname, db_name, create_db, []
47     )
48     engine = db_model._sa_engine
49     db_model.username = user
50     db_model.password = password
51     db_model.save()
52     dbconn.execute_msar_func_with_engine(
53         engine,
54         'create_basic_mathesar_user',
55         db_model.username,
56         db_model.password
57     )
58     _load_sample_data(db_model._sa_engine, sample_data)
59     return db_model
60 
61 
62 def _save_and_install(
63         db_model, db_name, root_db, nickname, create_db, sample_data
64 ):
65     db_model.name = nickname
66     db_model.db_name = db_name
67     _validate_db_model(db_model)
68     db_model.save()
69     try:
70         install.install_mathesar(
71             database_name=db_model.db_name,
72             username=db_model.username,
73             password=db_model.password,
74             hostname=db_model.host,
75             port=db_model.port,
76             skip_confirm=True,
77             create_db=create_db,
78             root_db=root_db,
79         )
80     except OperationalError as e:
81         db_model.delete()
82         raise e
83     _load_sample_data(db_model._sa_engine, sample_data)
84     return db_model
85 
86 
87 def _load_sample_data(engine, sample_data):
88     DATASET_MAP = {
89         'library_management': load_library_dataset,
90         'movie_collection': load_movies_dataset,
91     }
92     for key in sample_data:
93         try:
94             DATASET_MAP[key](engine, safe_mode=True)
95         except DuplicateSchema:
96             # We swallow this error, since otherwise we'll raise an error on the
97             # front end even though installation generally succeeded.
98             continue
99     reset_reflection()
100 
101 
102 def _validate_db_model(db_model):
103     internal_db_model = Database.create_from_settings_key('default')
104     if (
105             internal_db_model is not None
106             and db_model.host == internal_db_model.host
107             and db_model.port == internal_db_model.port
108             and db_model.db_name == internal_db_model.db_name
109     ):
110         raise BadInstallationTarget(
111             "Mathesar can't be installed in the internal DB namespace"
112         )
113
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/mathesar/utils/connections.py b/mathesar/utils/connections.py
--- a/mathesar/utils/connections.py
+++ b/mathesar/utils/connections.py
@@ -1,6 +1,6 @@
 """Utilities to help with creating and managing connections in Mathesar."""
 from psycopg2.errors import DuplicateSchema
-from sqlalchemy.exc import OperationalError
+from sqlalchemy.exc import OperationalError, ProgrammingError
 from mathesar.models.base import Database
 from db import install, connection as dbconn
 from mathesar.state import reset_reflection
@@ -92,10 +92,11 @@
     for key in sample_data:
         try:
             DATASET_MAP[key](engine, safe_mode=True)
-        except DuplicateSchema:
-            # We swallow this error, since otherwise we'll raise an error on the
-            # front end even though installation generally succeeded.
-            continue
+        except ProgrammingError as e:
+            if isinstance(e.orig, DuplicateSchema):
+                # We swallow this error, since otherwise we'll raise an error on the
+                # front end even though installation generally succeeded.
+                continue
     reset_reflection()

{"golden_diff": "diff --git a/mathesar/utils/connections.py b/mathesar/utils/connections.py\n--- a/mathesar/utils/connections.py\n+++ b/mathesar/utils/connections.py\n@@ -1,6 +1,6 @@\n \"\"\"Utilities to help with creating and managing connections in Mathesar.\"\"\"\n from psycopg2.errors import DuplicateSchema\n-from sqlalchemy.exc import OperationalError\n+from sqlalchemy.exc import OperationalError, ProgrammingError\n from mathesar.models.base import Database\n from db import install, connection as dbconn\n from mathesar.state import reset_reflection\n@@ -92,10 +92,11 @@\n     for key in sample_data:\n         try:\n             DATASET_MAP[key](engine, safe_mode=True)\n-        except DuplicateSchema:\n-            # We swallow this error, since otherwise we'll raise an error on the\n-            # front end even though installation generally succeeded.\n-            continue\n+        except ProgrammingError as e:\n+            if isinstance(e.orig, DuplicateSchema):\n+                # We swallow this error, since otherwise we'll raise an error on the\n+                # front end even though installation generally succeeded.\n+                continue\n     reset_reflection()\n", "issue": "New connection creation flow does not handle schema creation failure scenarios\n## Description\r\n1. Create a database manually, say `test_db`, and create a schema called `Library Management` in it.\r\n1. In Mathesar, create a new connection to the database.\r\n1. While creating the connection, in the \"New PostgreSQL Database Connection\" modal, check the \"Library Management\" schema in the \"Schemas to install\" field.\r\n   ![Screenshot 2024-01-26 at 1 01 21\u202fPM](https://github.com/mathesar-foundation/mathesar/assets/11850603/ec5c109b-d779-4d56-96a0-1c209e3deb3d)\r\n1. Click on 'Add Connection'.\r\n1. An error is thrown:\r\n   - `\"(psycopg2.errors.DuplicateSchema) schema \\\"Library Management\\\" already exists\\n\\n[SQL: CREATE SCHEMA \\\"Library Management\\\";]\\n(Background on this error at: https://sqlalche.me/e/14/f405)\"`.\r\n   - This is expected.\r\n1. The modal remains open. Uncheck the \"Library Management\" schema, and try clicking on 'Add Connection' again.\r\n1. Another error is thrown:\r\n   - `duplicate key value violates unique constraint \"mathesar_database_name_key\" DETAIL: Key (name)=(pavish_install_test_1) already exists.`\r\n   - From the user perpective, there's no clear next step for them to take when this happens.\r\n1. Close the modal and refresh the page, notice that the connection has been created already.\r\n1. Open the connection, and notice that it shows `Schemas (0)`.\r\n1. Forcing a manual reflection using the \"Sync external changes\" button shows the public schema and existing tables.\r\n\r\n### Note:\r\n* I performed the above steps 1-3 only as a means to force an error in the schema creation flow. While testing, I encountered an error which occurred due to an unknown reason. \r\n* This issue primarily focuses on steps 4-10.\r\n\r\n## Expected behavior\r\n* When an error occurs, there should not be an inconsistent middle state in the DB vs the UI.\r\n* When schema creation fails, we should try to revert back to the original state, or we should make it extra clear on the UI that it's a multi-step process.\n", "before_files": [{"content": "\"\"\"Utilities to help with creating and managing connections in Mathesar.\"\"\"\nfrom psycopg2.errors import DuplicateSchema\nfrom sqlalchemy.exc import OperationalError\nfrom mathesar.models.base import Database\nfrom db import install, connection as dbconn\nfrom mathesar.state import reset_reflection\nfrom demo.install.library_dataset import load_library_dataset\nfrom demo.install.movies_dataset import load_movies_dataset\n\n\nclass BadInstallationTarget(Exception):\n    \"\"\"Raise when an attempt is made to install on a disallowed target\"\"\"\n    pass\n\n\ndef copy_connection_from_preexisting(\n        connection, nickname, db_name, create_db, sample_data\n):\n    if connection['connection_type'] == 'internal_database':\n        db_model = Database.create_from_settings_key('default')\n    elif connection['connection_type'] == 'user_database':\n        db_model = Database.current_objects.get(id=connection['id'])\n        db_model.id = None\n    else:\n        raise KeyError(\"connection_type\")\n    root_db = db_model.db_name\n    return _save_and_install(\n        db_model, db_name, root_db, nickname, create_db, sample_data\n    )\n\n\ndef create_connection_from_scratch(\n        user, password, host, port, nickname, db_name, sample_data\n):\n    db_model = Database(username=user, password=password, host=host, port=port)\n    root_db = db_name\n    return _save_and_install(\n        db_model, db_name, root_db, nickname, False, sample_data\n    )\n\n\ndef create_connection_with_new_user(\n        connection, user, password, nickname, db_name, create_db, sample_data\n):\n    db_model = copy_connection_from_preexisting(\n        connection, nickname, db_name, create_db, []\n    )\n    engine = db_model._sa_engine\n    db_model.username = user\n    db_model.password = password\n    db_model.save()\n    dbconn.execute_msar_func_with_engine(\n        engine,\n        'create_basic_mathesar_user',\n        db_model.username,\n        db_model.password\n    )\n    _load_sample_data(db_model._sa_engine, sample_data)\n    return db_model\n\n\ndef _save_and_install(\n        db_model, db_name, root_db, nickname, create_db, sample_data\n):\n    db_model.name = nickname\n    db_model.db_name = db_name\n    _validate_db_model(db_model)\n    db_model.save()\n    try:\n        install.install_mathesar(\n            database_name=db_model.db_name,\n            username=db_model.username,\n            password=db_model.password,\n            hostname=db_model.host,\n            port=db_model.port,\n            skip_confirm=True,\n            create_db=create_db,\n            root_db=root_db,\n        )\n    except OperationalError as e:\n        db_model.delete()\n        raise e\n    _load_sample_data(db_model._sa_engine, sample_data)\n    return db_model\n\n\ndef _load_sample_data(engine, sample_data):\n    DATASET_MAP = {\n        'library_management': load_library_dataset,\n        'movie_collection': load_movies_dataset,\n    }\n    for key in sample_data:\n        try:\n            DATASET_MAP[key](engine, safe_mode=True)\n        except DuplicateSchema:\n            # We swallow this error, since otherwise we'll raise an error on the\n            # front end even though installation generally succeeded.\n            continue\n    reset_reflection()\n\n\ndef _validate_db_model(db_model):\n    internal_db_model = Database.create_from_settings_key('default')\n    if (\n            internal_db_model is not None\n            and db_model.host == internal_db_model.host\n            and db_model.port == internal_db_model.port\n            and db_model.db_name == internal_db_model.db_name\n    ):\n        raise BadInstallationTarget(\n            \"Mathesar can't be installed in the internal DB namespace\"\n        )\n", "path": "mathesar/utils/connections.py"}], "after_files": [{"content": "\"\"\"Utilities to help with creating and managing connections in Mathesar.\"\"\"\nfrom psycopg2.errors import DuplicateSchema\nfrom sqlalchemy.exc import OperationalError, ProgrammingError\nfrom mathesar.models.base import Database\nfrom db import install, connection as dbconn\nfrom mathesar.state import reset_reflection\nfrom demo.install.library_dataset import load_library_dataset\nfrom demo.install.movies_dataset import load_movies_dataset\n\n\nclass BadInstallationTarget(Exception):\n    \"\"\"Raise when an attempt is made to install on a disallowed target\"\"\"\n    pass\n\n\ndef copy_connection_from_preexisting(\n        connection, nickname, db_name, create_db, sample_data\n):\n    if connection['connection_type'] == 'internal_database':\n        db_model = Database.create_from_settings_key('default')\n    elif connection['connection_type'] == 'user_database':\n        db_model = Database.current_objects.get(id=connection['id'])\n        db_model.id = None\n    else:\n        raise KeyError(\"connection_type\")\n    root_db = db_model.db_name\n    return _save_and_install(\n        db_model, db_name, root_db, nickname, create_db, sample_data\n    )\n\n\ndef create_connection_from_scratch(\n        user, password, host, port, nickname, db_name, sample_data\n):\n    db_model = Database(username=user, password=password, host=host, port=port)\n    root_db = db_name\n    return _save_and_install(\n        db_model, db_name, root_db, nickname, False, sample_data\n    )\n\n\ndef create_connection_with_new_user(\n        connection, user, password, nickname, db_name, create_db, sample_data\n):\n    db_model = copy_connection_from_preexisting(\n        connection, nickname, db_name, create_db, []\n    )\n    engine = db_model._sa_engine\n    db_model.username = user\n    db_model.password = password\n    db_model.save()\n    dbconn.execute_msar_func_with_engine(\n        engine,\n        'create_basic_mathesar_user',\n        db_model.username,\n        db_model.password\n    )\n    _load_sample_data(db_model._sa_engine, sample_data)\n    return db_model\n\n\ndef _save_and_install(\n        db_model, db_name, root_db, nickname, create_db, sample_data\n):\n    db_model.name = nickname\n    db_model.db_name = db_name\n    _validate_db_model(db_model)\n    db_model.save()\n    try:\n        install.install_mathesar(\n            database_name=db_model.db_name,\n            username=db_model.username,\n            password=db_model.password,\n            hostname=db_model.host,\n            port=db_model.port,\n            skip_confirm=True,\n            create_db=create_db,\n            root_db=root_db,\n        )\n    except OperationalError as e:\n        db_model.delete()\n        raise e\n    _load_sample_data(db_model._sa_engine, sample_data)\n    return db_model\n\n\ndef _load_sample_data(engine, sample_data):\n    DATASET_MAP = {\n        'library_management': load_library_dataset,\n        'movie_collection': load_movies_dataset,\n    }\n    for key in sample_data:\n        try:\n            DATASET_MAP[key](engine, safe_mode=True)\n        except ProgrammingError as e:\n            if isinstance(e.orig, DuplicateSchema):\n                # We swallow this error, since otherwise we'll raise an error on the\n                # front end even though installation generally succeeded.\n                continue\n    reset_reflection()\n\n\ndef _validate_db_model(db_model):\n    internal_db_model = Database.create_from_settings_key('default')\n    if (\n            internal_db_model is not None\n            and db_model.host == internal_db_model.host\n            and db_model.port == internal_db_model.port\n            and db_model.db_name == internal_db_model.db_name\n    ):\n        raise BadInstallationTarget(\n            \"Mathesar can't be installed in the internal DB namespace\"\n        )\n", "path": "mathesar/utils/connections.py"}]}