We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Duplicate schema creation
**Describe the bug**
We are currently able to create a new schema with an existing schema name, creating duplicates on our mathesar_schema table.

**Expected behavior**
* Schema name should be unique per db in mathesar_schema table.
* If a new schema creation is attempted with the same name as an existing schema, a 400 should be thrown with proper error message.

--- 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/views/api.py`
Content:
```
1 import logging
2 from rest_framework import status, viewsets
3 from rest_framework.exceptions import NotFound, ValidationError
4 from rest_framework.mixins import ListModelMixin, RetrieveModelMixin, CreateModelMixin
5 from rest_framework.response import Response
6 from django.core.cache import cache
7 from django_filters import rest_framework as filters
8 
9 
10 from mathesar.database.utils import get_non_default_database_keys
11 from mathesar.models import Table, Schema, DataFile
12 from mathesar.pagination import DefaultLimitOffsetPagination, TableLimitOffsetPagination
13 from mathesar.serializers import TableSerializer, SchemaSerializer, RecordSerializer, DataFileSerializer
14 from mathesar.utils.schemas import create_schema_and_object, reflect_schemas_from_database
15 from mathesar.utils.tables import reflect_tables_from_schema
16 from mathesar.utils.api import create_table_from_datafile, create_datafile
17 from mathesar.filters import SchemaFilter, TableFilter
18 
19 logger = logging.getLogger(__name__)
20 
21 DB_REFLECTION_KEY = 'database_reflected_recently'
22 DB_REFLECTION_INTERVAL = 60 * 5  # we reflect DB changes every 5 minutes
23 
24 
25 def reflect_db_objects():
26     if not cache.get(DB_REFLECTION_KEY):
27         for database_key in get_non_default_database_keys():
28             reflect_schemas_from_database(database_key)
29         for schema in Schema.objects.all():
30             reflect_tables_from_schema(schema)
31         cache.set(DB_REFLECTION_KEY, True, DB_REFLECTION_INTERVAL)
32 
33 
34 class SchemaViewSet(viewsets.GenericViewSet, ListModelMixin, RetrieveModelMixin):
35     def get_queryset(self):
36         reflect_db_objects()
37         return Schema.objects.all().order_by('-created_at')
38 
39     serializer_class = SchemaSerializer
40     pagination_class = DefaultLimitOffsetPagination
41     filter_backends = (filters.DjangoFilterBackend,)
42     filterset_class = SchemaFilter
43 
44     def create(self, request):
45         schema = create_schema_and_object(request.data['name'], request.data['database'])
46         serializer = SchemaSerializer(schema)
47         return Response(serializer.data, status=status.HTTP_201_CREATED)
48 
49 
50 class TableViewSet(viewsets.GenericViewSet, ListModelMixin, RetrieveModelMixin,
51                    CreateModelMixin):
52     def get_queryset(self):
53         reflect_db_objects()
54         return Table.objects.all().order_by('-created_at')
55 
56     serializer_class = TableSerializer
57     pagination_class = DefaultLimitOffsetPagination
58     filter_backends = (filters.DjangoFilterBackend,)
59     filterset_class = TableFilter
60 
61     def create(self, request):
62         serializer = TableSerializer(data=request.data, context={'request': request})
63         if serializer.is_valid():
64             return create_table_from_datafile(request, serializer.validated_data)
65         else:
66             raise ValidationError(serializer.errors)
67 
68 
69 class RecordViewSet(viewsets.ViewSet):
70     # There is no "update" method.
71     # We're not supporting PUT requests because there aren't a lot of use cases
72     # where the entire record needs to be replaced, PATCH suffices for updates.
73     queryset = Table.objects.all().order_by('-created_at')
74 
75     def list(self, request, table_pk=None):
76         paginator = TableLimitOffsetPagination()
77         records = paginator.paginate_queryset(self.queryset, request, table_pk)
78         serializer = RecordSerializer(records, many=True)
79         return paginator.get_paginated_response(serializer.data)
80 
81     def retrieve(self, request, pk=None, table_pk=None):
82         table = Table.objects.get(id=table_pk)
83         record = table.get_record(pk)
84         if not record:
85             raise NotFound
86         serializer = RecordSerializer(record)
87         return Response(serializer.data)
88 
89     def create(self, request, table_pk=None):
90         table = Table.objects.get(id=table_pk)
91         # We only support adding a single record through the API.
92         assert isinstance((request.data), dict)
93         record = table.create_record_or_records(request.data)
94         serializer = RecordSerializer(record)
95         return Response(serializer.data, status=status.HTTP_201_CREATED)
96 
97     def partial_update(self, request, pk=None, table_pk=None):
98         table = Table.objects.get(id=table_pk)
99         record = table.update_record(pk, request.data)
100         serializer = RecordSerializer(record)
101         return Response(serializer.data)
102 
103     def destroy(self, request, pk=None, table_pk=None):
104         table = Table.objects.get(id=table_pk)
105         table.delete_record(pk)
106         return Response(status=status.HTTP_204_NO_CONTENT)
107 
108 
109 class DatabaseKeyViewSet(viewsets.ViewSet):
110     def list(self, request):
111         return Response(get_non_default_database_keys())
112 
113 
114 class DataFileViewSet(viewsets.GenericViewSet, ListModelMixin, RetrieveModelMixin, CreateModelMixin):
115     queryset = DataFile.objects.all().order_by('-created_at')
116     serializer_class = DataFileSerializer
117     pagination_class = DefaultLimitOffsetPagination
118 
119     def create(self, request):
120         serializer = DataFileSerializer(data=request.data, context={'request': request})
121         if serializer.is_valid():
122             return create_datafile(request, serializer.validated_data['file'])
123         else:
124             raise ValidationError(serializer.errors)
125
```

Path: `mathesar/utils/schemas.py`
Content:
```
1 from db.schemas import (
2     create_schema, get_schema_oid_from_name, get_mathesar_schemas_with_oids
3 )
4 from mathesar.database.base import create_mathesar_engine
5 from mathesar.models import Schema
6 
7 
8 def create_schema_and_object(name, database):
9     engine = create_mathesar_engine(database)
10     create_schema(name, engine)
11     schema_oid = get_schema_oid_from_name(name, engine)
12     schema = Schema.objects.create(oid=schema_oid, database=database)
13     return schema
14 
15 
16 def reflect_schemas_from_database(database):
17     engine = create_mathesar_engine(database)
18     db_schema_oids = {
19         schema["oid"] for schema in get_mathesar_schemas_with_oids(engine)
20     }
21     schemas = [
22         Schema.objects.get_or_create(oid=oid, database=database)
23         for oid in db_schema_oids
24     ]
25     for schema in Schema.objects.all():
26         if schema.oid not in db_schema_oids:
27             schema.delete()
28     return schemas
29
```
--- 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/schemas.py b/mathesar/utils/schemas.py
--- a/mathesar/utils/schemas.py
+++ b/mathesar/utils/schemas.py
@@ -1,5 +1,8 @@
+from rest_framework.exceptions import ValidationError
+
 from db.schemas import (
-    create_schema, get_schema_oid_from_name, get_mathesar_schemas_with_oids
+    create_schema, get_schema_oid_from_name, get_mathesar_schemas,
+    get_mathesar_schemas_with_oids
 )
 from mathesar.database.base import create_mathesar_engine
 from mathesar.models import Schema
@@ -7,6 +10,11 @@
 
 def create_schema_and_object(name, database):
     engine = create_mathesar_engine(database)
+
+    all_schemas = get_mathesar_schemas(engine)
+    if name in all_schemas:
+        raise ValidationError({"name": "Schema name is not unique"})
+
     create_schema(name, engine)
     schema_oid = get_schema_oid_from_name(name, engine)
     schema = Schema.objects.create(oid=schema_oid, database=database)
diff --git a/mathesar/views/api.py b/mathesar/views/api.py
--- a/mathesar/views/api.py
+++ b/mathesar/views/api.py
@@ -42,9 +42,13 @@
     filterset_class = SchemaFilter
 
     def create(self, request):
-        schema = create_schema_and_object(request.data['name'], request.data['database'])
-        serializer = SchemaSerializer(schema)
-        return Response(serializer.data, status=status.HTTP_201_CREATED)
+        serializer = SchemaSerializer(data=request.data)
+        if serializer.is_valid():
+            schema = create_schema_and_object(request.data['name'], request.data['database'])
+            serializer = SchemaSerializer(schema)
+            return Response(serializer.data, status=status.HTTP_201_CREATED)
+        else:
+            raise ValidationError(serializer.errors)
 
 
 class TableViewSet(viewsets.GenericViewSet, ListModelMixin, RetrieveModelMixin,

{"golden_diff": "diff --git a/mathesar/utils/schemas.py b/mathesar/utils/schemas.py\n--- a/mathesar/utils/schemas.py\n+++ b/mathesar/utils/schemas.py\n@@ -1,5 +1,8 @@\n+from rest_framework.exceptions import ValidationError\n+\n from db.schemas import (\n-    create_schema, get_schema_oid_from_name, get_mathesar_schemas_with_oids\n+    create_schema, get_schema_oid_from_name, get_mathesar_schemas,\n+    get_mathesar_schemas_with_oids\n )\n from mathesar.database.base import create_mathesar_engine\n from mathesar.models import Schema\n@@ -7,6 +10,11 @@\n \n def create_schema_and_object(name, database):\n     engine = create_mathesar_engine(database)\n+\n+    all_schemas = get_mathesar_schemas(engine)\n+    if name in all_schemas:\n+        raise ValidationError({\"name\": \"Schema name is not unique\"})\n+\n     create_schema(name, engine)\n     schema_oid = get_schema_oid_from_name(name, engine)\n     schema = Schema.objects.create(oid=schema_oid, database=database)\ndiff --git a/mathesar/views/api.py b/mathesar/views/api.py\n--- a/mathesar/views/api.py\n+++ b/mathesar/views/api.py\n@@ -42,9 +42,13 @@\n     filterset_class = SchemaFilter\n \n     def create(self, request):\n-        schema = create_schema_and_object(request.data['name'], request.data['database'])\n-        serializer = SchemaSerializer(schema)\n-        return Response(serializer.data, status=status.HTTP_201_CREATED)\n+        serializer = SchemaSerializer(data=request.data)\n+        if serializer.is_valid():\n+            schema = create_schema_and_object(request.data['name'], request.data['database'])\n+            serializer = SchemaSerializer(schema)\n+            return Response(serializer.data, status=status.HTTP_201_CREATED)\n+        else:\n+            raise ValidationError(serializer.errors)\n \n \n class TableViewSet(viewsets.GenericViewSet, ListModelMixin, RetrieveModelMixin,\n", "issue": "Duplicate schema creation\n**Describe the bug**\r\nWe are currently able to create a new schema with an existing schema name, creating duplicates on our mathesar_schema table.\r\n\r\n**Expected behavior**\r\n* Schema name should be unique per db in mathesar_schema table.\r\n* If a new schema creation is attempted with the same name as an existing schema, a 400 should be thrown with proper error message.\n", "before_files": [{"content": "import logging\nfrom rest_framework import status, viewsets\nfrom rest_framework.exceptions import NotFound, ValidationError\nfrom rest_framework.mixins import ListModelMixin, RetrieveModelMixin, CreateModelMixin\nfrom rest_framework.response import Response\nfrom django.core.cache import cache\nfrom django_filters import rest_framework as filters\n\n\nfrom mathesar.database.utils import get_non_default_database_keys\nfrom mathesar.models import Table, Schema, DataFile\nfrom mathesar.pagination import DefaultLimitOffsetPagination, TableLimitOffsetPagination\nfrom mathesar.serializers import TableSerializer, SchemaSerializer, RecordSerializer, DataFileSerializer\nfrom mathesar.utils.schemas import create_schema_and_object, reflect_schemas_from_database\nfrom mathesar.utils.tables import reflect_tables_from_schema\nfrom mathesar.utils.api import create_table_from_datafile, create_datafile\nfrom mathesar.filters import SchemaFilter, TableFilter\n\nlogger = logging.getLogger(__name__)\n\nDB_REFLECTION_KEY = 'database_reflected_recently'\nDB_REFLECTION_INTERVAL = 60 * 5  # we reflect DB changes every 5 minutes\n\n\ndef reflect_db_objects():\n    if not cache.get(DB_REFLECTION_KEY):\n        for database_key in get_non_default_database_keys():\n            reflect_schemas_from_database(database_key)\n        for schema in Schema.objects.all():\n            reflect_tables_from_schema(schema)\n        cache.set(DB_REFLECTION_KEY, True, DB_REFLECTION_INTERVAL)\n\n\nclass SchemaViewSet(viewsets.GenericViewSet, ListModelMixin, RetrieveModelMixin):\n    def get_queryset(self):\n        reflect_db_objects()\n        return Schema.objects.all().order_by('-created_at')\n\n    serializer_class = SchemaSerializer\n    pagination_class = DefaultLimitOffsetPagination\n    filter_backends = (filters.DjangoFilterBackend,)\n    filterset_class = SchemaFilter\n\n    def create(self, request):\n        schema = create_schema_and_object(request.data['name'], request.data['database'])\n        serializer = SchemaSerializer(schema)\n        return Response(serializer.data, status=status.HTTP_201_CREATED)\n\n\nclass TableViewSet(viewsets.GenericViewSet, ListModelMixin, RetrieveModelMixin,\n                   CreateModelMixin):\n    def get_queryset(self):\n        reflect_db_objects()\n        return Table.objects.all().order_by('-created_at')\n\n    serializer_class = TableSerializer\n    pagination_class = DefaultLimitOffsetPagination\n    filter_backends = (filters.DjangoFilterBackend,)\n    filterset_class = TableFilter\n\n    def create(self, request):\n        serializer = TableSerializer(data=request.data, context={'request': request})\n        if serializer.is_valid():\n            return create_table_from_datafile(request, serializer.validated_data)\n        else:\n            raise ValidationError(serializer.errors)\n\n\nclass RecordViewSet(viewsets.ViewSet):\n    # There is no \"update\" method.\n    # We're not supporting PUT requests because there aren't a lot of use cases\n    # where the entire record needs to be replaced, PATCH suffices for updates.\n    queryset = Table.objects.all().order_by('-created_at')\n\n    def list(self, request, table_pk=None):\n        paginator = TableLimitOffsetPagination()\n        records = paginator.paginate_queryset(self.queryset, request, table_pk)\n        serializer = RecordSerializer(records, many=True)\n        return paginator.get_paginated_response(serializer.data)\n\n    def retrieve(self, request, pk=None, table_pk=None):\n        table = Table.objects.get(id=table_pk)\n        record = table.get_record(pk)\n        if not record:\n            raise NotFound\n        serializer = RecordSerializer(record)\n        return Response(serializer.data)\n\n    def create(self, request, table_pk=None):\n        table = Table.objects.get(id=table_pk)\n        # We only support adding a single record through the API.\n        assert isinstance((request.data), dict)\n        record = table.create_record_or_records(request.data)\n        serializer = RecordSerializer(record)\n        return Response(serializer.data, status=status.HTTP_201_CREATED)\n\n    def partial_update(self, request, pk=None, table_pk=None):\n        table = Table.objects.get(id=table_pk)\n        record = table.update_record(pk, request.data)\n        serializer = RecordSerializer(record)\n        return Response(serializer.data)\n\n    def destroy(self, request, pk=None, table_pk=None):\n        table = Table.objects.get(id=table_pk)\n        table.delete_record(pk)\n        return Response(status=status.HTTP_204_NO_CONTENT)\n\n\nclass DatabaseKeyViewSet(viewsets.ViewSet):\n    def list(self, request):\n        return Response(get_non_default_database_keys())\n\n\nclass DataFileViewSet(viewsets.GenericViewSet, ListModelMixin, RetrieveModelMixin, CreateModelMixin):\n    queryset = DataFile.objects.all().order_by('-created_at')\n    serializer_class = DataFileSerializer\n    pagination_class = DefaultLimitOffsetPagination\n\n    def create(self, request):\n        serializer = DataFileSerializer(data=request.data, context={'request': request})\n        if serializer.is_valid():\n            return create_datafile(request, serializer.validated_data['file'])\n        else:\n            raise ValidationError(serializer.errors)\n", "path": "mathesar/views/api.py"}, {"content": "from db.schemas import (\n    create_schema, get_schema_oid_from_name, get_mathesar_schemas_with_oids\n)\nfrom mathesar.database.base import create_mathesar_engine\nfrom mathesar.models import Schema\n\n\ndef create_schema_and_object(name, database):\n    engine = create_mathesar_engine(database)\n    create_schema(name, engine)\n    schema_oid = get_schema_oid_from_name(name, engine)\n    schema = Schema.objects.create(oid=schema_oid, database=database)\n    return schema\n\n\ndef reflect_schemas_from_database(database):\n    engine = create_mathesar_engine(database)\n    db_schema_oids = {\n        schema[\"oid\"] for schema in get_mathesar_schemas_with_oids(engine)\n    }\n    schemas = [\n        Schema.objects.get_or_create(oid=oid, database=database)\n        for oid in db_schema_oids\n    ]\n    for schema in Schema.objects.all():\n        if schema.oid not in db_schema_oids:\n            schema.delete()\n    return schemas\n", "path": "mathesar/utils/schemas.py"}], "after_files": [{"content": "import logging\nfrom rest_framework import status, viewsets\nfrom rest_framework.exceptions import NotFound, ValidationError\nfrom rest_framework.mixins import ListModelMixin, RetrieveModelMixin, CreateModelMixin\nfrom rest_framework.response import Response\nfrom django.core.cache import cache\nfrom django_filters import rest_framework as filters\n\n\nfrom mathesar.database.utils import get_non_default_database_keys\nfrom mathesar.models import Table, Schema, DataFile\nfrom mathesar.pagination import DefaultLimitOffsetPagination, TableLimitOffsetPagination\nfrom mathesar.serializers import TableSerializer, SchemaSerializer, RecordSerializer, DataFileSerializer\nfrom mathesar.utils.schemas import create_schema_and_object, reflect_schemas_from_database\nfrom mathesar.utils.tables import reflect_tables_from_schema\nfrom mathesar.utils.api import create_table_from_datafile, create_datafile\nfrom mathesar.filters import SchemaFilter, TableFilter\n\nlogger = logging.getLogger(__name__)\n\nDB_REFLECTION_KEY = 'database_reflected_recently'\nDB_REFLECTION_INTERVAL = 60 * 5  # we reflect DB changes every 5 minutes\n\n\ndef reflect_db_objects():\n    if not cache.get(DB_REFLECTION_KEY):\n        for database_key in get_non_default_database_keys():\n            reflect_schemas_from_database(database_key)\n        for schema in Schema.objects.all():\n            reflect_tables_from_schema(schema)\n        cache.set(DB_REFLECTION_KEY, True, DB_REFLECTION_INTERVAL)\n\n\nclass SchemaViewSet(viewsets.GenericViewSet, ListModelMixin, RetrieveModelMixin):\n    def get_queryset(self):\n        reflect_db_objects()\n        return Schema.objects.all().order_by('-created_at')\n\n    serializer_class = SchemaSerializer\n    pagination_class = DefaultLimitOffsetPagination\n    filter_backends = (filters.DjangoFilterBackend,)\n    filterset_class = SchemaFilter\n\n    def create(self, request):\n        serializer = SchemaSerializer(data=request.data)\n        if serializer.is_valid():\n            schema = create_schema_and_object(request.data['name'], request.data['database'])\n            serializer = SchemaSerializer(schema)\n            return Response(serializer.data, status=status.HTTP_201_CREATED)\n        else:\n            raise ValidationError(serializer.errors)\n\n\nclass TableViewSet(viewsets.GenericViewSet, ListModelMixin, RetrieveModelMixin,\n                   CreateModelMixin):\n    def get_queryset(self):\n        reflect_db_objects()\n        return Table.objects.all().order_by('-created_at')\n\n    serializer_class = TableSerializer\n    pagination_class = DefaultLimitOffsetPagination\n    filter_backends = (filters.DjangoFilterBackend,)\n    filterset_class = TableFilter\n\n    def create(self, request):\n        serializer = TableSerializer(data=request.data, context={'request': request})\n        if serializer.is_valid():\n            return create_table_from_datafile(request, serializer.validated_data)\n        else:\n            raise ValidationError(serializer.errors)\n\n\nclass RecordViewSet(viewsets.ViewSet):\n    # There is no \"update\" method.\n    # We're not supporting PUT requests because there aren't a lot of use cases\n    # where the entire record needs to be replaced, PATCH suffices for updates.\n    queryset = Table.objects.all().order_by('-created_at')\n\n    def list(self, request, table_pk=None):\n        paginator = TableLimitOffsetPagination()\n        records = paginator.paginate_queryset(self.queryset, request, table_pk)\n        serializer = RecordSerializer(records, many=True)\n        return paginator.get_paginated_response(serializer.data)\n\n    def retrieve(self, request, pk=None, table_pk=None):\n        table = Table.objects.get(id=table_pk)\n        record = table.get_record(pk)\n        if not record:\n            raise NotFound\n        serializer = RecordSerializer(record)\n        return Response(serializer.data)\n\n    def create(self, request, table_pk=None):\n        table = Table.objects.get(id=table_pk)\n        # We only support adding a single record through the API.\n        assert isinstance((request.data), dict)\n        record = table.create_record_or_records(request.data)\n        serializer = RecordSerializer(record)\n        return Response(serializer.data, status=status.HTTP_201_CREATED)\n\n    def partial_update(self, request, pk=None, table_pk=None):\n        table = Table.objects.get(id=table_pk)\n        record = table.update_record(pk, request.data)\n        serializer = RecordSerializer(record)\n        return Response(serializer.data)\n\n    def destroy(self, request, pk=None, table_pk=None):\n        table = Table.objects.get(id=table_pk)\n        table.delete_record(pk)\n        return Response(status=status.HTTP_204_NO_CONTENT)\n\n\nclass DatabaseKeyViewSet(viewsets.ViewSet):\n    def list(self, request):\n        return Response(get_non_default_database_keys())\n\n\nclass DataFileViewSet(viewsets.GenericViewSet, ListModelMixin, RetrieveModelMixin, CreateModelMixin):\n    queryset = DataFile.objects.all().order_by('-created_at')\n    serializer_class = DataFileSerializer\n    pagination_class = DefaultLimitOffsetPagination\n\n    def create(self, request):\n        serializer = DataFileSerializer(data=request.data, context={'request': request})\n        if serializer.is_valid():\n            return create_datafile(request, serializer.validated_data['file'])\n        else:\n            raise ValidationError(serializer.errors)\n", "path": "mathesar/views/api.py"}, {"content": "from rest_framework.exceptions import ValidationError\n\nfrom db.schemas import (\n    create_schema, get_schema_oid_from_name, get_mathesar_schemas,\n    get_mathesar_schemas_with_oids\n)\nfrom mathesar.database.base import create_mathesar_engine\nfrom mathesar.models import Schema\n\n\ndef create_schema_and_object(name, database):\n    engine = create_mathesar_engine(database)\n\n    all_schemas = get_mathesar_schemas(engine)\n    if name in all_schemas:\n        raise ValidationError({\"name\": \"Schema name is not unique\"})\n\n    create_schema(name, engine)\n    schema_oid = get_schema_oid_from_name(name, engine)\n    schema = Schema.objects.create(oid=schema_oid, database=database)\n    return schema\n\n\ndef reflect_schemas_from_database(database):\n    engine = create_mathesar_engine(database)\n    db_schema_oids = {\n        schema[\"oid\"] for schema in get_mathesar_schemas_with_oids(engine)\n    }\n    schemas = [\n        Schema.objects.get_or_create(oid=oid, database=database)\n        for oid in db_schema_oids\n    ]\n    for schema in Schema.objects.all():\n        if schema.oid not in db_schema_oids:\n            schema.delete()\n    return schemas\n", "path": "mathesar/utils/schemas.py"}]}