We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Couldn't create new rows for the tables that use IDENTITY for their id columns.
## Description
<!-- A clear and concise description of what the bug is. -->
We have moved to using a different method for ids generations for our default id column as discussed [here](https://github.com/centerofci/mathesar/pull/3016), however this results in an inablility to create new rows using the ui.

![Screenshot from 2023-08-02 17-23-18](https://github.com/centerofci/mathesar/assets/52523023/7de5498c-a354-4405-8c2f-aa796e4588a0)

For some reason the call to the `records/` api isn't being made. Using the browsable api works as expected.

## Expected behavior
<!-- A clear and concise description of what you expected to happen. -->
We should be able to use the ui to create rows for this kind of tables. 

## To Reproduce
<!-- How can we recreate this bug? Please try to provide a Minimal, Complete, and Verifiable (http://stackoverflow.com/help/mcve) example if code-related. -->
- Create a new table using the psql shell: `SELECT msar.add_mathesar_table(2200, 'test_table', NULL, NULL, NULL);` <-- This will create a new table in your `public` shema.
- Sync external changes.
- Try to add new rows to that table. 



--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `db/columns/operations/select.py`
Content:
```
1 import warnings
2 
3 from sqlalchemy import and_, asc, cast, select, text, exists
4 
5 from db.columns.exceptions import DynamicDefaultWarning
6 from db.connection import execute_msar_func_with_engine
7 from db.tables.operations.select import reflect_table_from_oid
8 from db.utils import execute_statement, get_pg_catalog_table
9 
10 
11 def get_column_attnum_from_names_as_map(table_oid, column_names, engine, metadata, connection_to_use=None):
12     statement = _get_columns_attnum_from_names(table_oid, column_names, engine, metadata=metadata)
13     attnums_tuple = execute_statement(engine, statement, connection_to_use).fetchall()
14     name_attnum_map = {attnum_tuple['attname']: attnum_tuple['attnum'] for attnum_tuple in attnums_tuple}
15     return name_attnum_map
16 
17 
18 def get_columns_attnum_from_names(table_oid, column_names, engine, metadata, connection_to_use=None):
19     """
20     Returns the respective list of attnum of the column names passed.
21      The order is based on the column order in the table and not by the order of the column names argument.
22     """
23     statement = _get_columns_attnum_from_names(table_oid, column_names, engine=engine, metadata=metadata)
24     attnums_tuple = execute_statement(engine, statement, connection_to_use).fetchall()
25     attnums = [attnum_tuple[0] for attnum_tuple in attnums_tuple]
26     return attnums
27 
28 
29 def get_column_attnum_from_name(table_oid, column_name, engine, metadata, connection_to_use=None):
30     statement = _get_columns_attnum_from_names(table_oid, [column_name], engine=engine, metadata=metadata)
31     return execute_statement(engine, statement, connection_to_use).scalar()
32 
33 
34 def _get_columns_attnum_from_names(table_oid, column_names, engine, metadata):
35     pg_attribute = get_pg_catalog_table("pg_attribute", engine=engine, metadata=metadata)
36     sel = select(pg_attribute.c.attnum, pg_attribute.c.attname).where(
37         and_(
38             pg_attribute.c.attrelid == table_oid,
39             pg_attribute.c.attname.in_(column_names)
40         )
41     ).order_by(asc(pg_attribute.c.attnum))
42     return sel
43 
44 
45 def get_column_attnums_from_tables(table_oids, engine, metadata, connection_to_use=None):
46     pg_attribute = get_pg_catalog_table("pg_attribute", engine, metadata=metadata)
47     sel = select(pg_attribute.c.attnum, pg_attribute.c.attrelid.label('table_oid')).where(
48         and_(
49             pg_attribute.c.attrelid.in_(table_oids),
50             # Ignore system columns
51             pg_attribute.c.attnum > 0,
52             # Ignore removed columns
53             pg_attribute.c.attisdropped.is_(False)
54         )
55     )
56     results = execute_statement(engine, sel, connection_to_use).fetchall()
57     return results
58 
59 
60 def get_map_of_attnum_and_table_oid_to_column_name(table_oids, engine, metadata, connection_to_use=None):
61     """
62     Order determined by the column order in the table.
63     """
64     triples_of_col_info = _get_triples_of_column_name_and_attnum_and_table_oid(
65         table_oids, None, engine, metadata, connection_to_use
66     )
67     return {
68         (attnum, table_oid): column_name
69         for column_name, attnum, table_oid
70         in triples_of_col_info
71     }
72 
73 
74 def get_column_names_from_attnums(table_oid, attnums, engine, metadata, connection_to_use=None):
75     return list(get_map_of_attnum_to_column_name(table_oid, attnums, engine, metadata, connection_to_use).values())
76 
77 
78 def get_map_of_attnum_to_column_name(table_oid, attnums, engine, metadata, connection_to_use=None):
79     """
80     Order determined by the column order in the table.
81     """
82     triples_of_col_info = _get_triples_of_column_name_and_attnum_and_table_oid(
83         [table_oid], attnums, engine, metadata, connection_to_use
84     )
85     return {
86         attnum: column_name
87         for column_name, attnum, _
88         in triples_of_col_info
89     }
90 
91 
92 def _get_triples_of_column_name_and_attnum_and_table_oid(
93     table_oids, attnums, engine, metadata, connection_to_use
94 ):
95     statement = _statement_for_triples_of_column_name_and_attnum_and_table_oid(
96         table_oids, attnums, engine, metadata
97     )
98     return execute_statement(engine, statement, connection_to_use).fetchall()
99 
100 
101 def get_column_default(table_oid, attnum, engine, metadata, connection_to_use=None):
102     default_dict = get_column_default_dict(
103         table_oid,
104         attnum,
105         engine,
106         metadata=metadata,
107         connection_to_use=connection_to_use,
108     )
109     if default_dict is not None:
110         return default_dict['value']
111 
112 
113 def get_column_default_dict(table_oid, attnum, engine, metadata, connection_to_use=None):
114     column = get_column_from_oid_and_attnum(
115         table_oid=table_oid,
116         attnum=attnum,
117         engine=engine,
118         metadata=metadata,
119         connection_to_use=connection_to_use,
120     )
121 
122     if column.server_default is None:
123         return
124 
125     is_dynamic = execute_msar_func_with_engine(
126         engine, 'is_default_possibly_dynamic', table_oid, attnum
127     ).fetchone()[0]
128     sql_text = str(column.server_default.arg)
129 
130     if is_dynamic:
131         warnings.warn(
132             "Dynamic column defaults are read only", DynamicDefaultWarning
133         )
134         default_value = sql_text
135     else:
136         # Defaults are often stored as text with SQL casts appended
137         # Ex: "'test default string'::character varying" or "'2020-01-01'::date"
138         # Here, we execute the cast to get the proper python value
139         default_value = execute_statement(
140             engine,
141             select(cast(text(sql_text), column.type)),
142             connection_to_use
143         ).scalar()
144 
145     return {"value": default_value, "is_dynamic": is_dynamic}
146 
147 
148 def determine_whether_column_contains_data(
149         table_oid, column_name, engine, metadata, connection_to_use=None
150 ):
151     """
152     Given a column, return True if it contains data, False otherwise.
153     """
154     sa_table = reflect_table_from_oid(
155         table_oid, engine, metadata=metadata, connection_to_use=connection_to_use,
156     )
157     sel = select(exists(1).where(sa_table.columns[column_name] != None))  # noqa
158     contains_data = execute_statement(engine, sel, connection_to_use).scalar()
159     return contains_data
160 
161 
162 def get_column_from_oid_and_attnum(table_oid, attnum, engine, metadata, connection_to_use=None):
163     sa_table = reflect_table_from_oid(table_oid, engine, metadata=metadata, connection_to_use=connection_to_use)
164     column_name = get_column_name_from_attnum(table_oid, attnum, engine, metadata=metadata, connection_to_use=connection_to_use)
165     sa_column = sa_table.columns[column_name]
166     return sa_column
167 
168 
169 def get_column_name_from_attnum(table_oid, attnum, engine, metadata, connection_to_use=None):
170     statement = _statement_for_triples_of_column_name_and_attnum_and_table_oid(
171         [table_oid], [attnum], engine, metadata=metadata,
172     )
173     column_name = execute_statement(engine, statement, connection_to_use).scalar()
174     return column_name
175 
176 
177 def _statement_for_triples_of_column_name_and_attnum_and_table_oid(
178     table_oids, attnums, engine, metadata
179 ):
180     """
181     Returns (column name, column attnum, column table's oid) tuples for each column that's in the
182     tables specified via `table_oids`, and, when `attnums` is not None, that has an attnum
183     specified in `attnums`.
184 
185     The order is based on the column order in the table and not on the order of the arguments.
186     """
187     pg_attribute = get_pg_catalog_table("pg_attribute", engine, metadata=metadata)
188     sel = select(pg_attribute.c.attname, pg_attribute.c.attnum, pg_attribute.c.attrelid)
189     wasnt_dropped = pg_attribute.c.attisdropped.is_(False)
190     table_oid_matches = pg_attribute.c.attrelid.in_(table_oids)
191     conditions = [wasnt_dropped, table_oid_matches]
192     if attnums is not None:
193         attnum_matches = pg_attribute.c.attnum.in_(attnums)
194         conditions.append(attnum_matches)
195     else:
196         attnum_positive = pg_attribute.c.attnum > 0
197         conditions.append(attnum_positive)
198     sel = sel.where(and_(*conditions))
199     return sel
200
```
--- 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/db/columns/operations/select.py b/db/columns/operations/select.py
--- a/db/columns/operations/select.py
+++ b/db/columns/operations/select.py
@@ -1,6 +1,6 @@
 import warnings
 
-from sqlalchemy import and_, asc, cast, select, text, exists
+from sqlalchemy import and_, asc, cast, select, text, exists, Identity
 
 from db.columns.exceptions import DynamicDefaultWarning
 from db.connection import execute_msar_func_with_engine
@@ -118,14 +118,16 @@
         metadata=metadata,
         connection_to_use=connection_to_use,
     )
+    default = column.server_default
 
-    if column.server_default is None:
+    if default is None:
         return
 
     is_dynamic = execute_msar_func_with_engine(
         engine, 'is_default_possibly_dynamic', table_oid, attnum
     ).fetchone()[0]
-    sql_text = str(column.server_default.arg)
+
+    sql_text = str(default.arg) if not isinstance(default, Identity) else 'identity'
 
     if is_dynamic:
         warnings.warn(

{"golden_diff": "diff --git a/db/columns/operations/select.py b/db/columns/operations/select.py\n--- a/db/columns/operations/select.py\n+++ b/db/columns/operations/select.py\n@@ -1,6 +1,6 @@\n import warnings\n \n-from sqlalchemy import and_, asc, cast, select, text, exists\n+from sqlalchemy import and_, asc, cast, select, text, exists, Identity\n \n from db.columns.exceptions import DynamicDefaultWarning\n from db.connection import execute_msar_func_with_engine\n@@ -118,14 +118,16 @@\n         metadata=metadata,\n         connection_to_use=connection_to_use,\n     )\n+    default = column.server_default\n \n-    if column.server_default is None:\n+    if default is None:\n         return\n \n     is_dynamic = execute_msar_func_with_engine(\n         engine, 'is_default_possibly_dynamic', table_oid, attnum\n     ).fetchone()[0]\n-    sql_text = str(column.server_default.arg)\n+\n+    sql_text = str(default.arg) if not isinstance(default, Identity) else 'identity'\n \n     if is_dynamic:\n         warnings.warn(\n", "issue": "Couldn't create new rows for the tables that use IDENTITY for their id columns.\n## Description\r\n<!-- A clear and concise description of what the bug is. -->\r\nWe have moved to using a different method for ids generations for our default id column as discussed [here](https://github.com/centerofci/mathesar/pull/3016), however this results in an inablility to create new rows using the ui.\r\n\r\n![Screenshot from 2023-08-02 17-23-18](https://github.com/centerofci/mathesar/assets/52523023/7de5498c-a354-4405-8c2f-aa796e4588a0)\r\n\r\nFor some reason the call to the `records/` api isn't being made. Using the browsable api works as expected.\r\n\r\n## Expected behavior\r\n<!-- A clear and concise description of what you expected to happen. -->\r\nWe should be able to use the ui to create rows for this kind of tables. \r\n\r\n## To Reproduce\r\n<!-- How can we recreate this bug? Please try to provide a Minimal, Complete, and Verifiable (http://stackoverflow.com/help/mcve) example if code-related. -->\r\n- Create a new table using the psql shell: `SELECT msar.add_mathesar_table(2200, 'test_table', NULL, NULL, NULL);` <-- This will create a new table in your `public` shema.\r\n- Sync external changes.\r\n- Try to add new rows to that table. \r\n\r\n\n", "before_files": [{"content": "import warnings\n\nfrom sqlalchemy import and_, asc, cast, select, text, exists\n\nfrom db.columns.exceptions import DynamicDefaultWarning\nfrom db.connection import execute_msar_func_with_engine\nfrom db.tables.operations.select import reflect_table_from_oid\nfrom db.utils import execute_statement, get_pg_catalog_table\n\n\ndef get_column_attnum_from_names_as_map(table_oid, column_names, engine, metadata, connection_to_use=None):\n    statement = _get_columns_attnum_from_names(table_oid, column_names, engine, metadata=metadata)\n    attnums_tuple = execute_statement(engine, statement, connection_to_use).fetchall()\n    name_attnum_map = {attnum_tuple['attname']: attnum_tuple['attnum'] for attnum_tuple in attnums_tuple}\n    return name_attnum_map\n\n\ndef get_columns_attnum_from_names(table_oid, column_names, engine, metadata, connection_to_use=None):\n    \"\"\"\n    Returns the respective list of attnum of the column names passed.\n     The order is based on the column order in the table and not by the order of the column names argument.\n    \"\"\"\n    statement = _get_columns_attnum_from_names(table_oid, column_names, engine=engine, metadata=metadata)\n    attnums_tuple = execute_statement(engine, statement, connection_to_use).fetchall()\n    attnums = [attnum_tuple[0] for attnum_tuple in attnums_tuple]\n    return attnums\n\n\ndef get_column_attnum_from_name(table_oid, column_name, engine, metadata, connection_to_use=None):\n    statement = _get_columns_attnum_from_names(table_oid, [column_name], engine=engine, metadata=metadata)\n    return execute_statement(engine, statement, connection_to_use).scalar()\n\n\ndef _get_columns_attnum_from_names(table_oid, column_names, engine, metadata):\n    pg_attribute = get_pg_catalog_table(\"pg_attribute\", engine=engine, metadata=metadata)\n    sel = select(pg_attribute.c.attnum, pg_attribute.c.attname).where(\n        and_(\n            pg_attribute.c.attrelid == table_oid,\n            pg_attribute.c.attname.in_(column_names)\n        )\n    ).order_by(asc(pg_attribute.c.attnum))\n    return sel\n\n\ndef get_column_attnums_from_tables(table_oids, engine, metadata, connection_to_use=None):\n    pg_attribute = get_pg_catalog_table(\"pg_attribute\", engine, metadata=metadata)\n    sel = select(pg_attribute.c.attnum, pg_attribute.c.attrelid.label('table_oid')).where(\n        and_(\n            pg_attribute.c.attrelid.in_(table_oids),\n            # Ignore system columns\n            pg_attribute.c.attnum > 0,\n            # Ignore removed columns\n            pg_attribute.c.attisdropped.is_(False)\n        )\n    )\n    results = execute_statement(engine, sel, connection_to_use).fetchall()\n    return results\n\n\ndef get_map_of_attnum_and_table_oid_to_column_name(table_oids, engine, metadata, connection_to_use=None):\n    \"\"\"\n    Order determined by the column order in the table.\n    \"\"\"\n    triples_of_col_info = _get_triples_of_column_name_and_attnum_and_table_oid(\n        table_oids, None, engine, metadata, connection_to_use\n    )\n    return {\n        (attnum, table_oid): column_name\n        for column_name, attnum, table_oid\n        in triples_of_col_info\n    }\n\n\ndef get_column_names_from_attnums(table_oid, attnums, engine, metadata, connection_to_use=None):\n    return list(get_map_of_attnum_to_column_name(table_oid, attnums, engine, metadata, connection_to_use).values())\n\n\ndef get_map_of_attnum_to_column_name(table_oid, attnums, engine, metadata, connection_to_use=None):\n    \"\"\"\n    Order determined by the column order in the table.\n    \"\"\"\n    triples_of_col_info = _get_triples_of_column_name_and_attnum_and_table_oid(\n        [table_oid], attnums, engine, metadata, connection_to_use\n    )\n    return {\n        attnum: column_name\n        for column_name, attnum, _\n        in triples_of_col_info\n    }\n\n\ndef _get_triples_of_column_name_and_attnum_and_table_oid(\n    table_oids, attnums, engine, metadata, connection_to_use\n):\n    statement = _statement_for_triples_of_column_name_and_attnum_and_table_oid(\n        table_oids, attnums, engine, metadata\n    )\n    return execute_statement(engine, statement, connection_to_use).fetchall()\n\n\ndef get_column_default(table_oid, attnum, engine, metadata, connection_to_use=None):\n    default_dict = get_column_default_dict(\n        table_oid,\n        attnum,\n        engine,\n        metadata=metadata,\n        connection_to_use=connection_to_use,\n    )\n    if default_dict is not None:\n        return default_dict['value']\n\n\ndef get_column_default_dict(table_oid, attnum, engine, metadata, connection_to_use=None):\n    column = get_column_from_oid_and_attnum(\n        table_oid=table_oid,\n        attnum=attnum,\n        engine=engine,\n        metadata=metadata,\n        connection_to_use=connection_to_use,\n    )\n\n    if column.server_default is None:\n        return\n\n    is_dynamic = execute_msar_func_with_engine(\n        engine, 'is_default_possibly_dynamic', table_oid, attnum\n    ).fetchone()[0]\n    sql_text = str(column.server_default.arg)\n\n    if is_dynamic:\n        warnings.warn(\n            \"Dynamic column defaults are read only\", DynamicDefaultWarning\n        )\n        default_value = sql_text\n    else:\n        # Defaults are often stored as text with SQL casts appended\n        # Ex: \"'test default string'::character varying\" or \"'2020-01-01'::date\"\n        # Here, we execute the cast to get the proper python value\n        default_value = execute_statement(\n            engine,\n            select(cast(text(sql_text), column.type)),\n            connection_to_use\n        ).scalar()\n\n    return {\"value\": default_value, \"is_dynamic\": is_dynamic}\n\n\ndef determine_whether_column_contains_data(\n        table_oid, column_name, engine, metadata, connection_to_use=None\n):\n    \"\"\"\n    Given a column, return True if it contains data, False otherwise.\n    \"\"\"\n    sa_table = reflect_table_from_oid(\n        table_oid, engine, metadata=metadata, connection_to_use=connection_to_use,\n    )\n    sel = select(exists(1).where(sa_table.columns[column_name] != None))  # noqa\n    contains_data = execute_statement(engine, sel, connection_to_use).scalar()\n    return contains_data\n\n\ndef get_column_from_oid_and_attnum(table_oid, attnum, engine, metadata, connection_to_use=None):\n    sa_table = reflect_table_from_oid(table_oid, engine, metadata=metadata, connection_to_use=connection_to_use)\n    column_name = get_column_name_from_attnum(table_oid, attnum, engine, metadata=metadata, connection_to_use=connection_to_use)\n    sa_column = sa_table.columns[column_name]\n    return sa_column\n\n\ndef get_column_name_from_attnum(table_oid, attnum, engine, metadata, connection_to_use=None):\n    statement = _statement_for_triples_of_column_name_and_attnum_and_table_oid(\n        [table_oid], [attnum], engine, metadata=metadata,\n    )\n    column_name = execute_statement(engine, statement, connection_to_use).scalar()\n    return column_name\n\n\ndef _statement_for_triples_of_column_name_and_attnum_and_table_oid(\n    table_oids, attnums, engine, metadata\n):\n    \"\"\"\n    Returns (column name, column attnum, column table's oid) tuples for each column that's in the\n    tables specified via `table_oids`, and, when `attnums` is not None, that has an attnum\n    specified in `attnums`.\n\n    The order is based on the column order in the table and not on the order of the arguments.\n    \"\"\"\n    pg_attribute = get_pg_catalog_table(\"pg_attribute\", engine, metadata=metadata)\n    sel = select(pg_attribute.c.attname, pg_attribute.c.attnum, pg_attribute.c.attrelid)\n    wasnt_dropped = pg_attribute.c.attisdropped.is_(False)\n    table_oid_matches = pg_attribute.c.attrelid.in_(table_oids)\n    conditions = [wasnt_dropped, table_oid_matches]\n    if attnums is not None:\n        attnum_matches = pg_attribute.c.attnum.in_(attnums)\n        conditions.append(attnum_matches)\n    else:\n        attnum_positive = pg_attribute.c.attnum > 0\n        conditions.append(attnum_positive)\n    sel = sel.where(and_(*conditions))\n    return sel\n", "path": "db/columns/operations/select.py"}], "after_files": [{"content": "import warnings\n\nfrom sqlalchemy import and_, asc, cast, select, text, exists, Identity\n\nfrom db.columns.exceptions import DynamicDefaultWarning\nfrom db.connection import execute_msar_func_with_engine\nfrom db.tables.operations.select import reflect_table_from_oid\nfrom db.utils import execute_statement, get_pg_catalog_table\n\n\ndef get_column_attnum_from_names_as_map(table_oid, column_names, engine, metadata, connection_to_use=None):\n    statement = _get_columns_attnum_from_names(table_oid, column_names, engine, metadata=metadata)\n    attnums_tuple = execute_statement(engine, statement, connection_to_use).fetchall()\n    name_attnum_map = {attnum_tuple['attname']: attnum_tuple['attnum'] for attnum_tuple in attnums_tuple}\n    return name_attnum_map\n\n\ndef get_columns_attnum_from_names(table_oid, column_names, engine, metadata, connection_to_use=None):\n    \"\"\"\n    Returns the respective list of attnum of the column names passed.\n     The order is based on the column order in the table and not by the order of the column names argument.\n    \"\"\"\n    statement = _get_columns_attnum_from_names(table_oid, column_names, engine=engine, metadata=metadata)\n    attnums_tuple = execute_statement(engine, statement, connection_to_use).fetchall()\n    attnums = [attnum_tuple[0] for attnum_tuple in attnums_tuple]\n    return attnums\n\n\ndef get_column_attnum_from_name(table_oid, column_name, engine, metadata, connection_to_use=None):\n    statement = _get_columns_attnum_from_names(table_oid, [column_name], engine=engine, metadata=metadata)\n    return execute_statement(engine, statement, connection_to_use).scalar()\n\n\ndef _get_columns_attnum_from_names(table_oid, column_names, engine, metadata):\n    pg_attribute = get_pg_catalog_table(\"pg_attribute\", engine=engine, metadata=metadata)\n    sel = select(pg_attribute.c.attnum, pg_attribute.c.attname).where(\n        and_(\n            pg_attribute.c.attrelid == table_oid,\n            pg_attribute.c.attname.in_(column_names)\n        )\n    ).order_by(asc(pg_attribute.c.attnum))\n    return sel\n\n\ndef get_column_attnums_from_tables(table_oids, engine, metadata, connection_to_use=None):\n    pg_attribute = get_pg_catalog_table(\"pg_attribute\", engine, metadata=metadata)\n    sel = select(pg_attribute.c.attnum, pg_attribute.c.attrelid.label('table_oid')).where(\n        and_(\n            pg_attribute.c.attrelid.in_(table_oids),\n            # Ignore system columns\n            pg_attribute.c.attnum > 0,\n            # Ignore removed columns\n            pg_attribute.c.attisdropped.is_(False)\n        )\n    )\n    results = execute_statement(engine, sel, connection_to_use).fetchall()\n    return results\n\n\ndef get_map_of_attnum_and_table_oid_to_column_name(table_oids, engine, metadata, connection_to_use=None):\n    \"\"\"\n    Order determined by the column order in the table.\n    \"\"\"\n    triples_of_col_info = _get_triples_of_column_name_and_attnum_and_table_oid(\n        table_oids, None, engine, metadata, connection_to_use\n    )\n    return {\n        (attnum, table_oid): column_name\n        for column_name, attnum, table_oid\n        in triples_of_col_info\n    }\n\n\ndef get_column_names_from_attnums(table_oid, attnums, engine, metadata, connection_to_use=None):\n    return list(get_map_of_attnum_to_column_name(table_oid, attnums, engine, metadata, connection_to_use).values())\n\n\ndef get_map_of_attnum_to_column_name(table_oid, attnums, engine, metadata, connection_to_use=None):\n    \"\"\"\n    Order determined by the column order in the table.\n    \"\"\"\n    triples_of_col_info = _get_triples_of_column_name_and_attnum_and_table_oid(\n        [table_oid], attnums, engine, metadata, connection_to_use\n    )\n    return {\n        attnum: column_name\n        for column_name, attnum, _\n        in triples_of_col_info\n    }\n\n\ndef _get_triples_of_column_name_and_attnum_and_table_oid(\n    table_oids, attnums, engine, metadata, connection_to_use\n):\n    statement = _statement_for_triples_of_column_name_and_attnum_and_table_oid(\n        table_oids, attnums, engine, metadata\n    )\n    return execute_statement(engine, statement, connection_to_use).fetchall()\n\n\ndef get_column_default(table_oid, attnum, engine, metadata, connection_to_use=None):\n    default_dict = get_column_default_dict(\n        table_oid,\n        attnum,\n        engine,\n        metadata=metadata,\n        connection_to_use=connection_to_use,\n    )\n    if default_dict is not None:\n        return default_dict['value']\n\n\ndef get_column_default_dict(table_oid, attnum, engine, metadata, connection_to_use=None):\n    column = get_column_from_oid_and_attnum(\n        table_oid=table_oid,\n        attnum=attnum,\n        engine=engine,\n        metadata=metadata,\n        connection_to_use=connection_to_use,\n    )\n    default = column.server_default\n\n    if default is None:\n        return\n\n    is_dynamic = execute_msar_func_with_engine(\n        engine, 'is_default_possibly_dynamic', table_oid, attnum\n    ).fetchone()[0]\n\n    sql_text = str(default.arg) if not isinstance(default, Identity) else 'identity'\n\n    if is_dynamic:\n        warnings.warn(\n            \"Dynamic column defaults are read only\", DynamicDefaultWarning\n        )\n        default_value = sql_text\n    else:\n        # Defaults are often stored as text with SQL casts appended\n        # Ex: \"'test default string'::character varying\" or \"'2020-01-01'::date\"\n        # Here, we execute the cast to get the proper python value\n        default_value = execute_statement(\n            engine,\n            select(cast(text(sql_text), column.type)),\n            connection_to_use\n        ).scalar()\n\n    return {\"value\": default_value, \"is_dynamic\": is_dynamic}\n\n\ndef determine_whether_column_contains_data(\n        table_oid, column_name, engine, metadata, connection_to_use=None\n):\n    \"\"\"\n    Given a column, return True if it contains data, False otherwise.\n    \"\"\"\n    sa_table = reflect_table_from_oid(\n        table_oid, engine, metadata=metadata, connection_to_use=connection_to_use,\n    )\n    sel = select(exists(1).where(sa_table.columns[column_name] != None))  # noqa\n    contains_data = execute_statement(engine, sel, connection_to_use).scalar()\n    return contains_data\n\n\ndef get_column_from_oid_and_attnum(table_oid, attnum, engine, metadata, connection_to_use=None):\n    sa_table = reflect_table_from_oid(table_oid, engine, metadata=metadata, connection_to_use=connection_to_use)\n    column_name = get_column_name_from_attnum(table_oid, attnum, engine, metadata=metadata, connection_to_use=connection_to_use)\n    sa_column = sa_table.columns[column_name]\n    return sa_column\n\n\ndef get_column_name_from_attnum(table_oid, attnum, engine, metadata, connection_to_use=None):\n    statement = _statement_for_triples_of_column_name_and_attnum_and_table_oid(\n        [table_oid], [attnum], engine, metadata=metadata,\n    )\n    column_name = execute_statement(engine, statement, connection_to_use).scalar()\n    return column_name\n\n\ndef _statement_for_triples_of_column_name_and_attnum_and_table_oid(\n    table_oids, attnums, engine, metadata\n):\n    \"\"\"\n    Returns (column name, column attnum, column table's oid) tuples for each column that's in the\n    tables specified via `table_oids`, and, when `attnums` is not None, that has an attnum\n    specified in `attnums`.\n\n    The order is based on the column order in the table and not on the order of the arguments.\n    \"\"\"\n    pg_attribute = get_pg_catalog_table(\"pg_attribute\", engine, metadata=metadata)\n    sel = select(pg_attribute.c.attname, pg_attribute.c.attnum, pg_attribute.c.attrelid)\n    wasnt_dropped = pg_attribute.c.attisdropped.is_(False)\n    table_oid_matches = pg_attribute.c.attrelid.in_(table_oids)\n    conditions = [wasnt_dropped, table_oid_matches]\n    if attnums is not None:\n        attnum_matches = pg_attribute.c.attnum.in_(attnums)\n        conditions.append(attnum_matches)\n    else:\n        attnum_positive = pg_attribute.c.attnum > 0\n        conditions.append(attnum_positive)\n    sel = sel.where(and_(*conditions))\n    return sel\n", "path": "db/columns/operations/select.py"}]}